diff --git a/krkn/scenario_plugins/triggers/__init__.py b/krkn/scenario_plugins/triggers/__init__.py
new file mode 100644
index 00000000..215c8548
--- /dev/null
+++ b/krkn/scenario_plugins/triggers/__init__.py
@@ -0,0 +1,18 @@
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
+from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
+from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
+
+__all__ = ["AbstractTrigger", "CommandTrigger", "TriggerManager"]
diff --git a/krkn/scenario_plugins/triggers/abstract_trigger.py b/krkn/scenario_plugins/triggers/abstract_trigger.py
new file mode 100644
index 00000000..e30be5b5
--- /dev/null
+++ b/krkn/scenario_plugins/triggers/abstract_trigger.py
@@ -0,0 +1,28 @@
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from abc import ABC, abstractmethod
+
+
+class AbstractTrigger(ABC):
+    """Base class for all trigger types."""
+
+    @abstractmethod
+    def evaluate(self) -> bool:
+        """Returns True if the trigger condition is met."""
+        pass
+
+    @abstractmethod
+    def describe(self) -> str:
+        """Human-readable description for logging."""
+        pass
diff --git a/krkn/scenario_plugins/triggers/command_trigger.py b/krkn/scenario_plugins/triggers/command_trigger.py
new file mode 100644
index 00000000..a663923d
--- /dev/null
+++ b/krkn/scenario_plugins/triggers/command_trigger.py
@@ -0,0 +1,95 @@
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import subprocess
+
+from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
+
+COMMAND_TIMEOUT_SECONDS = 30
+
+
+class CommandTrigger(AbstractTrigger):
+    """Trigger that runs a shell command and checks its exit code."""
+
+    def __init__(self, config: dict):
+        self._cmd = config.get("cmd") or config.get("inline")
+        if not self._cmd:
+            raise ValueError(
+                "command trigger requires either 'cmd' or 'inline' field"
+            )
+        try:
+            self._expected_rc = int(config.get("expected_rc", 0))
+        except (TypeError, ValueError):
+            raise ValueError(
+                f"expected_rc must be an integer, got {config.get('expected_rc')!r}"
+            )
+        if not 0 <= self._expected_rc <= 255:
+            raise ValueError(
+                f"expected_rc must be between 0 and 255, got {self._expected_rc}"
+            )
+        self._last_result: bool | None = None
+
+    def evaluate(self) -> bool:
+        try:
+            result = subprocess.run(
+                self._cmd,
+                shell=True,
+                timeout=COMMAND_TIMEOUT_SECONDS,
+                capture_output=True,
+                text=True,
+            )
+            met = result.returncode == self._expected_rc
+            logging.debug(
+                f"command trigger: rc={result.returncode} "
+                f"expected={self._expected_rc} cmd='{self._cmd}'"
+            )
+            if result.stderr:
+                logging.debug(
+                    f"command trigger stderr: {result.stderr.strip()}"
+                )
+        except subprocess.TimeoutExpired:
+            logging.warning(
+                f"command trigger timed out after {COMMAND_TIMEOUT_SECONDS}s: "
+                f"{self._cmd}"
+            )
+            met = False
+        except FileNotFoundError:
+            logging.error(
+                f"command trigger binary not found: {self._cmd}"
+            )
+            met = False
+        except Exception as e:
+            logging.error(
+                f"command trigger unexpected error: {e}: {self._cmd}"
+            )
+            met = False
+
+        # Log only on state change
+        if met != self._last_result:
+            if met:
+                logging.info(
+                    f"trigger condition satisfied: {self._cmd}"
+                )
+            else:
+                logging.info(
+                    f"trigger condition not satisfied: {self._cmd}"
+                )
+        self._last_result = met
+        return met
+
+    def describe(self) -> str:
+        return (
+            f"command trigger: '{self._cmd}' "
+            f"(expected rc={self._expected_rc})"
+        )
diff --git a/krkn/scenario_plugins/triggers/trigger_manager.py b/krkn/scenario_plugins/triggers/trigger_manager.py
new file mode 100644
index 00000000..5d6e6e88
--- /dev/null
+++ b/krkn/scenario_plugins/triggers/trigger_manager.py
@@ -0,0 +1,169 @@
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import time
+
+from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
+from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
+
+VALID_MODES = {"all_of", "any_of"}
+VALID_ON_TIMEOUT = {"skip", "fail", "run_anyway"}
+
+DEFAULT_TIMEOUT = 300
+DEFAULT_INTERVAL = 5
+DEFAULT_MODE = "all_of"
+DEFAULT_ON_TIMEOUT = "skip"
+
+
+class TriggerManager:
+    """Orchestrates polling across multiple triggers."""
+
+    def __init__(self, trigger_config: dict):
+        conditions = trigger_config.get("conditions")
+        if not conditions:
+            raise ValueError(
+                "trigger config must include a non-empty 'conditions' list"
+            )
+        if not isinstance(conditions, list):
+            raise ValueError(
+                "trigger 'conditions' must be a list, "
+                f"got {type(conditions).__name__}"
+            )
+
+        self._mode = trigger_config.get("mode", DEFAULT_MODE)
+        if self._mode not in VALID_MODES:
+            raise ValueError(
+                f"invalid trigger mode '{self._mode}', "
+                f"must be one of: {', '.join(sorted(VALID_MODES))}"
+            )
+
+        self._on_timeout = trigger_config.get("on_timeout", DEFAULT_ON_TIMEOUT)
+        if self._on_timeout not in VALID_ON_TIMEOUT:
+            raise ValueError(
+                f"invalid on_timeout '{self._on_timeout}', "
+                f"must be one of: {', '.join(sorted(VALID_ON_TIMEOUT))}"
+            )
+
+        self._timeout = trigger_config.get("timeout", DEFAULT_TIMEOUT)
+        self._interval = trigger_config.get("interval", DEFAULT_INTERVAL)
+
+        try:
+            self._timeout = float(self._timeout)
+            self._interval = float(self._interval)
+        except (TypeError, ValueError):
+            raise ValueError(
+                f"timeout and interval must be numeric, "
+                f"got timeout={trigger_config.get('timeout')!r}, "
+                f"interval={trigger_config.get('interval')!r}"
+            )
+
+        if self._timeout <= 0:
+            raise ValueError(
+                f"timeout must be positive, got {self._timeout}"
+            )
+        if self._interval <= 0:
+            raise ValueError(
+                f"interval must be positive, got {self._interval}"
+            )
+
+        self._triggers: list[AbstractTrigger] = []
+        for condition in trigger_config["conditions"]:
+            self._triggers.append(self._build_trigger(condition))
+
+        # Track per-trigger satisfaction state for get_status
+        self._trigger_states: list[bool | None] = [None] * len(self._triggers)
+
+    @property
+    def on_timeout(self) -> str:
+        return self._on_timeout
+
+    @staticmethod
+    def _build_trigger(condition_config: dict) -> AbstractTrigger:
+        """Factory method that creates a trigger from a condition config."""
+        trigger_type = condition_config.get("type")
+        if not trigger_type:
+            raise ValueError("each condition must have a 'type' field")
+
+        if trigger_type == "command":
+            return CommandTrigger(condition_config)
+
+        raise ValueError(f"unknown trigger type: '{trigger_type}'")
+
+    def wait_for_triggers(self) -> bool:
+        """Polls triggers until conditions are met or timeout expires.
+
+        Returns True if conditions were met, False if timed out.
+        """
+        logging.info(
+            f"waiting for triggers: mode={self._mode}, "
+            f"timeout={self._timeout}s, interval={self._interval}s, "
+            f"on_timeout={self._on_timeout}"
+        )
+        deadline = time.monotonic() + self._timeout
+
+        while time.monotonic() < deadline:
+            results = []
+            for i, trigger in enumerate(self._triggers):
+                result = trigger.evaluate()
+                self._trigger_states[i] = result
+                results.append(result)
+
+            logging.debug(
+                f"trigger poll: {[r for r in results]}"
+            )
+
+            if self._mode == "all_of" and all(results):
+                logging.info("all trigger conditions satisfied")
+                return True
+            elif self._mode == "any_of" and any(results):
+                logging.info("at least one trigger condition satisfied")
+                return True
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                break
+            time.sleep(min(self._interval, remaining))
+
+        logging.warning(
+            f"triggers timed out after {self._timeout}s "
+            f"(on_timeout={self._on_timeout})"
+        )
+        return False
+
+    def describe(self) -> str:
+        """Human-readable summary of all triggers."""
+        parts = [
+            f"TriggerManager(mode={self._mode}, "
+            f"timeout={self._timeout}s, interval={self._interval}s, "
+            f"on_timeout={self._on_timeout})",
+        ]
+        for i, trigger in enumerate(self._triggers):
+            parts.append(f"  [{i}] {trigger.describe()}")
+        return "\n".join(parts)
+
+    def get_status(self) -> dict:
+        """Returns current state of each trigger for the signal server."""
+        return {
+            "mode": self._mode,
+            "timeout": self._timeout,
+            "interval": self._interval,
+            "on_timeout": self._on_timeout,
+            "triggers": [
+                {
+                    "description": trigger.describe(),
+                    "satisfied": self._trigger_states[i],
+                }
+                for i, trigger in enumerate(self._triggers)
+            ],
+        }
diff --git a/run_kraken.py b/run_kraken.py
index ad551acc..015dd666 100644
--- a/run_kraken.py
+++ b/run_kraken.py
@@ -66,6 +66,7 @@ from krkn.rollback.command import (
     list_rollback as list_rollback_command,
     execute_rollback as execute_rollback_command,
 )
