diff --git a/osism/commands/configuration.py b/osism/commands/configuration.py
index 250ce7b4..e44ceb8d 100644
--- a/osism/commands/configuration.py
+++ b/osism/commands/configuration.py
@@ -8,11 +8,30 @@
 from osism import utils
 
 
+class _PassthroughParser(argparse.ArgumentParser):
+    """Argument parser that forwards unrecognized arguments to Ansible.
+
+    ``argparse.REMAINDER`` only starts capturing at the first *positional*
+    token, but everything forwarded to Ansible here is option-like (``-e``,
+    ``--limit``, ...) and there is no natural leading positional, so a
+    remainder would reject exactly the arguments it exists to forward.
+    Delegating ``parse_args()`` to ``parse_known_args()`` lets the parser
+    consume only its own options and forward everything else verbatim,
+    preserving order.
+    """
+
+    def parse_args(self, args=None, namespace=None):
+        namespace, arguments = self.parse_known_args(args, namespace)
+        namespace.arguments = arguments
+        return namespace
+
+
 class Sync(Command):
     def get_parser(self, prog_name):
-        parser = super(Sync, self).get_parser(prog_name)
-        parser.add_argument(
-            "arguments", nargs=argparse.REMAINDER, help="Other arguments for Ansible"
+        parser = _PassthroughParser(
+            prog=prog_name,
+            description=self.get_description(),
+            allow_abbrev=False,
         )
         return parser
 
@@ -35,5 +54,5 @@ def take_action(self, parsed_args):
             f"It takes a moment until task {t.task_id} (sync configuration) has been started and output is visible here."
         )
 
-        rc = handle_task(t, True, format, 60)
+        rc = handle_task(t, True, "log", 60)
         return rc
diff --git a/osism/commands/reconciler.py b/osism/commands/reconciler.py
index 4f3fdbd4..5b7ae6bb 100644
--- a/osism/commands/reconciler.py
+++ b/osism/commands/reconciler.py
@@ -15,6 +15,10 @@ def take_action(self, parsed_args):
         logger.info(
             "The osism reconciler command is deprecated and will be removed. Use osism service reconciler."
         )
