fix: accept list of triggers/links in @env.task decorator (#1037)

## Summary

`@env.task(triggers=[trigger, ...])` previously surfaced as
`DeploymentError: ... 'list' object has no attribute 'env_vars'` at
deploy time. The decorator's coercion in `_task_environment.py` only
checked for `tuple`, so a list was wrapped into a 1-tuple containing the
list — and the inner list leaked through to `to_task_trigger`, which
tried to read `.env_vars` off it.

The user-facing docstring at `_trigger.py:724` uses
`triggers=[my_trigger]` (list syntax), so users routinely hit this.

## Changes

- `src/flyte/_task_environment.py:369-370` — coerce triggers by element
type (`isinstance(triggers, Trigger)`) and links by container type
(`Link` is a non-runtime-checkable Protocol, so we check for
list/tuple). Both now correctly handle a single object, a list, or a
tuple.
- `tests/user_api/test_triggers.py` — added
`test_task_with_triggers_as_list` and
`test_task_with_single_element_list_trigger` covering the regression.

## Sentry

Closes the deploy-time failure tracked in
[FLYTE-SDK-17](https://unionai.sentry.io/share/issue/20054cd7e859428193c471b057c9a032/).

## Test plan

- [x] `pytest tests/user_api/test_triggers.py` — 32 passed (30 existing
+ 2 new)
- [x] `pytest tests/user_api/test_task_environment.py
tests/user_api/test_triggers.py tests/flyte/test_trigger.py
tests/flyte/internal/runtime/test_trigger_serde.py` — 93 passed
- [ ] Manual: deploy a task with `triggers=[t1, t2]` against a real
backend (no longer raises)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Haytham Abuelfutuh <haytham@afutuh.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: EngHabu

src/flyte/_task_environment.py

@@ -366,8 +366,8 @@ def decorator(func: F) -> AsyncFunctionTaskTemplate[P, R, F]:
                 queue=queue or self.queue,
                 interruptible=interruptible if interruptible is not None else self.interruptible,
                 entrypoint=entrypoint,
-                triggers=triggers if isinstance(triggers, tuple) else (triggers,),
-                links=links if isinstance(links, tuple) else (links,),
+                triggers=(triggers,) if isinstance(triggers, Trigger) else tuple(triggers),
+                links=tuple(links) if isinstance(links, (list, tuple)) else (links,),
                 task_resolver=task_resolver,
             )
             self._tasks[task_name] = tmpl

tests/user_api/test_triggers.py

@@ -180,6 +180,39 @@ async def task_with_triggers(trigger_time: datetime, x: int = 1) -> str:
     assert len(task_with_triggers.triggers) == 2
 
 
+def test_task_with_triggers_as_list():
+    """Triggers passed as a list must be normalized to a tuple of Trigger objects.
+
+    Regression: a list was being wrapped into a 1-tuple containing the list, which
+    later surfaced as `'list' object has no attribute 'env_vars'` during deploy.
+    """
+    env = flyte.TaskEnvironment(name="test_env")
+
+    trigger1 = flyte.Trigger.hourly(trigger_time_input_key="trigger_time")
+    trigger2 = flyte.Trigger.daily(trigger_time_input_key="trigger_time")
+
+    @env.task(triggers=[trigger1, trigger2])
+    async def task_with_list_triggers(trigger_time: datetime, x: int = 1) -> str:
+        return f"Executed at {trigger_time.isoformat()}"
+
+    assert task_with_list_triggers.triggers == (trigger1, trigger2)
+    assert all(isinstance(t, flyte.Trigger) for t in task_with_list_triggers.triggers)
+
+
+def test_task_with_single_element_list_trigger():
+    """A single-element list of triggers should not be wrapped twice."""
+    env = flyte.TaskEnvironment(name="test_env")
+
+    trigger = flyte.Trigger.hourly(trigger_time_input_key="trigger_time")
+
+    @env.task(triggers=[trigger])
+    async def task_single_list_trigger(trigger_time: datetime, x: int = 1) -> str:
+        return f"Executed at {trigger_time.isoformat()}"
+
+    assert task_single_list_trigger.triggers == (trigger,)
+    assert isinstance(task_single_list_trigger.triggers[0], flyte.Trigger)
+
+
 def test_task_with_trigger_defaults_no_overlap():
     """Test task with defaults that don't overlap with trigger inputs"""
     env = flyte.TaskEnvironment(name="test_env")