[Flow] add update_automation_configuration keywords

Author: GuillaumeDSM

packages/commons/octobot_commons/dsl_interpreter/__init__.py

@@ -39,6 +39,9 @@
     ExpressionOperator,
     PreComputingCallOperator,
     ReCallableOperatorMixin,
+    SignalableOperatorMixin,
+    OperatorSignal,
+    OperatorSignals,
     ReCallingOperatorResult,
     ReCallingOperatorResultKeys,
     ProcessBoundOperatorMixin,
@@ -75,6 +78,9 @@
     "ExpressionOperator",
     "PreComputingCallOperator",
     "ReCallableOperatorMixin",
+    "SignalableOperatorMixin",
+    "OperatorSignal",
+    "OperatorSignals",
     "ProcessBoundOperatorMixin",
     "is_process_bound",
     "InterpreterDependency",

packages/commons/octobot_commons/dsl_interpreter/operators/__init__.py

@@ -50,6 +50,11 @@
     ReCallingOperatorResult,
     ReCallingOperatorResultKeys,
 )
+from octobot_commons.dsl_interpreter.operators.signalable_operator_mixin import (
+    OperatorSignal,
+    OperatorSignals,
+    SignalableOperatorMixin,
+)
 from octobot_commons.dsl_interpreter.operators.process_bound_operator_mixin import (
     ProcessBoundOperatorMixin,
     is_process_bound,
@@ -69,6 +74,9 @@
     "ReCallableOperatorMixin",
     "ReCallingOperatorResult",
     "ReCallingOperatorResultKeys",
+    "SignalableOperatorMixin",
+    "OperatorSignal",
+    "OperatorSignals",
     "ProcessBoundOperatorMixin",
     "is_process_bound",
 ]

packages/commons/octobot_commons/dsl_interpreter/operators/process_bound_operator_mixin.py

@@ -14,13 +14,16 @@
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
 
+import asyncio
 import pathlib
 import socket
 import subprocess
+import time
 import typing
 
 import octobot_commons.dsl_interpreter.operator as dsl_interpreter_operator
 import octobot_commons.errors as commons_errors
+import octobot_commons.logging as commons_logging
 import octobot_commons.process_util as process_util
 
 
@@ -52,6 +55,37 @@ def request_graceful_stop(
             )
         return process_util.request_graceful_stop_via_sigterm(self.pid, logger=logger)
 
+    async def wait_until_pid_stopped(
+        self,
+        pid: int,
+        *,
+        logger: typing.Optional[typing.Any] = None,
+        timeout_seconds: float,
+        poll_interval: float = 0.2,
+    ) -> None:
+        """Poll until ``pid`` is gone or ``timeout_seconds`` elapses (after e.g. SIGTERM)."""
+        resolved_logger = logger or commons_logging.get_logger(self.__class__.__name__)
+        if pid <= 0:
+            resolved_logger.info(
+                "wait_until_pid_stopped: pid=%s treated as already stopped (non-positive)",
+                pid,
+            )
+            return
+        resolved_logger.info(
+            "wait_until_pid_stopped: waiting for pid=%s to exit (timeout=%ss)",
+            pid,
+            timeout_seconds,
+        )
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            if not process_util.pid_is_running(pid):
+                resolved_logger.info("wait_until_pid_stopped: pid=%s exited", pid)
+                return
+            await asyncio.sleep(poll_interval)
+        raise commons_errors.DSLInterpreterError(
+            f"Timed out after {timeout_seconds}s waiting for pid={pid} to exit."
+        )
+
     def spawn_subprocess(
         self,
         argv: list[str],

packages/commons/octobot_commons/dsl_interpreter/operators/re_callable_operator_mixin.py

@@ -13,8 +13,6 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-import contextlib
-import contextvars
 import dataclasses
 import enum
 import time
@@ -73,66 +71,19 @@ def get_script_override(result: typing.Any) -> typing.Optional[str]:
         )
 
     @staticmethod
-    def get_keyword(result: typing.Any) -> typing.Optional[str]:
+    def get_keyword(result: dict[str, typing.Any]) -> typing.Optional[str]:
         """
         Returns the keyword from the re-calling operator result.
         """
         return result[ReCallingOperatorResult.__name__]["keyword"]
 
 