+
+        # Check if tasks are locked before starting the worker
+        utils.check_task_lock_and_exit()
+
         # NOTE: use python interface in the future, works for the moment
         default_concurrency = min(multiprocessing.cpu_count(), 4)
         concurrency = int(
diff --git a/osism/commands/service.py b/osism/commands/service.py
index cfad1569..fd40523d 100644
--- a/osism/commands/service.py
+++ b/osism/commands/service.py
@@ -6,6 +6,7 @@
 import time
 
 from cliff.command import Command
+from loguru import logger
 
 from osism import utils
 
@@ -101,6 +102,10 @@ def take_action(self, parsed_args):
                 observer.stop()
                 observer.join()
 
+        else:
+            logger.error(f"Unknown service type: {service}")
+            return 1
+
     def watchdog_inventory_on_any_event(self, event):
         from osism.tasks import reconciler
 
diff --git a/osism/commands/task.py b/osism/commands/task.py
index 0fd8eca1..c7b7d817 100644
--- a/osism/commands/task.py
+++ b/osism/commands/task.py
@@ -13,7 +13,7 @@ def take_action(self, parsed_args):
         from celery import Celery
         from osism.tasks import Config
 
-        task = parsed_args.task
+        task = parsed_args.task[0]
 
         app = Celery("task")
         app.config_from_object(Config)
diff --git a/osism/commands/worker.py b/osism/commands/worker.py
index ac69d059..24b40115 100644
--- a/osism/commands/worker.py
+++ b/osism/commands/worker.py
@@ -5,6 +5,8 @@
 import subprocess
 
 from cliff.command import Command
+from loguru import logger
+
 from osism import utils
 
 
@@ -39,9 +41,11 @@ def take_action(self, parsed_args):
         elif queue == "osism-kubernetes":
             tasks = "kubernetes"
             queue = "kubernetes"
-        # kolla-ansible, ceph-ansible, osism-ansible
-        else:
+        elif queue in ["kolla-ansible", "ceph-ansible", "osism-ansible"]:
             tasks = queue[:-8]
+        else:
+            logger.error(f"Unknown worker type: {queue}")
+            return 1
 
         # NOTE: use python interface in the future, works for the moment
         if tasks == "osism":
diff --git a/tests/unit/commands/_helpers.py b/tests/unit/commands/_helpers.py
new file mode 100644
index 00000000..e8e9d595
--- /dev/null
+++ b/tests/unit/commands/_helpers.py
@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared helpers for the ``osism.commands`` unit tests.
+
+The command tests all instantiate a cliff command, parse a small argument
+vector and - for the scheduling and service commands - assert that the task
+lock is consulted before any work is dispatched. These helpers centralize
+those recurring steps so the individual tests stay focused on behavior.
+"""
+
+from unittest.mock import MagicMock
+
+
+def make_command(command_class):
+    """Instantiate a cliff command with a mocked app and command namespace."""
+    return command_class(MagicMock(), MagicMock())
+
+
+def parse_args(command_class, args):
+    """Build ``command_class`` and parse ``args`` with its own parser.
+
+    Returns the ``(command, parsed_args)`` pair; parser-only tests discard the
+    command, while ``take_action`` tests drive it afterwards.
+    """
+    cmd = make_command(command_class)
+    return cmd, cmd.get_parser("test").parse_args(args)
+
+
+def assert_not_called_before_lock_check(guarded_mock):
+    """Return a ``check_task_lock_and_exit`` side effect that enforces ordering.
+
+    Scheduling and service commands must consult the task lock *before* they
+    schedule a Celery task or spawn a subprocess. Install the returned callable
+    as the mocked ``check_task_lock_and_exit`` side effect; it fails the test if
+    ``guarded_mock`` has already run by the time the lock is checked.
+    """
+    return lambda *args, **kwargs: guarded_mock.assert_not_called()
diff --git a/tests/unit/commands/test_configuration.py b/tests/unit/commands/test_configuration.py
new file mode 100644
index 00000000..fb2f7305
--- /dev/null
+++ b/tests/unit/commands/test_configuration.py
@@ -0,0 +1,61 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism configuration sync`` command.
+
+``Sync`` schedules the ``configuration`` playbook on the manager environment
+and returns the task's exit code via ``handle_task``.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import configuration
+
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+def test_parser_forwards_mixed_positional_and_option_arguments():
+    _, parsed_args = parse_args(configuration.Sync, ["playbook", "-e", "foo=1"])
+    assert parsed_args.arguments == ["playbook", "-e", "foo=1"]
+
+
+def test_parser_forwards_leading_option_arguments():
+    # The parser delegates to ``parse_known_args`` (see ``_PassthroughParser``),
+    # so option-like tokens (``-e``, ``--limit``, ...) are forwarded verbatim
+    # even without a leading positional.
+    _, parsed_args = parse_args(configuration.Sync, ["-e", "foo=1"])
+    assert parsed_args.arguments == ["-e", "foo=1"]
+
+
+@pytest.mark.parametrize("rc", [0, 2])
+def test_take_action_schedules_configuration_sync_and_returns_rc(rc):
+    cmd, parsed_args = parse_args(configuration.Sync, ["playbook", "-e", "foo=1"])
+
+    with patch(
+        "osism.commands.configuration.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.tasks.ansible.run.delay") as mock_delay, patch(
+        "osism.tasks.handle_task", return_value=rc
+    ):
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_delay)
+        result = cmd.take_action(parsed_args)
+
+    mock_check.assert_called_once()
+    mock_delay.assert_called_once_with(
+        "manager",
+        "configuration",
+        ["playbook", "-e", "foo=1"],
+        auto_release_time=60,
+    )
+    assert result == rc
+
+
+def test_take_action_waits_with_log_format():
+    cmd, parsed_args = parse_args(configuration.Sync, [])
+
+    with patch("osism.commands.configuration.utils.check_task_lock_and_exit"), patch(
+        "osism.tasks.ansible.run.delay"
+    ) as mock_delay, patch("osism.tasks.handle_task", return_value=0) as mock_handle:
+        cmd.take_action(parsed_args)
+
+    mock_handle.assert_called_once_with(mock_delay.return_value, True, "log", 60)
diff --git a/tests/unit/commands/test_container.py b/tests/unit/commands/test_container.py
new file mode 100644
index 00000000..0faa8727
--- /dev/null
+++ b/tests/unit/commands/test_container.py
@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism container`` command.
+
+``Run`` executes ``docker`` commands on a remote host via SSH, either from
+the command line (one-shot) or from an interactive prompt loop.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import container
+
+from ._helpers import parse_args
+
+
+def _run(args, *, prompt_side_effect=None, known_hosts=True):
+    cmd, parsed_args = parse_args(container.Run, args)
+
+    with patch("osism.commands.container.subprocess.call") as mock_call, patch(
+        "osism.commands.container.ensure_known_hosts_file", return_value=known_hosts
+    ), patch(
+        "osism.commands.container.prompt", side_effect=prompt_side_effect
+    ) as mock_prompt, patch(
+        "osism.commands.container.settings.OPERATOR_USER", "testuser"
+    ):
+        cmd.take_action(parsed_args)
+
+    return mock_call, mock_prompt
+
+
+def _ssh_command(mock_call):
+    """Flatten the ``subprocess.call`` invocation to a single command string.
+
+    The tests assert only the observable contract (ssh target and docker
+    command), so they hold whether the command is passed as a shell string or
+    as an argv vector.
+    """
+    command = mock_call.call_args[0][0]
+    return command if isinstance(command, str) else " ".join(command)
+
+
+def test_parser_splits_host_and_remainder_command():
+    _, parsed_args = parse_args(container.Run, ["node1", "ps", "-a"])
+    assert parsed_args.host == ["node1"]
+    assert parsed_args.command == ["ps", "-a"]
+
+
+def test_one_shot_command_runs_docker_via_ssh():
+    mock_call, mock_prompt = _run(["node1", "ps", "-a"])
+
+    mock_prompt.assert_not_called()
+    mock_call.assert_called_once()
+    ssh_command = _ssh_command(mock_call)
+    assert "testuser@node1" in ssh_command
+    assert ssh_command.endswith("docker ps -a")
+
+
+def test_known_hosts_failure_warns_but_still_connects(loguru_logs):
+    mock_call, _ = _run(["node1", "ps"], known_hosts=False)
+
+    assert any(
+        record["level"] == "WARNING" and "/share/known_hosts" in record["message"]
+        for record in loguru_logs
+    )
+    mock_call.assert_called_once()
+
+
+def test_interactive_prompt_runs_commands_until_exit():
+    mock_call, mock_prompt = _run(["node1"], prompt_side_effect=["ps", "exit"])
+
+    assert mock_prompt.call_count == 2
+    mock_call.assert_called_once()
+    ssh_command = _ssh_command(mock_call)
+    assert "testuser@node1" in ssh_command
+    assert ssh_command.endswith("docker ps")
+
+
+@pytest.mark.parametrize("keyword", ["Exit", "exit", "EXIT"])
+def test_interactive_exit_keywords_break_immediately(keyword):
+    mock_call, mock_prompt = _run(["node1"], prompt_side_effect=[keyword])
+
+    mock_prompt.assert_called_once()
+    mock_call.assert_not_called()
diff --git a/tests/unit/commands/test_lock.py b/tests/unit/commands/test_lock.py
new file mode 100644
index 00000000..8f8965dc
--- /dev/null
+++ b/tests/unit/commands/test_lock.py
@@ -0,0 +1,205 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism lock`` / ``unlock`` / ``lock status`` commands.
+
+The commands manage the Redis-backed task lock via ``osism.utils``; the lock
+logic itself is covered by the ``osism.utils`` tests, so everything is mocked
+at the command module here.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import lock
+
+from ._helpers import make_command, parse_args
+
+
+def _lock(args, *, lock_info, set_result=True):
+    cmd, parsed_args = parse_args(lock.Lock, args)
+
+    with patch(
+        "osism.commands.lock.utils.is_task_locked", return_value=lock_info
+    ), patch(
+        "osism.commands.lock.utils.set_task_lock", return_value=set_result
+    ) as mock_set:
+        result = cmd.take_action(parsed_args)
+
+    return result, mock_set
+
+
+def _unlock(*, lock_info, remove_result=True):
+    cmd, parsed_args = parse_args(lock.Unlock, [])
+
+    with patch(
+        "osism.commands.lock.utils.is_task_locked", return_value=lock_info
+    ), patch(
+        "osism.commands.lock.utils.remove_task_lock", return_value=remove_result
+    ) as mock_remove:
+        result = cmd.take_action(parsed_args)
+
+    return result, mock_remove
+
+
+def _status(lock_info):
+    cmd, parsed_args = parse_args(lock.LockStatus, [])
+
+    with patch("osism.commands.lock.utils.is_task_locked", return_value=lock_info):
+        return cmd.take_action(parsed_args)
+
+
+def test_lock_warns_when_already_locked(loguru_logs):
+    result, mock_set = _lock(
+        [], lock_info={"locked": True, "user": "alice", "timestamp": "2026-01-01"}
+    )
+
+    assert result is None
+    mock_set.assert_not_called()
+    assert any(
+        record["level"] == "WARNING"
+        and "already locked by alice at 2026-01-01" in record["message"]
+        for record in loguru_logs
+    )
+    assert not any("Existing reason" in record["message"] for record in loguru_logs)
+
+
+def test_lock_warns_with_existing_reason(loguru_logs):
+    _lock(
+        [],
+        lock_info={
+            "locked": True,
+            "user": "alice",
+            "timestamp": "2026-01-01",
+            "reason": "upgrade",
+        },
+    )
+
+    assert any(
+        record["level"] == "WARNING" and "Existing reason: upgrade" in record["message"]
+        for record in loguru_logs
+    )
+
+
+@pytest.mark.parametrize("lock_info", [None, {"locked": False}])
+def test_lock_passes_user_and_reason_through(lock_info):
+    result, mock_set = _lock(
+        ["--user", "bob", "--reason", "maintenance"], lock_info=lock_info
+    )
+
+    assert result is None
+    mock_set.assert_called_once_with("bob", "maintenance")
+
+
+def test_lock_user_defaults_to_operator_user():
+    with patch("osism.commands.lock.settings.OPERATOR_USER", "testuser"):
+        result, mock_set = _lock([], lock_info=None)
+
+    assert result is None
+    mock_set.assert_called_once_with("testuser", None)
+
+
+def test_lock_user_help_bakes_in_operator_user_at_parser_build_time():
+    with patch("osism.commands.lock.settings.OPERATOR_USER", "testuser"):
+        parser = make_command(lock.Lock).get_parser("test")
+
+    # The default operator user is interpolated into the ``--user`` help text
+    # when the parser is built, not when the command runs; the rendered help
+    # is the public surface that exposes it.
+    assert "testuser" in parser.format_help()
+
+
+def test_lock_returns_nonzero_when_setting_lock_fails(loguru_logs):
+    result, _ = _lock([], lock_info=None, set_result=False)
+
+    assert result == 1
+    assert any(
+        record["level"] == "ERROR" and "Failed to set task lock" in record["message"]
+        for record in loguru_logs
+    )
+
+
+@pytest.mark.parametrize("lock_info", [None, {"locked": False}])
+def test_unlock_is_a_noop_when_not_locked(loguru_logs, lock_info):
+    result, mock_remove = _unlock(lock_info=lock_info)
+
+    assert result is None
+    mock_remove.assert_not_called()
+    assert any(
+        "Tasks are not currently locked" in record["message"] for record in loguru_logs
+    )
+
+
+def test_unlock_removes_lock_and_logs_previous_owner(loguru_logs):
+    result, mock_remove = _unlock(
+        lock_info={"locked": True, "user": "alice", "timestamp": "2026-01-01"}
+    )
+
+    assert result is None
+    mock_remove.assert_called_once_with()
+    assert any(
+        "was set by alice at 2026-01-01" in record["message"] for record in loguru_logs
+    )
+    assert not any("Previous reason" in record["message"] for record in loguru_logs)
+
+
+def test_unlock_logs_previous_reason_when_present(loguru_logs):
+    _unlock(
+        lock_info={
+            "locked": True,
+            "user": "alice",
+            "timestamp": "2026-01-01",
+            "reason": "upgrade",
+        }
+    )
+
+    assert any(
+        "Previous reason: upgrade" in record["message"] for record in loguru_logs
+    )
+
+
+def test_unlock_returns_nonzero_when_removing_lock_fails(loguru_logs):
+    result, _ = _unlock(
+        lock_info={"locked": True, "user": "alice", "timestamp": "2026-01-01"},
+        remove_result=False,
+    )
+
+    assert result == 1
+    assert any(
+        record["level"] == "ERROR" and "Failed to remove task lock" in record["message"]
+        for record in loguru_logs
+    )
+
+
+def test_status_locked_with_reason(loguru_logs):
+    _status(
+        {
+            "locked": True,
+            "user": "alice",
+            "timestamp": "2026-01-01",
+            "reason": "upgrade",
+        }
+    )
+
+    assert any(
+        "Tasks are LOCKED by alice at 2026-01-01" in record["message"]
+        for record in loguru_logs
+    )
+    assert any("Reason: upgrade" in record["message"] for record in loguru_logs)
+
+
+def test_status_locked_falls_back_to_unknown(loguru_logs):
+    _status({"locked": True})
+
+    assert any(
+        "Tasks are LOCKED by unknown at unknown" in record["message"]
+        for record in loguru_logs
+    )
+    assert not any("Reason:" in record["message"] for record in loguru_logs)
+
+
+@pytest.mark.parametrize("lock_info", [None, {"locked": False}])
+def test_status_unlocked(loguru_logs, lock_info):
+    _status(lock_info)
+
+    assert any("Tasks are UNLOCKED" in record["message"] for record in loguru_logs)
diff --git a/tests/unit/commands/test_noset.py b/tests/unit/commands/test_noset.py
new file mode 100644
index 00000000..b1186f5d
--- /dev/null
+++ b/tests/unit/commands/test_noset.py
@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism noset`` commands.
+
+``NoMaintenance`` and ``NoBootstrap`` mirror the ``osism set`` commands with
+``status=False``; they differ only in the playbook name.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import noset
+
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+@pytest.mark.parametrize(
+    "command_class, playbook",
+    [
+        (noset.NoMaintenance, "state-maintenance"),
+        (noset.NoBootstrap, "state-bootstrap"),
+    ],
+)
+def test_take_action_schedules_state_playbook(command_class, playbook):
+    cmd, parsed_args = parse_args(command_class, ["node1"])
+
+    with patch(
+        "osism.commands.noset.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.tasks.ansible.run.delay") as mock_delay:
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_delay)
+        cmd.take_action(parsed_args)
+
+    mock_check.assert_called_once()
+    mock_delay.assert_called_once_with(
+        "generic", playbook, ["-e status=False", "-l node1"]
+    )
diff --git a/tests/unit/commands/test_reconciler.py b/tests/unit/commands/test_reconciler.py
index 319489ae..30052f49 100644
--- a/tests/unit/commands/test_reconciler.py
+++ b/tests/unit/commands/test_reconciler.py
@@ -2,19 +2,119 @@
 
 """Tests for the ``osism reconciler`` commands.
 
-These focus on the exit-code contract: a command must return a non-zero exit
-status when it gives up waiting for a task (a timeout is an operational
-failure), rather than falling through to an implicit success.
+``Run`` is the deprecated foreground worker runner; ``Sync`` schedules an
+inventory sync via Celery and (by default) waits for its output. The exit-code
+contract matters most: a command must return a non-zero exit status when it
+gives up waiting for a task (a timeout is an operational failure), rather than
+falling through to an implicit success.
 """
 
 from unittest.mock import MagicMock, patch
 
 from osism.commands import reconciler
 
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+def _run_worker(monkeypatch, *, env=None, cpu_count=16):
+    monkeypatch.delenv("OSISM_CELERY_CONCURRENCY", raising=False)
+    if env is not None:
+        monkeypatch.setenv("OSISM_CELERY_CONCURRENCY", env)
+
+    cmd, parsed_args = parse_args(reconciler.Run, [])
+
+    with patch(
+        "osism.commands.reconciler.utils.check_task_lock_and_exit"
+    ) as mock_check, patch(
+        "osism.commands.reconciler.subprocess.Popen"
+    ) as mock_popen, patch(
+        "osism.commands.reconciler.multiprocessing.cpu_count",
+        return_value=cpu_count,
+    ):
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_popen)
+        cmd.take_action(parsed_args)
+
+    return mock_check, mock_popen
+
+
+def _run_sync(args, *, fetch_return=0):
+    cmd, parsed_args = parse_args(reconciler.Sync, args)
+
+    with patch(
+        "osism.commands.reconciler.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.tasks.reconciler.run.delay") as mock_delay, patch(
+        "osism.commands.reconciler.utils.fetch_task_output",
+        return_value=fetch_return,
+    ) as mock_fetch:
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_delay)
+        result = cmd.take_action(parsed_args)
+
+    return result, mock_check, mock_delay, mock_fetch
+
+
+def test_run_logs_deprecation_warning(loguru_logs, monkeypatch):
+    _run_worker(monkeypatch)
+    assert any(
+        "deprecated" in record["message"]
+        and "osism service reconciler" in record["message"]
+        for record in loguru_logs
+    )
+
+
+def test_run_checks_task_lock_before_starting_worker(monkeypatch):
+    mock_check, mock_popen = _run_worker(monkeypatch)
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once()
+
+
+def test_run_concurrency_defaults_to_cpu_count_capped_at_four(monkeypatch):
+    _, mock_popen = _run_worker(monkeypatch, cpu_count=16)
+    mock_popen.assert_called_once_with(
+        "celery -A osism.tasks.reconciler worker -n reconciler --loglevel=INFO -Q reconciler -c 4",
+        shell=True,
+    )
+    mock_popen.return_value.wait.assert_called_once_with()
+
+
+def test_run_concurrency_from_env(monkeypatch):
+    _, mock_popen = _run_worker(monkeypatch, env="2")
+    assert "-c 2" in mock_popen.call_args[0][0]
+
+
+def test_sync_waits_for_task_output_and_returns_its_result():
+    result, mock_check, mock_delay, mock_fetch = _run_sync([], fetch_return=0)
+
+    mock_check.assert_called_once()
+    mock_delay.assert_called_once_with(publish=True)
+    mock_fetch.assert_called_once_with(mock_delay.return_value.id, timeout=300)
+    assert result == 0
+
+
+def test_sync_no_wait_schedules_without_fetching_output(loguru_logs):
+    result, _, mock_delay, mock_fetch = _run_sync(["--no-wait"])
+
+    mock_delay.assert_called_once_with(publish=False)
+    mock_fetch.assert_not_called()
+    assert result is None
+    assert any("No more output" in record["message"] for record in loguru_logs)
+
+
+def test_sync_task_timeout_argument_is_forwarded():
+    _, _, _, mock_fetch = _run_sync(["--task-timeout", "60"])
+    mock_fetch.assert_called_once()
+    assert mock_fetch.call_args[1]["timeout"] == 60
+
+
+def test_sync_task_timeout_default_from_env(monkeypatch):
+    # The environment variable is read at parser-build time; argparse applies
+    # ``type=int`` to string defaults, so the value arrives as an int.
+    monkeypatch.setenv("OSISM_TASK_TIMEOUT", "600")
+    _, parsed_args = parse_args(reconciler.Sync, [])
+    assert parsed_args.task_timeout == 600
+
 
 def test_sync_returns_nonzero_on_task_timeout():
-    cmd = reconciler.Sync(MagicMock(), MagicMock())
-    parsed_args = cmd.get_parser("test").parse_args([])
+    cmd, parsed_args = parse_args(reconciler.Sync, [])
 
     with patch("osism.commands.reconciler.utils.check_task_lock_and_exit"), patch(
         "osism.tasks.reconciler.run.delay", return_value=MagicMock()
diff --git a/tests/unit/commands/test_service.py b/tests/unit/commands/test_service.py
new file mode 100644
index 00000000..8088019f
--- /dev/null
+++ b/tests/unit/commands/test_service.py
@@ -0,0 +1,168 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism service`` command.
+
+``Run`` launches one of the long-running OSISM services (API, listener,
+Celery beat/flower, reconciler worker, inventory watchdog) as subprocesses,
+except for the watchdog, which polls the inventory directory in-process.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from osism.commands import service
+
+from ._helpers import assert_not_called_before_lock_check, make_command, parse_args
+
+
+class _Stop(Exception):
+    """Sentinel to escape the watchdog ``while True`` loop."""
+
+
+def _run_service(args):
+    cmd, parsed_args = parse_args(service.Run, args)
+
+    with patch(
+        "osism.commands.service.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.commands.service.subprocess.Popen") as mock_popen:
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_popen)
+        result = cmd.take_action(parsed_args)
+
+    return result, mock_check, mock_popen
+
+
+def test_api_starts_uvicorn():
+    result, mock_check, mock_popen = _run_service(["api"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once_with(
+        "uvicorn osism.api:app --host 0.0.0.0 --port 8000", shell=True
+    )
+    mock_popen.return_value.wait.assert_called_once_with()
+    assert result is None
+
+
+def test_listener_starts_listener_service():
+    result, mock_check, mock_popen = _run_service(["listener"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once_with(
+        "python3 -c 'from osism.services import listener; listener.main()'",
+        shell=True,
+    )
+    mock_popen.return_value.wait.assert_called_once_with()
+    assert result is None
+
+
+def test_beat_starts_one_scheduler_per_task_module():
+    from osism.tasks import Config
+
+    result, mock_check, mock_popen = _run_service(["beat"])
+
+    mock_check.assert_called_once()
+    modules = [
+        "ansible",
+        "ceph",
+        "conductor",
+        "kolla",
+        "netbox",
+        "openstack",
+        "reconciler",
+    ]
+    assert mock_popen.call_count == len(modules)
+    for popen_call, module in zip(mock_popen.call_args_list, modules):
+        assert popen_call[0][0] == [
+            "celery",
+            "-A",
+            f"osism.tasks.{module}",
+            "--broker",
+            Config.broker_url,
+            "beat",
+            "-s",
+            f"/tmp/celerybeat-schedule-osism.tasks.{module}.db",
+        ]
+    assert mock_popen.return_value.wait.call_count == len(modules)
+    assert result is None
+
+
+def test_flower_starts_celery_flower():
+    from osism.tasks import Config
+
+    result, mock_check, mock_popen = _run_service(["flower"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once_with(
+        ["celery", "--broker", Config.broker_url, "flower"]
+    )
+    mock_popen.return_value.wait.assert_called_once_with()
+    assert result is None
+
+
+def test_reconciler_concurrency_defaults_to_cpu_count_capped_at_four(monkeypatch):
+    monkeypatch.delenv("OSISM_CELERY_CONCURRENCY", raising=False)
+    with patch("osism.commands.service.multiprocessing.cpu_count", return_value=16):
+        result, mock_check, mock_popen = _run_service(["reconciler"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once_with(
+        "celery -A osism.tasks.reconciler worker -n reconciler --loglevel=INFO -Q reconciler -c 4",
+        shell=True,
+    )
+    mock_popen.return_value.wait.assert_called_once_with()
+    assert result is None
+
+
+def test_reconciler_concurrency_from_env(monkeypatch):
+    monkeypatch.setenv("OSISM_CELERY_CONCURRENCY", "2")
+    _, _, mock_popen = _run_service(["reconciler"])
+    assert "-c 2" in mock_popen.call_args[0][0]
+
+
+def test_unknown_service_type_errors(loguru_logs):
+    result, mock_check, mock_popen = _run_service(["bogus"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_not_called()
+    assert result == 1
+    assert any(
+        record["level"] == "ERROR" and "bogus" in record["message"]
+        for record in loguru_logs
+    )
+
+
+def test_watchdog_observes_inventory_and_cleans_up():
+    cmd, parsed_args = parse_args(service.Run, ["watchdog"])
+
+    with patch(
+        "osism.commands.service.utils.check_task_lock_and_exit"
+    ) as mock_check, patch(
+        "watchdog.observers.polling.PollingObserver"
+    ) as mock_observer_class, patch(
+        "osism.commands.service.time.sleep", side_effect=_Stop
+    ):
+        with pytest.raises(_Stop):
+            cmd.take_action(parsed_args)
+
+    mock_check.assert_called_once()
+    observer = mock_observer_class.return_value
+
+    event_handler = observer.schedule.call_args[0][0]
+    assert event_handler.on_any_event == cmd.watchdog_inventory_on_any_event
+    observer.schedule.assert_called_once_with(
+        event_handler, "/opt/configuration/inventory", recursive=True
+    )
+    observer.start.assert_called_once_with()
+    # The ``finally`` block must stop and join the observer even when the
+    # loop exits via an exception.
+    observer.stop.assert_called_once_with()
+    observer.join.assert_called_once_with()
+
+
+def test_watchdog_inventory_event_triggers_inventory_sync():
+    cmd = make_command(service.Run)
+
+    with patch("osism.tasks.reconciler.run.delay") as mock_delay:
+        cmd.watchdog_inventory_on_any_event(MagicMock())
+
+    mock_delay.assert_called_once_with()
diff --git a/tests/unit/commands/test_set.py b/tests/unit/commands/test_set.py
new file mode 100644
index 00000000..69b10535
--- /dev/null
+++ b/tests/unit/commands/test_set.py
@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism set`` commands.
+
+``Maintenance`` and ``Bootstrap`` schedule the corresponding state playbook
+with ``status=True`` for a single host; they differ only in the playbook
+name.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import set as set_cmd
+
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+@pytest.mark.parametrize(
+    "command_class, playbook",
+    [
+        (set_cmd.Maintenance, "state-maintenance"),
+        (set_cmd.Bootstrap, "state-bootstrap"),
+    ],
+)
+def test_take_action_schedules_state_playbook(command_class, playbook):
+    cmd, parsed_args = parse_args(command_class, ["node1"])
+
+    with patch(
+        "osism.commands.set.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.tasks.ansible.run.delay") as mock_delay:
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_delay)
+        cmd.take_action(parsed_args)
+
+    mock_check.assert_called_once()
+    mock_delay.assert_called_once_with(
+        "generic", playbook, ["-e status=True", "-l node1"]
+    )
diff --git a/tests/unit/commands/test_task.py b/tests/unit/commands/test_task.py
new file mode 100644
index 00000000..82a23421
--- /dev/null
+++ b/tests/unit/commands/test_task.py
@@ -0,0 +1,32 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism task revoke`` command.
+
+``Revoke`` is a thin wrapper around the Celery control API: it builds an app
+from ``osism.tasks.Config`` and revokes the given task with termination.
+"""
+
+from unittest.mock import patch
+
+from osism.commands import task
+
+from ._helpers import parse_args
+
+
+def test_parser_task_is_a_one_element_list():
+    _, parsed_args = parse_args(task.Revoke, ["abc"])
+    assert parsed_args.task == ["abc"]
+
+
+def test_revoke_terminates_task_by_id():
+    from osism.tasks import Config
+
+    cmd, parsed_args = parse_args(task.Revoke, ["abc"])
+
+    with patch("celery.Celery") as mock_celery:
+        cmd.take_action(parsed_args)
+
+    mock_celery.assert_called_once_with("task")
+    app = mock_celery.return_value
+    app.config_from_object.assert_called_once_with(Config)
+    app.control.revoke.assert_called_once_with("abc", terminate=True)
diff --git a/tests/unit/commands/test_worker.py b/tests/unit/commands/test_worker.py
new file mode 100644
index 00000000..62d6852f
--- /dev/null
+++ b/tests/unit/commands/test_worker.py
@@ -0,0 +1,106 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism worker`` command.
+
+``Run`` maps a worker type to a Celery tasks module and starts the worker
+process via ``subprocess.Popen``. The concurrency default is derived from
+``OSISM_CELERY_CONCURRENCY`` (or the CPU count, capped at 4) at parser-build
+time.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import worker
+
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+def _run_worker(args):
+    cmd, parsed_args = parse_args(worker.Run, args)
+
+    with patch(
+        "osism.commands.worker.utils.check_task_lock_and_exit"
+    ) as mock_check, patch("osism.commands.worker.subprocess.Popen") as mock_popen:
+        mock_check.side_effect = assert_not_called_before_lock_check(mock_popen)
+        result = cmd.take_action(parsed_args)
+
+    return result, mock_check, mock_popen
+
+
+@pytest.mark.parametrize("cpu_count, expected", [(2, 2), (16, 4)])
+def test_parser_default_workers_is_cpu_count_capped_at_four(
+    monkeypatch, cpu_count, expected
+):
+    monkeypatch.delenv("OSISM_CELERY_CONCURRENCY", raising=False)
+    with patch(
+        "osism.commands.worker.multiprocessing.cpu_count", return_value=cpu_count
+    ):
+        _, parsed_args = parse_args(worker.Run, ["openstack"])
+    assert parsed_args.number_of_workers == expected
+
+
+def test_parser_default_workers_from_env(monkeypatch):
+    # The environment variable is read at parser-build time.
+    monkeypatch.setenv("OSISM_CELERY_CONCURRENCY", "7")
+    _, parsed_args = parse_args(worker.Run, ["openstack"])
+    assert parsed_args.number_of_workers == 7
+
+
+@pytest.mark.parametrize(
+    "worker_type, expected_command",
+    [
+        (
+            "openstack",
+            "celery -A osism.tasks.openstack worker -n openstack --loglevel=INFO -Q openstack -c 1",
+        ),
+        (
+            "netbox",
+            "celery -A osism.tasks.netbox worker -n netbox --loglevel=INFO -Q netbox -c 1",
+        ),
+        (
+            "conductor",
+            "celery -A osism.tasks.conductor worker -n conductor --loglevel=INFO -Q conductor -c 1",
+        ),
+        (
+            "osism-kubernetes",
+            "celery -A osism.tasks.kubernetes worker -n kubernetes --loglevel=INFO -Q kubernetes -c 1",
+        ),
+        (
+            "kolla-ansible",
+            "celery -A osism.tasks.kolla worker -n kolla-ansible --loglevel=INFO -Q kolla-ansible -c 1",
+        ),
+        (
+            "ceph-ansible",
+            "celery -A osism.tasks.ceph worker -n ceph-ansible --loglevel=INFO -Q ceph-ansible -c 1",
+        ),
+        (
+            "osism-ansible",
+            "celery -A osism.tasks.ansible worker -n osism-ansible --loglevel=INFO -Q osism-ansible -c 1",
+        ),
+    ],
+)
+def test_take_action_maps_worker_type_to_celery_command(worker_type, expected_command):
+    _, mock_check, mock_popen = _run_worker(["--number-of-workers", "1", worker_type])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_called_once_with(expected_command, shell=True)
+    mock_popen.return_value.wait.assert_called_once_with()
+
+
+def test_take_action_respects_number_of_workers():
+    _, _, mock_popen = _run_worker(["--number-of-workers", "3", "openstack"])
+    assert "-c 3" in mock_popen.call_args[0][0]
+
+
+def test_take_action_rejects_unknown_worker_type(loguru_logs):
+    result, mock_check, mock_popen = _run_worker(["--number-of-workers", "1", "foo"])
+
+    mock_check.assert_called_once()
+    mock_popen.assert_not_called()
+    assert result == 1
+    assert any(
+        record["level"] == "ERROR" and "foo" in record["message"]
+        for record in loguru_logs
+    )