feat(odin): Make Actions support opt-out per scheduled/continuous task

cognite/extractorutils/unstable/core/base.py

@@ -124,9 +124,8 @@ def my_task_function(self, task_context: TaskContext) -> None:
 _T = TypeVar("_T", bound=ExtractorConfig)
 
 ACTIONABLE_TASK_TYPES: tuple[type[ScheduledTask], type[ContinuousTask]] = (ScheduledTask, ContinuousTask)
-"""Task types that get auto-generated ``Start``/``Stop`` actions and are tracked in
-``_running_task_tokens``. Kept as a single constant so every ``isinstance`` check that gates this
-behavior stays in sync - add a type here (and only here) to make a new task type actionable."""
+"""Task types that can support Start/Stop actions.
+Whether a given task supports actions is set by its ``actionable`` flag."""
 
 
 class FullConfig(Generic[_T]):
@@ -391,12 +390,7 @@ def add_action(self, action: CustomAction) -> None:
         if any(a.name == action.name for a in self._custom_actions):
             raise ValueError(f"Custom action '{action.name}' is already registered.")
 
-        reserved_names = {
-            name
-            for t in self._tasks
-            if isinstance(t, ACTIONABLE_TASK_TYPES)
-            for name in (f"Start {t.name}", f"Stop {t.name}")
-        }
+        reserved_names = {name for t in self._tasks if t.actionable for name in (f"Start {t.name}", f"Stop {t.name}")}
         if action.name in reserved_names:
             raise ValueError(f"Custom action name '{action.name}' conflicts with an auto-generated task action.")
 
@@ -413,7 +407,7 @@ def _get_startup_request(self) -> StartupRequest:
         available_actions: list[AvailableActionWrite] = []
 
         for t in self._tasks:
-            if isinstance(t, ACTIONABLE_TASK_TYPES):
+            if t.actionable:
                 available_actions.append(
                     AvailableActionWrite(name=f"Start {t.name}", type=ActionType.start_task, task=t.name)
                 )