-# Per logical context (e.g. asyncio Task), not process-global: concurrent or nested `interprete`
-# calls do not share this value. Values are operator `get_name()` strings for which
-# `set_execution_stop()` is active; `get_execution_stop()` checks membership for one class only.
-_execution_stop_operator_names: contextvars.ContextVar[frozenset[str]] = contextvars.ContextVar(
-    "re_callable_execution_stop_operator_names",
-    default=frozenset(),
-)
-
-
 class ReCallableOperatorMixin:
     """
     Mixin for re-callable operators.
     """
     LAST_EXECUTION_RESULT_KEY = "last_execution_result"
 
-    @classmethod
-    def get_execution_stop(cls) -> bool:
-        """
-        True when this operator class is the current execution_stop target
-        (see set_execution_stop). Scopes per operator get_name() so one class
-        does not see another's stop mode.
-        """
-        return cls.get_name() in _execution_stop_operator_names.get()
-
-    @classmethod
-    @contextlib.contextmanager
-    def set_execution_stop(cls) -> typing.Iterator[None]:
-        """
-        Context manager: for the duration of the block, get_execution_stop() is
-        True for this class only (other re-callable classes remain False unless
-        also entered in an outer or parallel context).
-        """
-        previous = _execution_stop_operator_names.get()
-        token = _execution_stop_operator_names.set(
-            frozenset(previous | {cls.get_name()})
-        )
-        try:
-            yield
-        finally:
-            _execution_stop_operator_names.reset(token)
-
-    @classmethod
-    def should_run_execution_stop_for_result( # pylint: disable=unused-argument
-        cls, re_calling_result: typing.Optional[dict]
-    ) -> bool:
-        """
-        When draining execution_stop for automation shutdown, return whether this
-        operator should run a stop branch for the given previous re-calling
-        result dict (the same shape as last_execution_result on the action).
-        Default: do nothing (subclasses e.g. run_octobot may override).
-        """
-        return False
-
     @classmethod
     def get_re_callable_parameters(cls) -> list[operator_parameter.OperatorParameter]:
         """

packages/commons/octobot_commons/dsl_interpreter/operators/signalable_operator_mixin.py

@@ -0,0 +1,77 @@
+#  Drakkar-Software OctoBot-Commons
+#  Copyright (c) Drakkar-Software, All rights reserved.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 3.0 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library.
+
+import enum
+import typing
+
+
+class OperatorSignal(enum.StrEnum):
+    """
+    Canonical operator signal strings. Call sites pass *.value where a plain str API is expected.
+    """
+    STOP = "STOP"
+    UPDATE_CONFIG = "UPDATE_CONFIG"
+
+
+class OperatorSignals:
+    """
+    Mutable map of DSL operator name to execution signal string for one interpreter/run.
+
+    ``sync`` clears and replaces ``signal_by_operator`` (same pattern as DSLExecutor before each
+    action execution).
+    """
+
+    def __init__(self):
+        self.signal_by_operator: typing.Dict[str, typing.Any] = {}
+
+    def sync(self, signals: typing.Dict[str, typing.Any]) -> None:
+        """
+        Replaces signals mapping with the given signals.
+        """
+        self.signal_by_operator.clear()
+        self.signal_by_operator.update(signals)
+
+
+class SignalableOperatorMixin:
+    """
+    Mixin for operators whose behavior depends on execution signals keyed by operator name.
+
+    Each instance holds an optional ``OperatorSignals`` shared for the DSL run (typically one per
+    interpreter). Callers fill the map via ``OperatorSignals.sync`` (e.g. DSLExecutor before
+    interpretation). ``get_name()`` identifies which map entry applies to ``matches_operator_signal``.
+    """
+
+    def __init__(self, signals: typing.Optional[OperatorSignals] = None):
+        self.signals: typing.Optional[OperatorSignals] = signals
+
+    def matches_operator_signal(self, signal: str) -> bool:
+        """Return whether ``self.signals`` maps this operator's name to ``signal``."""
+        if self.signals is None:
+            return False
+        return self.signals.signal_by_operator.get(self.get_name()) == signal # type: ignore
+
+    @classmethod
+    def should_dispatch_operator_signal_for_result( # pylint: disable=unused-argument
+        cls,
+        signal: str,
+        re_calling_result: typing.Optional[dict],
+    ) -> bool:
+        """
+        When draining dispatcher-driven operator signals for automation shutdown, whether this
+        operator should run its branch for the given previous re-calling payload.
+        Default: do nothing; subclasses override.
+        """
+        return False

packages/commons/octobot_commons/process_util.py