+from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
 
 # removes TripleDES warning
 import warnings
@@ -412,6 +413,36 @@ def main(options, command: Optional[str]) -> int:
                 logging.error(f"⛔ Class: {class_name} Module: {module_name}")
                 logging.error(f"⚠️ {error}\n")
 
+        # Evaluate top-level triggers before starting health checks or chaos
+        trigger_config = config.get("triggers")
+        if trigger_config:
+            try:
+                trigger_manager = TriggerManager(trigger_config)
+                logging.info(
+                    "waiting for triggers before starting chaos:\n%s",
+                    trigger_manager.describe(),
+                )
+                triggered = trigger_manager.wait_for_triggers()
+                if not triggered:
+                    on_timeout = trigger_manager.on_timeout
+                    if on_timeout == "skip":
+                        logging.warning(
+                            "trigger timed out, skipping all scenarios"
+                        )
+                        chaos_scenarios = []
+                    elif on_timeout == "fail":
+                        logging.error(
+                            "trigger timed out, exiting with failure"
+                        )
+                        return 1
+                    else:
+                        logging.warning(
+                            "trigger timed out, running scenarios anyway"
+                        )
+            except ValueError as e:
+                logging.error("invalid trigger configuration: %s", e)
+                return 1
+
         # Start all health check plugins discovered via config_key_map.
         # Returns list of (plugin, worker_thread, telemetry_queue);
         # worker_thread is None for self-threading plugins (e.g. virt).
