diff --git a/osism/commands/apply.py b/osism/commands/apply.py
index 423fa7ac6..ab9cd81c7 100644
--- a/osism/commands/apply.py
+++ b/osism/commands/apply.py
@@ -13,6 +13,22 @@
 from osism.data.enums import Role
 
 
+def _collect_result(result):
+    """Wait for a background task, absorbing an expected play failure.
+
+    A non-zero Ansible rc raises in the worker and is re-raised here. The rc
+    has already been read from the task-output stream, so the exception adds
+    nothing; anything else is unexpected and propagates.
+    """
+    from osism.tasks import AnsibleFailure
+
+    try:
+        result.get()
+    except AnsibleFailure as exc:
+        # Not logger.debug: the sink is pinned to INFO, so that would be silent.
+        logger.info(f"Background task reported a failed play: {exc}")
+
+
 class Run(Command):
     def get_parser(self, prog_name):
         parser = super(Run, self).get_parser(prog_name)
@@ -118,7 +134,7 @@ def handle_loadbalancer_task(self, t, wait, format, timeout):
         # ImportError: sys.meta_path is None, Python is likely shutting down
 
         if not wait:
-            t.parent.get()
+            _collect_result(t.parent)
 
         # process the child tasks
         if format == "log":
@@ -128,7 +144,7 @@ def handle_loadbalancer_task(self, t, wait, format, timeout):
                 )
 
         # As explained above, it is neceesary to wait for all tasks.
-        t.get()
+        _collect_result(t)
 
         return rc
 
@@ -266,6 +282,8 @@ def handle_collection(
             )
             logger.info("Tasks are running in the background")
 
