feat(triggers): add CronTrigger component and triggers category

The trigger is now an in-flow component, not an external entity.
Dragging CronTrigger into a flow declares the schedule; removing it
removes the schedule. The component's class name ("CronTrigger")
is the immutable identifier persisted in node ids — pinned by a unit
test so any accidental rename surfaces as a test failure rather than
breaking every saved flow.

Files (all new):
  src/lfx/src/lfx/components/triggers/cron_trigger.py
      The Component class. Five inputs (cron_expression, timezone
      dropdown with combobox=True for free-text IANA, max_attempts
      with RangeSpec(1..10), payload JSON, worker-fed fire_time)
      and a single Message output. build_event() echoes the
      worker-injected fire_time when present and falls back to the
      current UTC instant for manual canvas runs — mirroring the
      Webhook component's 'inert outside the trigger path' contract.
  src/lfx/src/lfx/components/triggers/constants.py
      DRY landing pad for the curated IANA list, default cron, default
      timezone, default max_attempts, and the upper attempt limit.
      The worker and any future trigger consumer import the same
      constants, so a change here propagates everywhere.
  src/lfx/src/lfx/components/triggers/__init__.py
      Lazy import pattern identical to every other category.

Three single-line additions to src/lfx/src/lfx/components/__init__.py
register the new category in (1) _dynamic_imports, (2) __all__, and
(3) the TYPE_CHECKING block. No other existing file touched.

Tests (new): src/backend/tests/unit/components/test_cron_trigger.py
covers class identity, palette metadata, input names + defaults,
timezone dropdown contents, output shape, manual run path, worker
injection path, status message, and whitespace defensiveness.
10/10 pass; ruff clean.

src/backend/tests/unit/components/test_cron_trigger.py