diff --git a/tests/test_triggers/__init__.py b/tests/test_triggers/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/test_triggers/test_command_trigger.py b/tests/test_triggers/test_command_trigger.py
new file mode 100644
index 00000000..e82775b7
--- /dev/null
+++ b/tests/test_triggers/test_command_trigger.py
@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+
+"""
+Test suite for CommandTrigger class
+
+Usage:
+    python -m coverage run -a -m unittest tests/test_triggers/test_command_trigger.py -v
+
+Assisted By: Claude Code
+"""
+
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
+
+from krkn.scenario_plugins.triggers.command_trigger import (
+    COMMAND_TIMEOUT_SECONDS,
+    CommandTrigger,
+)
+
+
+class TestCommandTrigger(unittest.TestCase):
+
+    def _make_trigger(self, **overrides):
+        """Build a CommandTrigger with sensible defaults."""
+        config = {"cmd": "echo hello", "expected_rc": 0}
+        config.update(overrides)
+        return CommandTrigger(config)
+
+    # ------------------------------------------------------------------
+    # evaluate() tests
+    # ------------------------------------------------------------------
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_success(self, mock_run):
+        """Command exits 0, expected_rc=0 -> returns True."""
+        mock_run.return_value = MagicMock(returncode=0)
+        trigger = self._make_trigger(cmd="echo ok", expected_rc=0)
+
+        self.assertTrue(trigger.evaluate())
+        mock_run.assert_called_once_with(
+            "echo ok",
+            shell=True,
+            timeout=COMMAND_TIMEOUT_SECONDS,
+            capture_output=True,
+            text=True,
+        )
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_failure(self, mock_run):
+        """Command exits 1, expected_rc=0 -> returns False."""
+        mock_run.return_value = MagicMock(returncode=1)
+        trigger = self._make_trigger(cmd="false", expected_rc=0)
+
+        self.assertFalse(trigger.evaluate())
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_custom_expected_rc(self, mock_run):
+        """Command exits 42, expected_rc=42 -> returns True."""
+        mock_run.return_value = MagicMock(returncode=42)
+        trigger = self._make_trigger(cmd="exit 42", expected_rc=42)
+
+        self.assertTrue(trigger.evaluate())
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_inline_command(self, mock_run):
+        """Uses 'inline' field instead of 'cmd'."""
+        mock_run.return_value = MagicMock(returncode=0)
+        trigger = CommandTrigger({"inline": "date"})
+
+        self.assertTrue(trigger.evaluate())
+        mock_run.assert_called_once_with(
+            "date",
+            shell=True,
+            timeout=COMMAND_TIMEOUT_SECONDS,
+            capture_output=True,
+            text=True,
+        )
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_cmd_takes_precedence_over_inline(self, mock_run):
+        """Both 'cmd' and 'inline' provided; 'cmd' is used."""
+        mock_run.return_value = MagicMock(returncode=0)
+        trigger = CommandTrigger({"cmd": "use-this", "inline": "not-this"})
+
+        trigger.evaluate()
+        args, _ = mock_run.call_args
+        self.assertEqual(args[0], "use-this")
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_timeout(self, mock_run):
+        """Command hangs, subprocess times out -> returns False."""
+        mock_run.side_effect = subprocess.TimeoutExpired(
+            cmd="sleep 999", timeout=COMMAND_TIMEOUT_SECONDS
+        )
+        trigger = self._make_trigger(cmd="sleep 999")
+
+        self.assertFalse(trigger.evaluate())
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_command_not_found(self, mock_run):
+        """Binary does not exist -> returns False."""
+        mock_run.side_effect = FileNotFoundError("No such file")
+        trigger = self._make_trigger(cmd="/nonexistent/binary")
+
+        self.assertFalse(trigger.evaluate())
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_unexpected_exception(self, mock_run):
+        """Unexpected exception (e.g. PermissionError) -> returns False, no crash."""
+        mock_run.side_effect = PermissionError("Permission denied")
+        trigger = self._make_trigger(cmd="/restricted/script.sh")
+
+        self.assertFalse(trigger.evaluate())
+
+    @patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
+    def test_evaluate_default_expected_rc(self, mock_run):
+        """No expected_rc in config -> defaults to 0."""
+        mock_run.return_value = MagicMock(returncode=0)
+        trigger = CommandTrigger({"cmd": "true"})
+
+        self.assertTrue(trigger.evaluate())
+        self.assertEqual(trigger._expected_rc, 0)
+
+    # ------------------------------------------------------------------
+    # describe() tests
+    # ------------------------------------------------------------------
+
+    def test_describe(self):
+        """Returns a meaningful description string."""
+        trigger = self._make_trigger(cmd="check_service.sh", expected_rc=0)
+        description = trigger.describe()
+
+        self.assertIn("check_service.sh", description)
+        self.assertIn("0", description)
+        self.assertIsInstance(description, str)
+        self.assertTrue(len(description) > 0)
+
+    # ------------------------------------------------------------------
+    # Validation tests
+    # ------------------------------------------------------------------
+
+    def test_missing_cmd_and_inline(self):
+        """Neither 'cmd' nor 'inline' provided -> raises ValueError."""
+        with self.assertRaises(ValueError) as ctx:
+            CommandTrigger({})
+
+        self.assertIn("cmd", str(ctx.exception).lower())
+        self.assertIn("inline", str(ctx.exception).lower())
+
+    def test_expected_rc_string_coerced_to_int(self):
+        """expected_rc='0' (string from envsubst) -> coerced to int 0."""
+        trigger = CommandTrigger({"cmd": "true", "expected_rc": "0"})
+        self.assertEqual(trigger._expected_rc, 0)
+        self.assertIsInstance(trigger._expected_rc, int)
+
+    def test_expected_rc_invalid_raises(self):
+        """expected_rc='abc' -> raises ValueError."""
+        with self.assertRaises(ValueError) as ctx:
+            CommandTrigger({"cmd": "true", "expected_rc": "abc"})
+        self.assertIn("expected_rc", str(ctx.exception))
+
+    def test_expected_rc_above_255_raises(self):
+        """expected_rc=999 -> raises ValueError (Unix rc range is 0-255)."""
+        with self.assertRaises(ValueError) as ctx:
+            CommandTrigger({"cmd": "true", "expected_rc": 999})
+        self.assertIn("255", str(ctx.exception))
+
+    def test_expected_rc_negative_raises(self):
+        """expected_rc=-1 -> raises ValueError."""
+        with self.assertRaises(ValueError) as ctx:
+            CommandTrigger({"cmd": "true", "expected_rc": -1})
+        self.assertIn("expected_rc", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_triggers/test_trigger_manager.py b/tests/test_triggers/test_trigger_manager.py
new file mode 100644
index 00000000..446df883
--- /dev/null
+++ b/tests/test_triggers/test_trigger_manager.py
@@ -0,0 +1,370 @@
+#!/usr/bin/env python3
+
+"""
+Test suite for TriggerManager class
+
+Usage:
+    python -m coverage run -a -m unittest tests/test_triggers/test_trigger_manager.py -v
+
+Assisted By: Claude Code
+"""
+
+import unittest
+from unittest.mock import patch
+
+from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
+from krkn.scenario_plugins.triggers.trigger_manager import (
+    DEFAULT_INTERVAL,
+    DEFAULT_MODE,
+    DEFAULT_ON_TIMEOUT,
+    DEFAULT_TIMEOUT,
+    TriggerManager,
+)
+
+
+class StubTrigger(AbstractTrigger):
+    """Concrete trigger for testing that returns a preconfigured value."""
+
+    def __init__(self, result: bool, name: str = "stub"):
+        self._result = result
+        self._name = name
+
+    def evaluate(self) -> bool:
+        return self._result
+
+    def describe(self) -> str:
+        return f"stub trigger '{self._name}' (result={self._result})"
+
+    def set_result(self, result: bool):
+        self._result = result
+
+
+def _make_config(**overrides):
+    """Build a minimal valid trigger config dict with overrides."""
+    config = {
+        "mode": "all_of",
+        "timeout": 10,
+        "interval": 1,
+        "on_timeout": "skip",
+        "conditions": [
+            {"type": "command", "cmd": "true"},
+        ],
+    }
+    config.update(overrides)
+    return config
+
+
+class TestTriggerManager(unittest.TestCase):
+
+    # ------------------------------------------------------------------
+    # wait_for_triggers() — mode tests
+    # ------------------------------------------------------------------
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_all_of_all_pass(self, mock_build):
+        """mode=all_of, all triggers pass -> returns True."""
+        t1 = StubTrigger(True, "t1")
+        t2 = StubTrigger(True, "t2")
+        mock_build.side_effect = [t1, t2]
+
+        config = _make_config(
+            mode="all_of",
+            conditions=[
+                {"type": "command", "cmd": "a"},
+                {"type": "command", "cmd": "b"},
+            ],
+        )
+        manager = TriggerManager(config)
+        self.assertTrue(manager.wait_for_triggers())
+
+    @patch("krkn.scenario_plugins.triggers.trigger_manager.time")
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_all_of_one_fails(self, mock_build, mock_time):
+        """mode=all_of, one trigger stays False -> loops until timeout, returns False."""
+        t1 = StubTrigger(True, "t1")
+        t2 = StubTrigger(False, "t2")
+        mock_build.side_effect = [t1, t2]
+
+        # Simulate: first call returns start time, subsequent calls advance past deadline
+        call_count = 0
+
+        def advancing_monotonic():
+            nonlocal call_count
+            call_count += 1
+            if call_count <= 2:
+                return 0.0
+            return 999.0  # past deadline
+
+        mock_time.monotonic.side_effect = advancing_monotonic
+        mock_time.sleep = lambda x: None
+
+        config = _make_config(
+            mode="all_of",
+            timeout=10,
+            conditions=[
+                {"type": "command", "cmd": "a"},
+                {"type": "command", "cmd": "b"},
+            ],
+        )
+        manager = TriggerManager(config)
+        self.assertFalse(manager.wait_for_triggers())
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_any_of_one_passes(self, mock_build):
+        """mode=any_of, one trigger passes -> returns True immediately."""
+        t1 = StubTrigger(False, "t1")
+        t2 = StubTrigger(True, "t2")
+        mock_build.side_effect = [t1, t2]
+
+        config = _make_config(
+            mode="any_of",
+            conditions=[
+                {"type": "command", "cmd": "a"},
+                {"type": "command", "cmd": "b"},
+            ],
+        )
+        manager = TriggerManager(config)
+        self.assertTrue(manager.wait_for_triggers())
+
+    # ------------------------------------------------------------------
+    # wait_for_triggers() — timeout tests
+    # ------------------------------------------------------------------
+
+    @patch("krkn.scenario_plugins.triggers.trigger_manager.time")
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_timeout_returns_false(self, mock_build, mock_time):
+        """Triggers never pass, timeout expires -> returns False."""
+        t1 = StubTrigger(False, "t1")
+        mock_build.side_effect = [t1]
+
+        call_count = 0
+
+        def advancing_monotonic():
+            nonlocal call_count
+            call_count += 1
+            if call_count <= 2:
+                return 0.0
+            return 999.0
+
+        mock_time.monotonic.side_effect = advancing_monotonic
+        mock_time.sleep = lambda x: None
+
+        config = _make_config(
+            timeout=10,
+            conditions=[{"type": "command", "cmd": "a"}],
+        )
+        manager = TriggerManager(config)
+        self.assertFalse(manager.wait_for_triggers())
+
+    # ------------------------------------------------------------------
+    # Default values
+    # ------------------------------------------------------------------
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_default_timeout(self, mock_build):
+        """No timeout in config -> defaults to 300."""
+        mock_build.return_value = StubTrigger(True)
+        config = {
+            "conditions": [{"type": "command", "cmd": "true"}],
+        }
+        manager = TriggerManager(config)
+        self.assertEqual(manager._timeout, DEFAULT_TIMEOUT)
+        self.assertEqual(manager._timeout, 300)
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_default_interval(self, mock_build):
+        """No interval in config -> defaults to 5."""
+        mock_build.return_value = StubTrigger(True)
+        config = {
+            "conditions": [{"type": "command", "cmd": "true"}],
+        }
+        manager = TriggerManager(config)
+        self.assertEqual(manager._interval, DEFAULT_INTERVAL)
+        self.assertEqual(manager._interval, 5)
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_default_mode(self, mock_build):
+        """No mode in config -> defaults to all_of."""
+        mock_build.return_value = StubTrigger(True)
+        config = {
+            "conditions": [{"type": "command", "cmd": "true"}],
+        }
+        manager = TriggerManager(config)
+        self.assertEqual(manager._mode, DEFAULT_MODE)
+        self.assertEqual(manager._mode, "all_of")
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_default_on_timeout(self, mock_build):
+        """No on_timeout in config -> defaults to skip."""
+        mock_build.return_value = StubTrigger(True)
+        config = {
+            "conditions": [{"type": "command", "cmd": "true"}],
+        }
+        manager = TriggerManager(config)
+        self.assertEqual(manager._on_timeout, DEFAULT_ON_TIMEOUT)
+        self.assertEqual(manager._on_timeout, "skip")
+
+    # ------------------------------------------------------------------
+    # Validation tests
+    # ------------------------------------------------------------------
+
+    def test_invalid_mode_raises(self):
+        """mode='invalid' -> raises ValueError."""
+        config = _make_config(mode="invalid")
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("invalid", str(ctx.exception))
+
+    def test_missing_conditions_raises(self):
+        """No 'conditions' key -> raises ValueError."""
+        config = {"mode": "all_of", "timeout": 10}
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("conditions", str(ctx.exception))
+
+    def test_empty_conditions_raises(self):
+        """conditions=[] -> raises ValueError."""
+        config = _make_config(conditions=[])
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("conditions", str(ctx.exception))
+
+    def test_conditions_not_a_list_raises(self):
+        """conditions='some string' -> raises ValueError."""
+        config = _make_config(conditions="kubectl check something")
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("list", str(ctx.exception))
+
+    def test_on_timeout_property(self):
+        """on_timeout property returns validated value."""
+        mock_build = patch.object(TriggerManager, "_build_trigger").start()
+        mock_build.return_value = StubTrigger(True)
+        config = _make_config(on_timeout="fail")
+        manager = TriggerManager(config)
+        self.assertEqual(manager.on_timeout, "fail")
+        patch.stopall()
+
+    def test_unknown_trigger_type_raises(self):
+        """type='kafka' -> raises ValueError."""
+        config = _make_config(
+            conditions=[{"type": "kafka", "topic": "events"}],
+        )
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("kafka", str(ctx.exception))
+
+    def test_negative_timeout_raises(self):
+        """timeout=-1 -> raises ValueError."""
+        config = _make_config(timeout=-1)
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("timeout", str(ctx.exception))
+
+    def test_zero_timeout_raises(self):
+        """timeout=0 -> raises ValueError."""
+        config = _make_config(timeout=0)
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("timeout", str(ctx.exception))
+
+    def test_negative_interval_raises(self):
+        """interval=-1 -> raises ValueError."""
+        config = _make_config(interval=-1)
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("interval", str(ctx.exception))
+
+    def test_zero_interval_raises(self):
+        """interval=0 -> raises ValueError."""
+        config = _make_config(interval=0)
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("interval", str(ctx.exception))
+
+    def test_string_timeout_raises(self):
+        """timeout='abc' -> raises ValueError."""
+        config = _make_config(timeout="abc")
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("numeric", str(ctx.exception))
+
+    def test_string_interval_raises(self):
+        """interval='xyz' -> raises ValueError."""
+        config = _make_config(interval="xyz")
+        with self.assertRaises(ValueError) as ctx:
+            TriggerManager(config)
+        self.assertIn("numeric", str(ctx.exception))
+
+    # ------------------------------------------------------------------
+    # get_status() tests
+    # ------------------------------------------------------------------
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_get_status(self, mock_build):
+        """get_status returns dict with trigger states."""
+        t1 = StubTrigger(True, "t1")
+        t2 = StubTrigger(False, "t2")
+        mock_build.side_effect = [t1, t2]
+
+        config = _make_config(
+            mode="any_of",
+            timeout=60,
+            interval=2,
+            on_timeout="fail",
+            conditions=[
+                {"type": "command", "cmd": "a"},
+                {"type": "command", "cmd": "b"},
+            ],
+        )
+        manager = TriggerManager(config)
+
+        status = manager.get_status()
+        self.assertEqual(status["mode"], "any_of")
+        self.assertEqual(status["timeout"], 60)
+        self.assertEqual(status["interval"], 2)
+        self.assertEqual(status["on_timeout"], "fail")
+        self.assertEqual(len(status["triggers"]), 2)
+
+        # Before any evaluation, states should be None
+        for trigger_info in status["triggers"]:
+            self.assertIn("description", trigger_info)
+            self.assertIn("satisfied", trigger_info)
+            self.assertIsNone(trigger_info["satisfied"])
+
+    # ------------------------------------------------------------------
+    # describe() tests
+    # ------------------------------------------------------------------
+
+    @patch.object(TriggerManager, "_build_trigger")
+    def test_describe(self, mock_build):
+        """describe returns human-readable summary."""
+        t1 = StubTrigger(True, "t1")
+        t2 = StubTrigger(False, "t2")
+        mock_build.side_effect = [t1, t2]
+
+        config = _make_config(
+            mode="all_of",
+            timeout=30,
+            interval=3,
+            on_timeout="skip",
+            conditions=[
+                {"type": "command", "cmd": "a"},
+                {"type": "command", "cmd": "b"},
+            ],
+        )
+        manager = TriggerManager(config)
+        desc = manager.describe()
+
+        self.assertIsInstance(desc, str)
+        self.assertIn("all_of", desc)
+        self.assertIn("30", desc)
+        self.assertIn("3", desc)
+        self.assertIn("skip", desc)
+        # Should include each trigger's description
+        self.assertIn("t1", desc)
+        self.assertIn("t2", desc)
+
+
+if __name__ == "__main__":
+    unittest.main()