+        return 0
+
     def _prepare_task(
         self,
         arguments,
diff --git a/osism/commands/wait.py b/osism/commands/wait.py
index 4ade62c93..c445dbac2 100644
--- a/osism/commands/wait.py
+++ b/osism/commands/wait.py
@@ -249,6 +249,16 @@ def take_action(self, parsed_args):
                     if output:
                         print(result.get())
 
+                elif result.state in ["FAILURE", "REVOKED"]:
+                    if format == "log":
+                        logger.error(f"Task {task_id} is in state {result.state}")
+                    elif format == "script":
+                        print(f"{task_id} = {result.state}")
+
+                    # Deliberately no result.get() here even with --output:
+                    # Celery re-raises the task's exception from it.
+                    rc = 1
+
                 elif result.state == "STARTED":
                     if format == "log":
                         logger.info(f"Task {task_id} is in state STARTED")
diff --git a/osism/tasks/__init__.py b/osism/tasks/__init__.py
index 47e327aa5..828de9bcf 100644
--- a/osism/tasks/__init__.py
+++ b/osism/tasks/__init__.py
@@ -19,6 +19,18 @@
 HOST_PATTERN = re.compile(r"^(ok|changed|failed|skipping|unreachable):\s+\[([^\]]+)\]")
 
 
+class AnsibleFailure(Exception):
+    """Raised when an Ansible run in a worker container exits non-zero.
+
+    Celery marks a task as failed only if it raises, and only a failed task
+    stops the rest of a chain from running.
+
+    The message must carry its own context: with the JSON result serializer
+    Celery keeps only the exception type and its string form. The play output
+    reaches the operator over the Redis output stream, not through here.
+    """
+
+
 class Config:
     broker_connection_retry_on_startup = True
     enable_utc = True
@@ -352,6 +364,14 @@ def run_ansible_in_environment(
             if publish:
                 utils.finish_task_output(request_id, rc=rc)
 
+            # Raise only after the rc has been logged and published, so the
+            # CLI paths that read it from the output stream are unaffected.
+            if rc != 0:
+                raise AnsibleFailure(
+                    f"{worker} play {role} in environment {environment} "
+                    f"failed with rc {rc}"
+                )
+
             return result
         finally:
             if lock:
diff --git a/tests/unit/commands/test_apply.py b/tests/unit/commands/test_apply.py
index c2229303b..edb6dc977 100644
--- a/tests/unit/commands/test_apply.py
+++ b/tests/unit/commands/test_apply.py
@@ -377,6 +377,22 @@ def test_handle_collection_applies_prepared_group(loguru_logs):
     assert "Tasks are running in the background" in messages
 
 
+def test_handle_collection_returns_zero_exit_code(loguru_logs):
+    """``handle_collection`` has to return an exit code, not ``None``.
+
+    ``take_action`` tests the result with ``if rc != 0``, and ``None != 0`` is
+    True, so returning ``None`` made every collection look like a failure and
+    skip the remaining ``//`` segments.
+    """
+    cmd = make_command(apply.Run)
+    cmd._handle_collection = MagicMock(return_value=MagicMock())
+
+    with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": [Role("a")]}):
+        rc = cmd.handle_collection(**_public_collection_kwargs())
+
+    assert rc == 0
+
+
 def test_handle_collection_show_tree_does_not_apply(loguru_logs):
     cmd = make_command(apply.Run)
     prepared = MagicMock()
@@ -517,6 +533,29 @@ def test_handle_loadbalancer_task_group_failure_propagates(mocker):
     handle_task.assert_called_once_with(t.parent, True, "log", 300)
 
 
+def test_handle_loadbalancer_task_absorbs_ansible_failure(mocker):
+    """A failed play must not surface as a traceback here.
+
+    Now that the task raises on a non-zero Ansible rc, ``t.get()`` re-raises
+    it. The rc has already been read from the task-output stream by
+    ``handle_task``, so the exception carries no new information and must be
+    absorbed -- unlike the unexpected exceptions pinned above, which still
+    propagate.
+    """
+    from osism.tasks import AnsibleFailure
+
+    handle_task = mocker.patch("osism.tasks.handle_task", return_value=4)
+    cmd = make_command(apply.Run)
+    t = MagicMock()
+    t.children = [MagicMock(task_id="child-1")]
+    t.get.side_effect = AnsibleFailure("play failed")
+
+    rc = cmd.handle_loadbalancer_task(t, True, "log", 300)
+
+    assert rc == 4
+    handle_task.assert_called_once_with(t.parent, True, "log", 300)
+
+
 # take_action
 
 
@@ -655,12 +694,6 @@ def test_take_action_routes_collection_to_handle_collection(take_action_mocks):
     assert cmd.handle_collection.call_args.args[4] == "testcollection"
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="handle_collection returns None instead of an exit code, so the "
-    "'//' loop treats a successful collection as failed, silently skips the "
-    "remaining segments and take_action returns None (exit 0)",
-)
 def test_take_action_collection_chain_continues_after_success(take_action_mocks):
     """A successful collection segment must not swallow the following ``//``
     segments: ``osism apply testcollection//other`` has to schedule the
diff --git a/tests/unit/commands/test_wait.py b/tests/unit/commands/test_wait.py
index d4fbb568b..67417ff85 100644
--- a/tests/unit/commands/test_wait.py
+++ b/tests/unit/commands/test_wait.py
@@ -494,3 +494,75 @@ def test_peek_failure_is_reported_at_a_visible_level(loguru_logs):
     notices = [r for r in loguru_logs if "stall reporting" in r["message"].lower()]
     assert len(notices) == 1
     assert notices[0]["level"] in ("WARNING", "ERROR")
+
+
+# terminal failure states
+
+
+def test_failed_task_sets_nonzero_exit_code(loguru_logs):
+    """A FAILURE task must set a non-zero exit code.
+
+    Before the fix the loop branched on PENDING/SUCCESS/STARTED only, so a
+    FAILURE task fell through every branch, was dropped from the queue without
+    being re-queued, and ``rc`` stayed 0 -- which is why an aborted collection
+    chain still let ``deploy-in-a-nutshell.sh`` exit 0 under ``set -e``.
+    """
+    mocks = _run_states(
+        ["taskid1"],
+        results=[_make_result("FAILURE")],
+    )
+
+    assert mocks.rc == 1
+    assert mocks.async_result.call_count == 1
+    mocks.sleep.assert_not_called()
+    assert "Task taskid1 is in state FAILURE" in [
+        record["message"] for record in loguru_logs
+    ]
+
+
+def test_revoked_task_sets_nonzero_exit_code(loguru_logs):
+    mocks = _run_states(
+        ["taskid1"],
+        results=[_make_result("REVOKED")],
+    )
+
+    assert mocks.rc == 1
+    assert "Task taskid1 is in state REVOKED" in [
+        record["message"] for record in loguru_logs
+    ]
+
+
+def test_failure_is_not_reset_by_a_later_successful_task():
+    """``rc`` has to survive the rest of the queue.
+
+    IDs are sorted and popped from the end, so ``taskid2`` is inspected first;
+    the SUCCESS branch that follows must not clear the recorded failure.
+    """
+    mocks = _run_states(
+        ["taskid1", "taskid2"],
+        results=[_make_result("FAILURE"), _make_result("SUCCESS")],
+    )
+
+    assert mocks.rc == 1
+
+
+def test_failed_task_with_output_does_not_fetch_the_result():
+    """``--output`` must not call ``result.get()`` on a failed task: Celery
+    re-raises the task's exception there, which would replace the exit code
+    with a traceback."""
+    result = _make_result("FAILURE")
+    mocks = _run_states(["taskid1", "--output"], results=[result])
+
+    assert mocks.rc == 1
+    result.get.assert_not_called()
+
+
+def test_script_format_prints_failure_state(capsys, loguru_logs):
+    mocks = _run_states(
+        ["taskid1", "--format", "script"],
+        results=[_make_result("FAILURE")],
+    )
+
+    assert mocks.rc == 1
+    assert capsys.readouterr().out == "taskid1 = FAILURE\n"
+    assert not any("taskid1" in record["message"] for record in loguru_logs)
diff --git a/tests/unit/tasks/test_init.py b/tests/unit/tasks/test_init.py
index 584d03ec5..624431e08 100644
--- a/tests/unit/tasks/test_init.py
+++ b/tests/unit/tasks/test_init.py
@@ -528,13 +528,51 @@ def test_run_ansible_logs_start_then_success_with_sorted_hosts(runner_mocks):
 
 
 def test_run_ansible_nonzero_rc_logs_failure(runner_mocks):
+    """A non-zero rc is still logged and still published to the output stream.
+
+    The raise added for chain abortion must happen *after* both, so the
+    non-collection CLI path -- which reads the rc back out of the Redis stream
+    rather than out of the Celery result -- keeps working unchanged.
+    """
     runner_mocks.popen.return_value = make_process(["ok: [node-1]\n"], rc=1)
-    run_ansible()
+    with pytest.raises(tasks.AnsibleFailure):
+        run_ansible()
     calls = runner_mocks.log_play.call_args_list
     assert calls[1].kwargs["result"] == "failure"
     runner_mocks.finish.assert_called_once_with("req-1", rc=1)
 
 
+def test_run_ansible_nonzero_rc_raises_with_context(runner_mocks):
+    """The exception message has to identify the play, since it is all the
+    Celery result backend keeps -- the output itself is not serialized."""
+    runner_mocks.popen.return_value = make_process(["ok: [node-1]\n"], rc=2)
+    with pytest.raises(tasks.AnsibleFailure) as excinfo:
+        run_ansible(worker="kolla-ansible", environment="kolla", role="keystone")
+    message = str(excinfo.value)
+    assert "kolla-ansible" in message
+    assert "kolla" in message
+    assert "keystone" in message
+    assert "2" in message
+
+
+def test_run_ansible_nonzero_rc_releases_lock_and_cleans_ssh_dir(runner_mocks):
+    """Raising must not leak the redlock or the per-task ControlPath dir; both
+    are released in ``finally`` blocks the raise passes through."""
+    lock = runner_mocks.create_redlock.return_value
+    runner_mocks.popen.return_value = make_process(["ok: [node-1]\n"], rc=1)
+
+    with pytest.raises(tasks.AnsibleFailure):
+        run_ansible(locking=True)
+
+    lock.release.assert_called_once_with()
+    runner_mocks.rmtree.assert_called_once()
+
+
+def test_run_ansible_zero_rc_does_not_raise(runner_mocks):
+    runner_mocks.popen.return_value = make_process(["ok: [node-1]\n"], rc=0)
+    assert run_ansible() == "ok: [node-1]\n"
+
+
 def test_run_ansible_duplicate_hosts_deduplicated(runner_mocks):
     runner_mocks.popen.return_value = make_process(
         ["ok: [node-1]\n", "changed: [node-1]\n"], rc=0