feat: triggers without automation + flyte.run(trigger) to fire on demand (#1535)

Triggers no longer require an automation. A trigger without one is a
named launch configuration (inputs, env vars, queue, notifications) that
only fires on demand. `flyte.run(trigger, ...)` fires any deployed
trigger from Python, as the trigger, so the run carries its registered
config and is recorded as trigger-fired.

```python
import flyte
import flyte.notify
import flyte.remote
from flyte.models import ActionPhase

env = flyte.TaskEnvironment(name="reports")

# No automation: nothing schedules this, it is fired on demand.
full_report = flyte.Trigger(
    name="full-report",
    inputs={"region": "all", "days": 30},
    env_vars={"REPORT_VERBOSE": "1"},
    notifications=flyte.notify.Email(on_phase=ActionPhase.SUCCEEDED, recipients=("me@example.com",)),
)

@env.task(triggers=(full_report,))
async def report(region: str = "all", days: int = 7) -> str:
    return f"report for {region!r} over {days} day(s)"

# Fire it programmatically, with the trigger's inputs / env vars / notifications.
trigger = flyte.remote.Trigger.get(name="full-report", task_name=report.name)
run = flyte.run(trigger)            # region="all", days=30
run = flyte.run(trigger, days=3)    # override one input, keep the rest
```

Also: `flyte create trigger` no longer requires `--schedule`. Examples
in `examples/triggers/manual.py` and
`examples/triggers/programmatic.py`.

Signed-off-by: Ketan Umare <kumare3@users.noreply.github.com>
Co-authored-by: Ketan Umare <kumare3@users.noreply.github.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

Author: kumare3

examples/triggers/manual.py

@@ -0,0 +1,104 @@
+"""Triggers without automation: a named, pre-bound launch configuration.
+
+A `Trigger` does not have to be scheduled. Leave `automation` unset and the
+trigger becomes a saved launch configuration for the task -- default inputs,
+queue, env vars, notifications -- that nothing fires on its own. It is fired
+on demand only (from the UI, or via the API), which makes it a convenient way
+to publish a handful of "blessed" ways to run a task without re-typing inputs.
+
+Try it (see the env vars below for where notifications are delivered):
+
+    SLACK_WEBHOOK_URL=... NOTIFICATION_EMAIL=... flyte deploy examples/triggers/manual.py env
+    flyte get trigger
+
+`report_on_demand` deploys with two such triggers, `quick-report` and
+`full-report`, alongside a regular scheduled one. Fire either manual trigger
+from the task's Triggers tab in the UI, or from Python with `flyte.run` (see
+`programmatic.py`). The run starts with the trigger's bound inputs, env vars
+and notifications.
+"""
+
+import os
+from datetime import datetime
+
+import flyte
+import flyte.notify
+from flyte.models import ActionPhase
+
+env = flyte.TaskEnvironment(name="manual_trigger_example")
+
+# Where notifications go. Read from the deploying shell so no credentials land
+# in the example:
+#
+#     SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... \
+#     NOTIFICATION_EMAIL=you@example.com \
+#     flyte deploy examples/triggers/manual.py env
+#
+# webhook.site is handy for seeing the Slack payload if you have no webhook yet.
+SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL", "https://webhook.site/")
+REPORT_RECIPIENTS = (os.environ.get("NOTIFICATION_EMAIL", "<EMAIL>"),)
+
+# No `automation=`: nothing schedules these. Each is just a named set of
+# inputs plus what to do when the run ends.
+#
+# Trigger inputs override the task's own defaults (`region="all"`, `days=7`
+# below) for every run fired through the trigger. Inputs the trigger does not
+# mention keep the task default, so `quick-report` still gets `as_of=None`.
+quick_report = flyte.Trigger(
+    name="quick-report",
+    inputs={"region": "us-east", "days": 1},
+    description="Yesterday only, for a fast sanity check",
+    # A quick check only needs to shout when something goes wrong.
+    notifications=flyte.notify.Slack(
+        on_phase=(ActionPhase.FAILED, ActionPhase.TIMED_OUT),
+        webhook_url=SLACK_WEBHOOK,
+        message=":x: quick-report {{.Run.Name}} ended in {{.Phase}}: {{.Error}}",
+    ),
+)
+
+full_report = flyte.Trigger(
+    name="full-report",
+    inputs={"region": "all", "days": 30},
+    description="The full monthly report",
+    env_vars={"REPORT_VERBOSE": "1"},
+    # The monthly report is worth an email on success and a Slack ping on failure.
+    notifications=(
+        flyte.notify.Email(
+            on_phase=ActionPhase.SUCCEEDED,
+            recipients=REPORT_RECIPIENTS,
+            subject="Monthly report {{.Run.Name}} is ready",
+            body="The full report finished.\nRun: {{.Run.Name}}\nProject/Domain: {{.Run.Project}}/{{.Run.Domain}}",
+        ),
+        flyte.notify.Slack(
+            on_phase=ActionPhase.FAILED,
+            webhook_url=SLACK_WEBHOOK,
+            message=":rotating_light: full-report {{.Run.Name}} failed: {{.Error}}",
+        ),
+    ),
+)
+
+# A scheduled trigger can sit next to the manual ones on the same task. Only a
+# schedule can bind `flyte.TriggerTime`, since a manual trigger has no fire time.
+nightly = flyte.Trigger(
+    name="nightly",
+    automation=flyte.Cron("0 2 * * *"),
+    inputs={"as_of": flyte.TriggerTime, "region": "all", "days": 1},
+    notifications=flyte.notify.Slack(
+        on_phase=ActionPhase.FAILED,
+        webhook_url=SLACK_WEBHOOK,
+        message=":rotating_light: nightly report {{.Run.Name}} failed: {{.Error}}",
+    ),
+)
+
+
+@env.task(triggers=(quick_report, full_report, nightly))
+async def report_on_demand(region: str = "all", days: int = 7, as_of: datetime | None = None) -> str:
+    as_of = as_of or datetime.now()
+    msg = f"report for region={region!r} over the last {days} day(s), as of {as_of.isoformat()}"
+    print(msg)
+    return msg
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+    flyte.deploy(env)

examples/triggers/programmatic.py

@@ -0,0 +1,75 @@
+"""Fire a deployed trigger from Python with `flyte.run`.
+
+A trigger is a saved launch configuration for a task: inputs, env vars, queue,
+notifications. Scheduled and artifact triggers fire on their own; a trigger
+with no automation (see `manual.py`) only fires when something asks for it.
+Either kind can be fired on demand by fetching it and passing it to
+`flyte.run`, exactly like a task:
+
+    trigger = flyte.remote.Trigger.get(name="full-report", task_name=...)
+    run = flyte.run(trigger)            # everything the trigger was deployed with
+    run = flyte.run(trigger, days=3)    # override one input, keep the rest
+
+The run is created *as* the trigger: the platform records the trigger as its
+origin (same as a scheduled fire), and the run carries the trigger's inputs,
+env vars, queue and notification rules. Keyword arguments override individual
+inputs; anything left out keeps the value the trigger was deployed with.
+`flyte.with_runcontext(...)` layers further overrides (env vars, queue, run
+name, ...) on top of the trigger's run spec.
+
+Try it, after deploying `manual.py`:
+
+    flyte deploy examples/triggers/manual.py env
+    python examples/triggers/programmatic.py
+"""
+
+import flyte
+import flyte.remote
+
+TASK_NAME = "manual_trigger_example.report_on_demand"
+
+
+def fire_as_deployed() -> flyte.remote.Run:
+    """Fire `full-report` with exactly what it was deployed with (region="all", days=30)."""
+    trigger = flyte.remote.Trigger.get(name="full-report", task_name=TASK_NAME)
+    return flyte.run(trigger)
+
+
+def fire_with_overrides() -> flyte.remote.Run:
+    """Override one input; `region` keeps the trigger's value, `days` becomes 3."""
+    trigger = flyte.remote.Trigger.get(name="full-report", task_name=TASK_NAME)
+    return flyte.run(trigger, days=3)
+
+
+def fire_with_runcontext() -> flyte.remote.Run:
+    """Layer run-level overrides on top of the trigger's run spec.
+
+    The trigger's own env vars and notification rules are kept; `EXTRA_FLAG` is added and
+    the run gets a fixed name. Anything `with_runcontext` sets wins over the trigger's value.
+    """
+    trigger = flyte.remote.Trigger.get(name="quick-report", task_name=TASK_NAME)
+    return flyte.with_runcontext(env_vars={"EXTRA_FLAG": "1"}, name="quick-report-from-python").run(trigger)
+
+
+def fire_every_trigger_on_task() -> list[flyte.remote.Run]:
+    """Triggers from `listall()` can be fired too; their details are fetched on demand.
+
+    Scheduled triggers (`nightly` here) fire just fine off-schedule: the platform stamps the
+    run start time, and any `flyte.TriggerTime` input is filled from it.
+    """
+    runs = []
+    for trigger in flyte.remote.Trigger.listall(task_name=TASK_NAME):
+        runs.append(flyte.run(trigger))
+    return runs
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+
+    run = fire_with_overrides()
+    print(f"fired full-report with days=3: {run.url}")
+    run.wait()
+    print(f"phase={run.phase} inputs={run.inputs()} outputs={run.outputs()}")
+
+    run = fire_with_runcontext()
+    print(f"fired quick-report with run-context overrides: {run.url}")

src/flyte/_internal/runtime/trigger_serde.py

@@ -183,7 +183,13 @@ async def to_task_trigger(
     if kickoff_arg_name is not None:
         context_kvs.append(literals_pb2.KeyValuePair(key=KICKOFF_TIME_INPUT_ARG_CONTEXT_KEY, value=kickoff_arg_name))
 
-    if isinstance(t.automation, OnArtifact):
+    if t.automation is None:
+        # No automation: the trigger is a named launch configuration that is only fired on
+        # demand (UI / API). Nothing schedules it, so there is no kickoff-time or artifact binding.
+        automation_spec = common_pb2.TriggerAutomationSpec(
+            type=common_pb2.TriggerAutomationSpecType.TYPE_NONE,
+        )
+    elif isinstance(t.automation, OnArtifact):
         # Note the contrast with the schedule branch below, which stashes the kickoff-time input
         # arg name under KICKOFF_TIME_INPUT_ARG_CONTEXT_KEY (see convert.py): a scheduled trigger
         # has to, because the offloaded inputs blob is written once and cannot carry the per-fire