@@ -0,0 +1,111 @@
+"""Unit tests for the CronTrigger component.
+
+Pure instantiation tests — no graph wiring, no DB. Verifies the
+component declares the immutable identifier, the expected inputs and
+output, and that the two execution paths (manual canvas run vs
+worker-injected fire) both produce a tz-aware ISO 8601 string message.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from lfx.components.triggers import CronTriggerComponent
+from lfx.components.triggers.constants import (
+    COMMON_TIMEZONES,
+    DEFAULT_CRON_EXPRESSION,
+    DEFAULT_MAX_ATTEMPTS,
+    DEFAULT_TIMEZONE,
+)
+from lfx.schema.message import Message
+
+
+def test_class_identity_is_immutable():
+    # ``name`` is persisted in the node id of every saved flow.
+    # Pinning the value here makes any accidental rename a test failure.
+    assert CronTriggerComponent.name == "CronTrigger"
+
+
+def test_metadata_for_palette():
+    assert CronTriggerComponent.display_name == "Cron Trigger"
+    assert CronTriggerComponent.icon == "clock"
+    assert "cron" in CronTriggerComponent.description.lower()
+
+
+def test_inputs_present_with_expected_names():
+    component = CronTriggerComponent()
+    names = {i.name for i in component.inputs}
+    assert names == {"cron_expression", "timezone", "max_attempts", "payload", "fire_time"}
+
+
+def test_inputs_carry_sensible_defaults():
+    component = CronTriggerComponent()
+    by_name = {i.name: i for i in component.inputs}
+    assert by_name["cron_expression"].value == DEFAULT_CRON_EXPRESSION
+    assert by_name["timezone"].value == DEFAULT_TIMEZONE
+    assert by_name["max_attempts"].value == DEFAULT_MAX_ATTEMPTS
+
+
+def test_timezone_dropdown_lists_common_iana_names():
+    component = CronTriggerComponent()
+    by_name = {i.name: i for i in component.inputs}
+    options = by_name["timezone"].options
+    # Combobox is true so other zones are allowed; the curated list
+    # must at least include UTC and a few well-known regions.
+    assert "UTC" in options
+    assert "America/Sao_Paulo" in options
+    assert "Europe/London" in options
+    # The exported tuple should match the dropdown one-to-one.
+    assert set(options) == set(COMMON_TIMEZONES)
+
+
+def test_output_is_single_message_emitter():
+    component = CronTriggerComponent()
+    assert len(component.outputs) == 1
+    only = component.outputs[0]
+    assert only.name == "event"
+    assert only.method == "build_event"
+
+
+def test_manual_canvas_run_returns_current_utc():
+    """No fire_time → component emits the call instant.
+
+    The exact value is unstable across runs, so we assert it parses
+    as ISO 8601 and is recent (within 5 seconds of the test's clock).
+    """
+    component = CronTriggerComponent()
+    before = datetime.now(timezone.utc)
+    message = component.build_event()
+    after = datetime.now(timezone.utc)
+    assert isinstance(message, Message)
+    parsed = datetime.fromisoformat(message.text)
+    assert before <= parsed <= after
+    assert parsed.tzinfo is not None
+
+
+def test_worker_injection_path_emits_provided_timestamp():
+    """Worker-set fire_time → component echoes it verbatim.
+
+    This is the contract relied on by the worker dispatcher: whatever
+    tz-aware ISO string the worker writes into the tweak shows up in
+    the downstream Message unchanged.
+    """
+    component = CronTriggerComponent()
+    component.fire_time = "2026-05-21T12:34:56+00:00"
+    message = component.build_event()
+    assert message.text == "2026-05-21T12:34:56+00:00"
+
+
+def test_status_string_includes_fire_time():
+    component = CronTriggerComponent()
+    component.fire_time = "2026-05-21T12:34:56+00:00"
+    component.build_event()
+    assert "2026-05-21T12:34:56+00:00" in str(component.status)
+
+
+def test_fire_time_whitespace_is_ignored():
+    """Defensive: a worker that injects "  ...  " still produces a clean message."""
+    component = CronTriggerComponent()
+    component.fire_time = "   2026-05-21T12:34:56+00:00   "
+    message = component.build_event()
+    assert message.text == "2026-05-21T12:34:56+00:00"

src/lfx/src/lfx/components/__init__.py

@@ -92,6 +92,7 @@ if TYPE_CHECKING:
         textsplitters,
         toolkits,
         tools,
+        triggers,
         twelvelabs,
         unstructured,
         upstash,
@@ -197,6 +198,7 @@ _dynamic_imports = {
     "textsplitters": "__module__",
     "toolkits": "__module__",
     "tools": "__module__",
+    "triggers": "__module__",
     "twelvelabs": "__module__",
     "unstructured": "__module__",
     "upstash": "__module__",
@@ -328,6 +330,7 @@ __all__ = [
     "textsplitters",
     "toolkits",
     "tools",
+    "triggers",
     "twelvelabs",
     "unstructured",
     "upstash",

src/lfx/src/lfx/components/triggers/__init__.py

@@ -0,0 +1,43 @@
+"""Trigger components.
+
+A trigger component fires the containing flow on some external signal
+(time, queue, event). The ``CronTrigger`` is the time-based one.
+
+Lazy import pattern matches every other category: ``_dynamic_imports``
+maps the class name to its module file and the global ``__getattr__``
+in ``lfx.components.__init__`` does the actual import on first
+attribute access. This keeps server cold-start times low when most
+trigger types are not used.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from lfx.components._importing import import_mod
+
+if TYPE_CHECKING:
+    from .cron_trigger import CronTriggerComponent
+
+_dynamic_imports: dict[str, str] = {
+    "CronTriggerComponent": "cron_trigger",
+}
+
+__all__ = ["CronTriggerComponent"]
+
+
+def __getattr__(attr_name: str) -> Any:
+    if attr_name not in _dynamic_imports:
+        msg = f"module '{__name__}' has no attribute '{attr_name}'"
+        raise AttributeError(msg)
+    try:
+        result = import_mod(attr_name, _dynamic_imports[attr_name], __spec__.parent)
+    except (ModuleNotFoundError, ImportError, AttributeError) as e:
+        msg = f"Could not import '{attr_name}' from '{__name__}': {e}"
+        raise AttributeError(msg) from e
+    globals()[attr_name] = result
+    return result
+
+
+def __dir__() -> list[str]:
+    return list(__all__)

src/lfx/src/lfx/components/triggers/constants.py

@@ -0,0 +1,40 @@
+"""Constants for the triggers component category.
+
+Kept in its own module so the component file stays focused on the
+``Component`` declaration and any downstream consumer (worker,
+discovery helper) can import the same authoritative list.
+"""
+
+from __future__ import annotations
+
+# Curated subset of IANA timezone names exposed in the CronTrigger
+# dropdown. The list is intentionally short — a sensible default that
+# covers the most common regions. Users needing a timezone outside
+# this set can type any IANA name via the combobox mode.
+#
+# The order is roughly geographic (Americas → Europe → Asia →
+# Australia → UTC last) so the dropdown reads naturally; UTC is also
+# the dropdown default and the safe fallback elsewhere in the system.
+COMMON_TIMEZONES: tuple[str, ...] = (
+    "UTC",
+    "America/New_York",
+    "America/Chicago",
+    "America/Denver",
+    "America/Los_Angeles",
+    "America/Sao_Paulo",
+    "America/Argentina/Buenos_Aires",
+    "America/Mexico_City",
+    "Europe/London",
+    "Europe/Berlin",
+    "Europe/Paris",
+    "Europe/Madrid",
+    "Asia/Tokyo",
+    "Asia/Shanghai",
+    "Asia/Kolkata",
+    "Australia/Sydney",
+)
+
+DEFAULT_TIMEZONE: str = "UTC"
+DEFAULT_CRON_EXPRESSION: str = "*/5 * * * *"
+DEFAULT_MAX_ATTEMPTS: int = 3
+MAX_ATTEMPTS_LIMIT: int = 10

src/lfx/src/lfx/components/triggers/cron_trigger.py

@@ -0,0 +1,132 @@
+"""Cron Trigger component.
+
+A trigger that fires the surrounding flow on a recurring cron schedule.
+The component lives in the flow canvas like any other node; its
+configuration (``cron_expression``, ``timezone``) is the source of
+truth and is read by the in-process trigger worker — there is no
+parallel registry table.
+
+Design notes:
+
+* ``name = "CronTrigger"`` is the immutable identifier. It is used
+  both by the backend detection helper (``services.triggers.discovery``)
+  and as the prefix of the node id in ``flow.data`` (e.g.
+  ``"CronTrigger-abc12"``). Renaming would be a breaking change for
+  every saved flow.
+* The ``fire_time`` input is invisible to manual canvas runs. The
+  worker injects it via the same tweak mechanism the webhook handler
+  uses, so the component can pass the actual fire instant downstream
+  without needing a dedicated runtime channel.
+* Manual canvas runs (Play button) are a no-op: the component emits
+  the current UTC time as the event message. That's deliberate —
+  matches the Webhook behaviour, which is also inert outside its HTTP
+  trigger path.
+* No ``is_active`` field: the presence of the node in the flow IS the
+  activation. To pause a trigger, the user removes the node (or the
+  containing flow). Simpler invariant, less to keep in sync.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from lfx.custom.custom_component.component import Component
+from lfx.field_typing.range_spec import RangeSpec
+from lfx.io import (
+    DropdownInput,
+    IntInput,
+    MessageTextInput,
+    MultilineInput,
+    Output,
+)
+from lfx.schema.message import Message
+
+from .constants import (
+    COMMON_TIMEZONES,
+    DEFAULT_CRON_EXPRESSION,
+    DEFAULT_MAX_ATTEMPTS,
+    DEFAULT_TIMEZONE,
+    MAX_ATTEMPTS_LIMIT,
+)
+
+
+class CronTriggerComponent(Component):
+    display_name = "Cron Trigger"
+    description = "Fire this flow on a recurring cron schedule."
+    documentation: str = "https://docs.langflow.org/component-cron-trigger"
+    # The class name pinned by the persisted node id. Do not rename.
+    name = "CronTrigger"
+    icon = "clock"
+
+    inputs = [
+        MessageTextInput(
+            name="cron_expression",
+            display_name="Cron Expression",
+            info=(
+                "Standard 5-field POSIX cron (minute hour day month weekday). "
+                "Example: '*/5 * * * *' fires every five minutes."
+            ),
+            value=DEFAULT_CRON_EXPRESSION,
+            required=True,
+            # No upstream connection — this is a config value, not data.
+            input_types=[],
+        ),
+        DropdownInput(
+            name="timezone",
+            display_name="Timezone",
+            info=(
+                "IANA timezone name used to interpret the cron expression. "
+                "Type any IANA name to use a timezone not in the list."
+            ),
+            options=list(COMMON_TIMEZONES),
+            value=DEFAULT_TIMEZONE,
+            combobox=True,
+        ),
+        IntInput(
+            name="max_attempts",
+            display_name="Max Attempts",
+            info="Retry budget per fire. Failed runs are retried with exponential backoff up to this many times.",
+            value=DEFAULT_MAX_ATTEMPTS,
+            range_spec=RangeSpec(min=1, max=MAX_ATTEMPTS_LIMIT, step=1),
+            advanced=True,
+        ),
+        MultilineInput(
+            name="payload",
+            display_name="Payload (JSON)",
+            info=(
+                "Optional JSON object merged into the SimplifiedAPIRequest fields "
+                "(input_value, input_type, output_type, tweaks, session_id) when the trigger fires."
+            ),
+            advanced=True,
+            input_types=[],
+        ),
+        # Worker-populated. Empty on manual canvas runs.
+        MessageTextInput(
+            name="fire_time",
+            display_name="Fire Time",
+            info="Set by the trigger worker at fire time. Empty on manual runs.",
+            advanced=True,
+            input_types=[],
+        ),
+    ]
+
+    outputs = [
+        Output(
+            display_name="Trigger Event",
+            name="event",
+            method="build_event",
+        ),
+    ]
+
+    def build_event(self) -> Message:
+        """Emit the event message that downstream nodes consume.
+
+        Returns the worker-injected ``fire_time`` when present, otherwise
+        the current UTC instant. Either value is a tz-aware ISO 8601
+        string, so downstream components see a consistent shape
+        regardless of how the flow was kicked off.
+        """
+        fire_time = (self.fire_time or "").strip()
+        text = fire_time or datetime.now(timezone.utc).isoformat()
+        self.status = f"Cron trigger fired at {text}"
+        return Message(text=text)