diff --git a/setup.cfg b/setup.cfg
index 34599d0e..2c2f2bef 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -6,4 +6,5 @@ extend-ignore =
     # See https://github.com/PyCQA/pycodestyle/issues/373
     E203,
 per-file-ignores =
-    tests/test_scontrol_parsing.py:E501
\ No newline at end of file
+    tests/test_scontrol_parsing.py:E501
+    tests/testcases/slurm_signal/handle_signal.py:F821
\ No newline at end of file
diff --git a/snakemake_executor_plugin_slurm/__init__.py b/snakemake_executor_plugin_slurm/__init__.py
index 13be6ed3..b3668bdf 100644
--- a/snakemake_executor_plugin_slurm/__init__.py
+++ b/snakemake_executor_plugin_slurm/__init__.py
@@ -47,6 +47,7 @@
     pending_jobs_for_rule,
     delete_slurm_environment,
     delete_empty_dirs,
+    get_slurm_signal_arg,
     set_gres_string,
 )
 from .job_status_query import (
@@ -352,6 +353,20 @@ class ExecutorSettings(ExecutorSettingsBase):
         },
     )
 
+    signal: Optional[str] = field(
+        default=None,
+        metadata={
+            "help": "Send signal to jobs before wall time (SLURM format). "
+            "Format: --slurm-signal=RULESIGNAL@TIME. "
+            "SIGNAL: name (SIGTERM) or number (15). TIME: seconds before wall time. "
+            "Use RULE='all' for all rules. Examples: "
+            "--slurm-signal=rule1:SIGTERM@30 --slurm-signal=rule2:SIGUSR1@60 "
+            "--slurm-signal=all:15@45",
+            "env_var": False,
+            "required": False,
+        },
+    )
+
     qos: Optional[str] = field(
         default=None,
         metadata={
@@ -636,9 +651,14 @@ def additional_general_args(self):
         passed to `exec_job`.
         """
         general_args = "--executor slurm-jobstep --jobs 1"
-        # need to pass
+        # need to pass, if passing as script is required
         if self.workflow.executor_settings.pass_command_as_script:
             general_args += " --slurm-jobstep-pass-command-as-script"
+        # need to pass, if signal settings are defined
+        if self.workflow.executor_settings.signal:
+            general_args += " --slurm-jobstep-signal " + shlex.quote(
+                self.workflow.executor_settings.signal
+            )
         return general_args
 
     def run_jobs(self, jobs: List[JobExecutorInterface]):
@@ -1088,6 +1108,13 @@ def run_job(self, job: JobExecutorInterface):
             failed_nodes=self._failed_nodes,
         )
 
+        call += get_slurm_signal_arg(
+            self.workflow.executor_settings.signal,
+            job.name,
+        )
+
+        # we exclude failed nodes from further job submissions, to avoid
+        # repeated failures.
         if self._failed_nodes:
             self.logger.debug(
                 "Excluding failed nodes from job submission: "
diff --git a/snakemake_executor_plugin_slurm/utils.py b/snakemake_executor_plugin_slurm/utils.py
index 4de29a4b..5b7c41e7 100644
--- a/snakemake_executor_plugin_slurm/utils.py
+++ b/snakemake_executor_plugin_slurm/utils.py
@@ -6,8 +6,9 @@
 import shlex
 import subprocess
 import re
+from functools import lru_cache
 from pathlib import Path
-from typing import Union
+from typing import Optional, Union
 
 from snakemake_interface_executor_plugins.dag import DAGExecutorInterface
 from snakemake_interface_executor_plugins.jobs import (
@@ -336,6 +337,138 @@ def delete_empty_dirs(path: Path) -> None:
         raise OSError(f"Failed to remove empty directory {path}: {e}") from e
 
 
+# only run this parser once per unique input string
+# chache the results for efficiency.
+@lru_cache(maxsize=None)
+def parse_slurm_signal_settings(signal_settings: Optional[str]) -> dict[str, str]:
+    """
+    Parse rule-specific SLURM signal settings (format: rule:SIGNAL@TIME).
+
+    - rule: rule name or 'all' for all rules
+    - SIGNAL: signal name (SIGTERM, SIGUSR1) or number (15, 10)
+    - TIME: seconds before wall time (must be >= 1)
+
+    Examples:
+    - rule1:SIGTERM@30 → SIGTERM at 30 secs
+    - rule2:SIGUSR1@60 → SIGUSR1 at 60 secs
+    - all:15@45 → all rules, signal 15 at 45 secs
+
+    Returns dict mapping rule names to SLURM signal specs: <signal>@<time>
+    """
+    if not signal_settings:
+        return {}
+
+    parsed_settings = {}
+    # Map signal names to numbers for SLURM
+    signal_map = {
+        "SIGHUP": 1,
+        "SIGINT": 2,
+        "SIGQUIT": 3,
+        "SIGILL": 4,
+        "SIGTRAP": 5,
+        "SIGABRT": 6,
+        "SIGBUS": 7,
+        "SIGFPE": 8,
+        "SIGKILL": 9,
+        "SIGUSR1": 10,
+        "SIGSEGV": 11,
+        "SIGUSR2": 12,
+        "SIGPIPE": 13,
+        "SIGALRM": 14,
+        "SIGTERM": 15,
+        "SIGCHLD": 17,
+        "SIGCONT": 18,
+        "SIGSTOP": 19,
+        "SIGTSTP": 20,
+        "SIGTTIN": 21,
+        "SIGTTOU": 22,
+        "SIGURG": 23,
+        "SIGXCPU": 24,
+        "SIGXFSZ": 25,
+        "SIGVTALRM": 26,
+        "SIGPROF": 27,
+        "SIGWINCH": 28,
+        "SIGIO": 29,
+        "SIGPWR": 30,
+        "SIGSYS": 31,
+    }
+
+    # Pattern: rule:SIGNAL@TIME
+    pattern = re.compile(r"^(?P<rule>\w+):(?P<signal>\w+)@(?P<seconds>\d+)$")
+
+    for raw_entry in signal_settings.split(","):
+        entry = raw_entry.strip()
+        if not entry:
+            raise WorkflowError(
+                "Invalid signal specification: empty entry found. "
+                "Expected 'rule:SIGNAL@TIME' (e.g., 'rule1:SIGTERM@30')."
+            )
+
+        match = pattern.fullmatch(entry)
+        if match is None:
+            raise WorkflowError(
+                f"Invalid signal specification '{entry}'. "
+                "Expected 'rule:SIGNAL@TIME' (e.g., 'rule1:SIGTERM@30', "
+                "'rule2:15@60', 'all:SIGUSR1@45')."
+            )
+
+        rule_name = match.group("rule")
+        if rule_name != "all" and rule_name in parsed_settings:
+            raise WorkflowError(
+                f"Duplicate signal specification for rule '{rule_name}'."
+            )
+
+        signal_str = match.group("signal")
+        # Validate signal name or number.
+        if signal_str.isdigit():
+            signal_num = int(signal_str)
+            if signal_num < 1 or signal_num > 31:
+                raise WorkflowError(
+                    f"Invalid signal number '{signal_num}'. Must be between 1 and 31."
+                )
+        else:
+            if signal_str not in signal_map:
+                raise WorkflowError(
+                    f"Invalid signal name '{signal_str}'. "
+                    f"Valid signals: {', '.join(sorted(signal_map.keys()))}"
+                )
+            signal_num = signal_map[signal_str]
+
+        seconds = int(match.group("seconds"))
+        if seconds < 1:
+            raise WorkflowError("Signal lead time must be at least 1 second.")
+
+        parsed_settings[rule_name] = f"{signal_str}@{seconds}"
+
+    return parsed_settings
+
+
+def get_slurm_signal_arg(signal_settings: Optional[str], rule_name: str) -> str:
+    """Return the sbatch --signal argument for the given rule, if configured.
+
+    Checks for rule-specific signal first, then falls back to 'all' if set.
+    """
+    # we need to check the signal settings, first.
+    if not signal_settings:
+        return ""
+
+    signal_settings_pattern = r"^\s*\w+:\w+@\d+\s*(,\s*\w+:\w+@\d+\s*)*$"
+    if not re.match(signal_settings_pattern, signal_settings):
+        raise WorkflowError(
+            f"Invalid signal specification: '{signal_settings}'. "
+            "Expected format: 'rule:SIGNAL@TIME' "
+            "(e.g., 'rule1:SIGTERM@30', 'rule2:15@60', "
+            "'all:SIGUSR1@45')."
+        )
+
+    parsed = parse_slurm_signal_settings(signal_settings)
+    # Check rule-specific first, then fall back to 'all'
+    signal_value = parsed.get(rule_name) or parsed.get("all")
+    if not signal_value:
+        return ""
+    return f" --signal={signal_value}"
+
+
 def set_gres_string(job: JobExecutorInterface) -> str:
     """
     Function to set the gres string for the SLURM job
diff --git a/snakemake_executor_plugin_slurm/validation.py b/snakemake_executor_plugin_slurm/validation.py
index 560bd170..3fa575f7 100644
--- a/snakemake_executor_plugin_slurm/validation.py
+++ b/snakemake_executor_plugin_slurm/validation.py
@@ -6,6 +6,7 @@
 from pathlib import Path
 from snakemake_interface_common.exceptions import WorkflowError
 from .job_status_query import get_min_job_age, is_query_tool_available
+from .utils import parse_slurm_signal_settings
 
 
 def validate_or_get_slurm_job_id(job_id, output):
@@ -99,6 +100,7 @@ def get_forbidden_slurm_options():
         r"--qos[=\s]": "quality of service",
         r"--nodes[=\s]|-N\s": "number of nodes",
         r"--clusters[=\s]": "cluster specification",
+        r"--signal[=\s]": "signal",
         # GPU options
         r"--gres[=\s]": "generic resources (GRES)",
         r"--gpus[=\s]": "GPU allocation",
@@ -218,6 +220,9 @@ def validate_executor_settings(settings, logger=None):
     if settings.delete_logfiles_older_than is not None:
         if not isinstance(settings.delete_logfiles_older_than, int):
             raise WorkflowError("delete-logfiles-older-than must be an integer (days).")
+    # signal
+    parse_slurm_signal_settings(settings.signal)
+
     # status_command warnings (optional logger)
     if logger:
         validate_status_command_settings(settings, logger)
diff --git a/tests/test_jobsignalling.py b/tests/test_jobsignalling.py
new file mode 100644
index 00000000..2ac993eb
--- /dev/null
+++ b/tests/test_jobsignalling.py
@@ -0,0 +1,230 @@
+"""Test suite for SLURM signal handling in job submission.
+
+This module provides fixtures and tests for SLURM pre-timeout signal support.
+Signals can be directed at either the job payload or the batch script controller,
+and are triggered N seconds before the wall time limit.
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+from snakemake_interface_common.exceptions import WorkflowError
+from snakemake_executor_plugin_slurm import Executor, ExecutorSettings
+from snakemake_executor_plugin_slurm.utils import get_slurm_signal_arg
+from snakemake_executor_plugin_slurm.validation import validate_slurm_extra
+
+
+@pytest.fixture
+def mock_job():
+    """Create a mock job with configurable resources and metadata.
+
+    Returns a factory function that constructs mock jobs for testing.
+    """
+
+    def _create_job(name="signal_shell", **resources):
+        mock_resources = MagicMock()
+        mock_resources.get.side_effect = lambda key, default=None: resources.get(
+            key, default
+        )
+        for key, value in resources.items():
+            setattr(mock_resources, key, value)
+
+        mock_job = MagicMock()
+        mock_job.resources = mock_resources
+        mock_job.name = name
+        mock_job.wildcards = {}
+        mock_job.is_group.return_value = False
+        mock_job.jobid = 1
+        mock_job.rule = MagicMock()
+        mock_job.rule.name = name
+        return mock_job
+
+    return _create_job
+
+
+class TestSlurmSignalParsing:
+    """Test signal specification parsing and transformation to sbatch arguments."""
+
+    def test_signal_targets_batch_script_by_default(self):
+        """Default behavior: signal targets the batch script (B: prefix)."""
+        assert (
+            get_slurm_signal_arg("signal_shell:SIGTERM@45", "signal_shell")
+            == " --signal=B:15@45"
+        )
+
+    def test_signal_with_reservation_target(self):
+        """When R is specified, signal targets the reservation."""
+        assert (
+            get_slurm_signal_arg("signal_python:R:SIGUSR2@30", "signal_python")
+            == " --signal=R:12@30"
+        )
+
+    def test_signal_number_format(self):
+        """Signals can be specified by number (1-31)."""
+        assert (
+            get_slurm_signal_arg("signal_shell:15@60", "signal_shell")
+            == " --signal=B:15@60"
+        )
+
+    def test_signal_is_ignored_for_other_rules(self):
+        """Signals configured for a rule do not apply to other rules."""
+        assert get_slurm_signal_arg("signal_shell:SIGTERM@60", "other_rule") == ""
+
+    def test_multiple_signals_in_one_setting(self):
+        """Multiple rules can have signals in a comma-separated list."""
+        setting = "signal_shell:SIGUSR1@30,signal_python:R:SIGTERM@45"
+        assert get_slurm_signal_arg(setting, "signal_shell") == " --signal=B:10@30"
+        assert get_slurm_signal_arg(setting, "signal_python") == " --signal=R:15@45"
+
+    def test_all_keyword_applies_to_any_rule(self):
+        """The 'all' keyword applies signal to any rule not explicitly configured."""
+        assert get_slurm_signal_arg("all:SIGUSR1@60", "any_rule") == " --signal=B:10@60"
+        assert (
+            get_slurm_signal_arg("all:R:SIGTERM@45", "other_rule")
+            == " --signal=R:15@45"
+        )
+
+    def test_explicit_rule_takes_precedence_over_all(self):
+        """An explicit rule setting overrides the 'all' setting."""
+        setting = "all:SIGUSR1@60,signal_python:SIGTERM@30"
+        # signal_python has explicit config
+        assert get_slurm_signal_arg(setting, "signal_python") == " --signal=B:15@30"
+        # other_rule falls back to 'all'
+        assert get_slurm_signal_arg(setting, "other_rule") == " --signal=B:10@60"
+
+
+class TestSlurmSignalValidation:
+    """Test signal specification validation."""
+
+    def test_invalid_signal_setting_is_rejected(self):
+        """Malformed signal specs are rejected at settings initialization."""
+        with pytest.raises(WorkflowError, match="Invalid signal specification"):
+            ExecutorSettings(signal="signal_shell:SIGTERM")
+
+    def test_reservation_scope_requires_reservation_setting(self):
+        """Using R: scope without --slurm-reservation should raise an error."""
+        with pytest.raises(
+            WorkflowError, match="targets reservation.*no --slurm-reservation"
+        ):
+            ExecutorSettings(signal="signal_shell:R:SIGTERM@30")
+
+    def test_reservation_scope_allowed_with_reservation_setting(self):
+        """Using R: scope with --slurm-reservation should not raise an error."""
+        # Should not raise
+        ExecutorSettings(
+            signal="signal_shell:R:SIGTERM@30", reservation="my_reservation"
+        )
+
+    def test_signal_cannot_be_overridden_via_slurm_extra(self, mock_job):
+        """Prevent silent override: slurm_extra cannot set --signal."""
+        job = mock_job(slurm_extra="--signal=B:SIGTERM@30")
+
+        with pytest.raises(WorkflowError, match=r"signal.*not allowed"):
+            validate_slurm_extra(job)
+
+    def test_slurm_extra_with_other_flags_is_allowed(self, mock_job):
+        """Other sbatch flags in slurm_extra are allowed."""
+        job = mock_job(slurm_extra="--mail-type=END --mail-user=test@example.com")
+        # Should not raise
+        validate_slurm_extra(job)
+
+
+class TestSlurmSignalInExecutor:
+    """Test signal argument injection into the sbatch command during job submission."""
+
+    def test_executor_run_job_adds_signal_argument(self, tmp_path, mock_job):
+        """Verify that Executor.run_job includes --signal in the sbatch call."""
+        executor = Executor.__new__(Executor)
+        executor.workflow = MagicMock(
+            executor_settings=ExecutorSettings(
+                signal="signal_python:R:SIGTERM@15",
+                reservation="my_reservation",
+                init_seconds_before_status_checks=1,
+            ),
+            workdir_init=".",
+        )
+        executor.logger = MagicMock()
+        executor._failed_nodes = set()
+        executor._main_event_loop = None
+        executor.slurm_logdir = tmp_path
+        executor.run_uuid = "test-run-uuid"
+
+        job = mock_job("signal_python", slurm_extra="")
+        job.threads = 1
+        job.resources.get.side_effect = lambda key, default=None: {
+            "runtime": 60,
+            "mem_mb_per_cpu": 1000,
+        }.get(key, default)
+
+        with patch("subprocess.Popen") as mock_popen:
+            process_mock = MagicMock()
+            process_mock.communicate.return_value = ("12345", "")
+            process_mock.returncode = 0
+            mock_popen.return_value = process_mock
+
+            with patch.object(executor, "format_job_exec", return_value="echo test"):
+                with patch.object(executor, "report_job_submission"):
+                    with patch.object(
+                        executor,
+                        "get_account_arg",
+                        return_value=iter([""]),
+                    ):
+                        with patch.object(
+                            executor,
+                            "get_partition_arg",
+                            return_value="",
+                        ):
+                            executor.run_job(job)
+
+            # Verify that --signal was included in the sbatch call
+            assert mock_popen.called
+            call_args = mock_popen.call_args[0][0]
+            assert "--signal=R:15@15" in call_args
+
+    def test_executor_ignores_signal_for_unrelated_rules(self, tmp_path, mock_job):
+        """Executor should not add --signal for rules not in the signal spec."""
+        executor = Executor.__new__(Executor)
+        executor.workflow = MagicMock(
+            executor_settings=ExecutorSettings(
+                signal="signal_python:R:SIGTERM@15",
+                reservation="my_reservation",
+                init_seconds_before_status_checks=1,
+            ),
+            workdir_init=".",
+        )
+        executor.logger = MagicMock()
+        executor._failed_nodes = set()
+        executor._main_event_loop = None
+        executor.slurm_logdir = tmp_path
+        executor.run_uuid = "test-run-uuid"
+
+        job = mock_job("other_rule", slurm_extra="")
+        job.threads = 1
+        job.resources.get.side_effect = lambda key, default=None: {
+            "runtime": 60,
+            "mem_mb_per_cpu": 1000,
+        }.get(key, default)
+
+        with patch("subprocess.Popen") as mock_popen:
+            process_mock = MagicMock()
+            process_mock.communicate.return_value = ("12345", "")
+            process_mock.returncode = 0
+            mock_popen.return_value = process_mock
+
+            with patch.object(executor, "format_job_exec", return_value="echo test"):
+                with patch.object(executor, "report_job_submission"):
+                    with patch.object(
+                        executor,
+                        "get_account_arg",
+                        return_value=iter([""]),
+                    ):
+                        with patch.object(
+                            executor,
+                            "get_partition_arg",
+                            return_value="",
+                        ):
+                            executor.run_job(job)
+
+            # Verify that --signal was NOT included
+            assert mock_popen.called
+            call_args = mock_popen.call_args[0][0]
+            assert "--signal" not in call_args
diff --git a/tests/testcases/slurm_signal/Snakefile b/tests/testcases/slurm_signal/Snakefile
new file mode 100644
index 00000000..17cc33b5
--- /dev/null
+++ b/tests/testcases/slurm_signal/Snakefile
@@ -0,0 +1,24 @@
+rule all:
+    input:
+        "results/shell.txt",
+        "results/python.txt"
+
+
+rule signal_shell:
+    output:
+        "results/shell.txt"
+    resources:
+        runtime=1,
+        mem_mb=128
+    shell:
+        "sh ./handle_signal.sh {output}"
+
+
+rule signal_python:
+    output:
+        "results/python.txt"
+    resources:
+        runtime=1,
+        mem_mb=128
+    script:
+        "handle_signal.py"
\ No newline at end of file
diff --git a/tests/testcases/slurm_signal/handle_signal.py b/tests/testcases/slurm_signal/handle_signal.py
new file mode 100644
index 00000000..256bbc83
--- /dev/null
+++ b/tests/testcases/slurm_signal/handle_signal.py
@@ -0,0 +1,23 @@
+import signal
+import time
+from pathlib import Path
+
+
+output_path = Path(str(snakemake.output[0]))
+signal_path = output_path.with_suffix(output_path.suffix + ".signal")
+signal_received = False
+
+
+def handle_signal(signum, _frame):
+    global signal_received
+    signal_received = True
+    signal_path.write_text(f"{signal.Signals(signum).name}\n")
+
+
+signal.signal(signal.SIGUSR1, handle_signal)
+
+for _ in range(45):
+    time.sleep(1)
+
+status = "python-finished-with-signal\n" if signal_received else "python-finished\n"
+output_path.write_text(status)
diff --git a/tests/testcases/slurm_signal/handle_signal.sh b/tests/testcases/slurm_signal/handle_signal.sh
new file mode 100644
index 00000000..a7b749a2
--- /dev/null
+++ b/tests/testcases/slurm_signal/handle_signal.sh
@@ -0,0 +1,15 @@
+#!/bin/sh
+set -eu
+
+output_file="$1"
+signal_file="${output_file}.signal"
+
+trap 'printf "SIGUSR1\n" > "$signal_file"' USR1
+
+count=0
+while [ "$count" -lt 45 ]; do
+    sleep 1
+    count=$((count + 1))
+done
+
+printf "shell-finished\n" > "$output_file"
\ No newline at end of file
diff --git a/tests/tests.py b/tests/tests.py
index bdba8d12..6e8007da 100644
--- a/tests/tests.py
+++ b/tests/tests.py
@@ -1,6 +1,7 @@
 import os
 import re
 import inspect
+import shutil
 from pathlib import Path
 from typing import Optional
 import snakemake.common.tests
@@ -191,6 +192,67 @@ def test_simple_workflow(self, tmp_path):
         assert report_found, "Efficiency report file not found"
 
 
+@pytest.mark.skipif(shutil.which("sbatch") is None, reason="requires sbatch")
+class TestSignalWorkflow(snakemake.common.tests.TestWorkflowsLocalStorageBase):
+    """Workflow fixture for manual cluster-side validation of SLURM signals."""
+
+    __test__ = True
+
+    def get_executor(self) -> str:
+        return "slurm"
+
+    def get_executor_settings(self) -> Optional[ExecutorSettingsBase]:
+        return ExecutorSettings(
+            signal="signal_shell:SIGUSR1@30,signal_python:SIGUSR1@30",
+            init_seconds_before_status_checks=2,
+        )
+
+    def run_workflow(self, test_name, tmp_path, deployment_method=frozenset()):
+        test_path = Path(__file__).parent / "testcases" / test_name
+        tmp_path = Path(tmp_path) / test_name
+        self._copy_test_files(test_path, tmp_path)
+
+        resource_settings = self.get_resource_settings()
+        resource_settings.cores = 1
+        resource_settings.nodes = 3
+
+        with api.SnakemakeApi(
+            settings.OutputSettings(
+                verbose=True,
+                show_failed_logs=True,
+            ),
+        ) as snakemake_api:
+            workflow_api = snakemake_api.workflow(
+                config_settings=self.get_config_settings(),
+                resource_settings=resource_settings,
+                storage_settings=settings.StorageSettings(
+                    default_storage_provider=self.get_default_storage_provider(),
+                    default_storage_prefix=self.get_default_storage_prefix(),
+                    shared_fs_usage=(
+                        settings.SharedFSUsage.all()
+                        if self.get_assume_shared_fs()
+                        else frozenset()
+                    ),
+                ),
+                deployment_settings=self.get_deployment_settings(deployment_method),
+                storage_provider_settings=self.get_default_storage_provider_settings(),
+                workdir=Path(tmp_path),
+                snakefile=tmp_path / "Snakefile",
+            )
+
+            workflow_api.dag().execute_workflow(
+                executor=self.get_executor(),
+                executor_settings=self.get_executor_settings(),
+                execution_settings=settings.ExecutionSettings(
+                    latency_wait=self.latency_wait,
+                ),
+                remote_execution_settings=self.get_remote_execution_settings(),
+            )
+
+    def test_signal_workflow(self, tmp_path):
+        self.run_workflow("slurm_signal", tmp_path)
+
+
 class TestWorkflowsRequeue(TestWorkflows):
     def get_executor_settings(self) -> Optional[ExecutorSettingsBase]:
         return ExecutorSettings(requeue=True, init_seconds_before_status_checks=1)