@@ -433,7 +427,7 @@ def _get_startup_request(self) -> StartupRequest:
             tasks=[
                 DtoTask(
                     type=TaskType.continuous if isinstance(t, ContinuousTask) else TaskType.batch,
-                    action=isinstance(t, ACTIONABLE_TASK_TYPES),
+                    action=t.actionable,
                     description=t.description,
                     name=t.name,
                 )
@@ -495,7 +489,7 @@ def add_task(self, task: Task) -> None:
         if any(t.name == task.name for t in self._tasks):
             raise ValueError(f"Task '{task.name}' is already registered.")
 
-        if isinstance(task, ACTIONABLE_TASK_TYPES):
+        if task.actionable:
             conflict_names = {f"Start {task.name}", f"Stop {task.name}"}
             for action in self._custom_actions:
                 if action.name in conflict_names:
@@ -563,12 +557,13 @@ def _run_task_with_token(
         self, task: ScheduledTask | ContinuousTask, child_token: CancellationToken | None = None
     ) -> None:
         """
-        Create a fresh child cancellation token for an actionable task run.
+        Run a scheduled or continuous task under a fresh child cancellation token.
 
         Stores the token in ``_running_task_tokens`` for external cancellation (e.g. a
         stop_task action), then executes the task. Called by the scheduler (for
         ``ScheduledTask``), the initial boot-time launch (for ``ContinuousTask``), and the
-        start_task action handler (for both).
+        start_task action handler (for both). Tasks that opted out of Actions
+        (``actionable=False``) also run through here; they just never receive Start/Stop actions.
 
         If ``child_token`` is supplied (the caller has already registered it in
         ``_running_task_tokens`` atomically), this method skips the registration step and
@@ -647,7 +642,7 @@ def _dispatch_single_action(self, action: Action) -> None:
                 )
             return
 
-        actionable_tasks = [t for t in self._tasks if isinstance(t, ACTIONABLE_TASK_TYPES)]
+        actionable_tasks = [t for t in self._tasks if t.actionable]
         scheduled_start_names = {f"Start {t.name}" for t in actionable_tasks}
         scheduled_stop_names = {f"Stop {t.name}" for t in actionable_tasks}
         custom_names = {a.name for a in self._custom_actions}
@@ -669,8 +664,9 @@ def _dispatch_single_action(self, action: Action) -> None:
 
     def _handle_start_task_action(self, action: Action) -> None:
         task_name = action.action_name[len("Start ") :]
+        # StartupTask is never actionable; the isinstance check only narrows the type for _run_task_with_token.
         task = next(
-            (t for t in self._tasks if t.name == task_name and isinstance(t, ACTIONABLE_TASK_TYPES)),
+            (t for t in self._tasks if t.name == task_name and t.actionable and isinstance(t, ACTIONABLE_TASK_TYPES)),
             None,
         )
         if task is None:

cognite/extractorutils/unstable/core/tasks.py

@@ -68,10 +68,12 @@ def __init__(
         name: str,
         target: TaskTarget,
         description: str | None = None,
+        actionable: bool = False,
     ) -> None:
         self.name = name
         self.target = target
         self.description = description
+        self.actionable = actionable
 
 
 class ScheduledTask(_Task):
@@ -85,6 +87,7 @@ class ScheduledTask(_Task):
         target: A callable that takes a ``TaskContext`` and performs the task.
         description: An optional description of the task.
         schedule: A ``ScheduleConfig`` object that defines the scheduling configuration for the task.
+        actionable: Whether the task supports Start/Stop actions. Enabled by default; set to ``False`` to opt out.
     """
 
     def __init__(
@@ -94,13 +97,20 @@ def __init__(
         target: TaskTarget,
         description: str | None = None,
         schedule: ScheduleConfig,
+        actionable: bool = True,
     ) -> None:
-        super().__init__(name=name, target=target, description=description)
+        super().__init__(name=name, target=target, description=description, actionable=actionable)
         self.schedule = schedule
 
     @classmethod
     def from_interval(
-        cls, *, interval: str, name: str, target: TaskTarget, description: str | None = None
+        cls,
+        *,
+        interval: str,
+        name: str,
+        target: TaskTarget,
+        description: str | None = None,
+        actionable: bool = True,
     ) -> "ScheduledTask":
         """
         Create a scheduled task that runs at regular intervals.
@@ -110,16 +120,26 @@ def from_interval(
             name: The name of the task.
             target: A callable that takes a ``TaskContext`` and performs the task.
             description: An optional description of the task.
+            actionable: Whether the task supports Start/Stop actions. Enabled by default.
         """
         return ScheduledTask(
             name=name,
             target=target,
             description=description,
+            actionable=actionable,
             schedule=IntervalConfig(type="interval", expression=TimeIntervalConfig(interval)),
         )
 
     @classmethod
-    def from_cron(cls, *, cron: str, name: str, target: TaskTarget, description: str | None = None) -> "ScheduledTask":
+    def from_cron(
+        cls,
+        *,
+        cron: str,
+        name: str,
+        target: TaskTarget,
+        description: str | None = None,
+        actionable: bool = True,
+    ) -> "ScheduledTask":
         """
         Create a scheduled task that runs according to a cron expression.
 
@@ -128,11 +148,13 @@ def from_cron(cls, *, cron: str, name: str, target: TaskTarget, description: str
             name: The name of the task.
             target: A callable that takes a ``TaskContext`` and performs the task.
             description: An optional description of the task.
+            actionable: Whether the task supports Start/Stop actions. Enabled by default.
         """
         return ScheduledTask(
             name=name,
             target=target,
             description=description,
+            actionable=actionable,
             schedule=CronConfig(type="cron", expression=cron),
         )
 
@@ -142,6 +164,12 @@ class ContinuousTask(_Task):
     A task that runs continuously.
 
     Continuous tasks are started when the extractor starts and are expected to run until the extractor stops.
+
+    Args:
+        name: The name of the task.
+        target: A callable that takes a ``TaskContext`` and performs the task.
+        description: An optional description of the task.
+        actionable: Whether the task supports Start/Stop actions. Enabled by default; set to ``False`` to opt out.
     """
 
     def __init__(
@@ -150,8 +178,9 @@ def __init__(
         name: str,
         target: TaskTarget,
         description: str | None = None,
+        actionable: bool = True,
     ) -> None:
-        super().__init__(name=name, target=target, description=description)
+        super().__init__(name=name, target=target, description=description, actionable=actionable)
 
 
 class StartupTask(_Task):

tests/test_unstable/test_action_dispatch.py

@@ -44,6 +44,32 @@ def test_dispatch_unrecognised_action_name_reports_failed() -> None:
     assert "DoesNotExist" in (updates[0].result_message or "")
 
 
+@pytest.mark.parametrize(
+    "make_task",
+    [
+        pytest.param(
+            lambda target: ScheduledTask.from_interval(interval="1h", name="quiet", target=target, actionable=False),
+            id="scheduled",
+        ),
+        pytest.param(lambda target: ContinuousTask(name="quiet", target=target, actionable=False), id="continuous"),
+    ],
+)
+@pytest.mark.parametrize("action_name", ["Start quiet", "Stop quiet"])
+def test_dispatch_action_on_opted_out_task_reports_failed(make_task: Callable, action_name: str) -> None:
+    extractor = _make_extractor()
+    ran = []
+    extractor.add_task(make_task(lambda ctx: ran.append(True)))
+
+    extractor._dispatch_single_action(_make_action("act-1", action_name))
+
+    updates = _queued_updates(extractor)
+    assert len(updates) == 1
+    assert updates[0].status == ActionStatus.failed
+    assert updates[0].result_message == f"No action named '{action_name}' registered"
+    assert not ran
+    assert "quiet" not in extractor._running_task_tokens
+
+
 @pytest.mark.parametrize(
     "register,action_name,expected_status",
     [

tests/test_unstable/test_action_registration.py

@@ -70,6 +70,45 @@ def test_continuous_task_also_produces_start_stop_actions() -> None:
     assert by_name["Stop listener"].type == ActionType.stop_task
 
 
+def test_tasks_are_actionable_by_default() -> None:
+    assert ScheduledTask.from_interval(interval="1h", name="a", target=lambda _: None).actionable
+    assert ScheduledTask.from_cron(cron="0 0 * * *", name="b", target=lambda _: None).actionable
+    assert ContinuousTask(name="c", target=lambda _: None).actionable
+    assert not StartupTask(name="d", target=lambda _: None).actionable
+
+
+def test_opted_out_tasks_produce_no_actions_and_are_flagged_non_actionable() -> None:
+    extractor = _make_extractor()
+    extractor.add_task(
+        ScheduledTask.from_interval(interval="1h", name="sched", target=lambda _: None, actionable=False)
+    )
+    extractor.add_task(ScheduledTask.from_cron(cron="0 0 * * *", name="cron", target=lambda _: None, actionable=False))
+    extractor.add_task(ContinuousTask(name="cont", target=lambda _: None, actionable=False))
+    extractor.add_task(ContinuousTask(name="kept", target=lambda _: None))
+    req = _startup_request(extractor)
+    assert req.available_actions is not None
+    assert {a.name for a in req.available_actions if a.type != ActionType.custom} == {"Start kept", "Stop kept"}
+    assert req.tasks is not None
+    flags = {t.name: t.action for t in req.tasks}
+    assert {n: flags[n] for n in ("sched", "cron", "cont", "kept")} == {
+        "sched": False,
+        "cron": False,
+        "cont": False,
+        "kept": True,
+    }
+
+
+def test_opted_out_task_name_does_not_reserve_action_names() -> None:
+    extractor = _make_extractor()
+    extractor.add_task(ContinuousTask(name="cont", target=lambda _: None, actionable=False))
+    extractor.add_action(CustomAction(name="Start cont", target=lambda _: None))
+    req = _startup_request(extractor)
+    assert req.available_actions is not None
+    by_name = {a.name: a for a in req.available_actions}
+    assert by_name["Start cont"].type == ActionType.custom
+    assert "Stop cont" not in by_name
+
+
 @pytest.mark.parametrize(
     "action_name,expected_type,expected_task",
     [