@@ -35,32 +35,65 @@ def spawn_managed_subprocess(
     working_directory: str,
     environment: typing.Optional[typing.Mapping[str, str]] = None,
     hide_console_window: bool = False,
+    forward_terminal_output: bool = False,
 ) -> subprocess.Popen:
     """
     Launch a child process without a shell (``creationflags``: hide console on Windows when asked).
+
+    When ``forward_terminal_output`` is True, the child inherits the parent stdout/stderr (live terminal
+    output). On Windows, ``hide_console_window`` is ignored in that case: ``CREATE_NO_WINDOW`` would
+    detach console output and hide logs even when streams are inherited.
+
+    When ``forward_terminal_output`` is False, stdout and stderr are discarded (``subprocess.DEVNULL``).
     """
     resolved_env = dict(environment) if environment is not None else os.environ.copy()
-    creationflags = subprocess.CREATE_NO_WINDOW if (hide_console_window and sys.platform == "win32") else 0
+    use_hidden_console = (
+        hide_console_window and sys.platform == "win32" and not forward_terminal_output
+    )
+    # subprocess.CREATE_NO_WINDOW exists only on Windows; tests may patch platform on Linux CI.
+    creationflags = (
+        getattr(subprocess, "CREATE_NO_WINDOW", 0) if use_hidden_console else 0
+    )
+    if forward_terminal_output:
+        child_stdout: typing.Optional[int] = None
+        child_stderr: typing.Optional[int] = None
+    else:
+        child_stdout = subprocess.DEVNULL
+        child_stderr = subprocess.DEVNULL
     return subprocess.Popen(
         argv,
         cwd=working_directory,
         env=resolved_env,
         creationflags=creationflags,
+        stdout=child_stdout,
+        stderr=child_stderr,
     )
 
 
-def pid_is_running(pid: int) -> bool:
-    """Best-effort: whether ``pid`` denotes a running OS process."""
+def pid_is_running(pid: int) -> bool: # pylint: disable=too-many-return-statements
+    """Best-effort: whether ``pid`` denotes a running OS process (zombies are treated as not running)."""
     if pid <= 0:
         return False
-    if psutil is not None:
-        try:
-            return psutil.Process(pid).is_running()
-        except psutil.NoSuchProcess:
+    try:
+        proc = psutil.Process(pid)
+    except psutil.NoSuchProcess:
+        return False
+    except psutil.AccessDenied:
+        return True
+    try:
+        if proc.status() == psutil.STATUS_ZOMBIE:
             return False
-        except psutil.AccessDenied:
-            return True
-    return pid > 0
+    except psutil.ZombieProcess:
+        return False
+    except psutil.NoSuchProcess:
+        # PID can disappear between Process() creation and status() (e.g. SIGTERM on Windows).
+        return False
+    try:
+        return proc.is_running()
+    except psutil.ZombieProcess:
+        return False
+    except psutil.NoSuchProcess:
+        return False
 
 
 def request_graceful_stop_via_sigterm(

packages/commons/tests/dsl_interpreter/operators/test_process_bound_operator_mixin.py

@@ -77,6 +77,30 @@ def test_delegates_to_process_util(self):
         stop_mock.assert_called_once_with(99, logger=mock.sentinel.log)
 
 
+@pytest.mark.asyncio
+class TestWaitUntilPidStopped:
+    async def test_non_positive_pid_returns_without_poll(self):
+        bound = process_bound_operator_mixin.ProcessBoundOperatorMixin()
+        with mock.patch.object(process_util, "pid_is_running") as running_mock:
+            await bound.wait_until_pid_stopped(0, timeout_seconds=5.0, logger=mock.Mock())
+        running_mock.assert_not_called()
+
+    async def test_returns_when_pid_not_running(self):
+        bound = process_bound_operator_mixin.ProcessBoundOperatorMixin()
+        with mock.patch.object(process_util, "pid_is_running", return_value=False):
+            await bound.wait_until_pid_stopped(7, timeout_seconds=5.0)
+
+    async def test_timeout_raises_dsl_error(self):
+        bound = process_bound_operator_mixin.ProcessBoundOperatorMixin()
+        with mock.patch.object(process_util, "pid_is_running", return_value=True):
+            with pytest.raises(commons_errors.DSLInterpreterError, match="Timed out"):
+                await bound.wait_until_pid_stopped(
+                    99,
+                    timeout_seconds=0.05,
+                    poll_interval=0.01,
+                )
+
+
 class TestSpawnSubprocess:
     def test_sets_self_pid_from_child_and_returns_popen(self):
         bound = process_bound_operator_mixin.ProcessBoundOperatorMixin()