fix: don't spend retry budget on undeliverable dispatch

src/server/dispatcher/base.py

@@ -130,6 +130,43 @@ def _grace_then_fail(
         self._requeue_task(task_id, reason=reason, count_retry=False)
         return False
 
+    def _grace_then_fail_undeliverable(
+        self,
+        task_id: str,
+        record: TaskRecord,
+        reason: str,
+        message: str,
+        extra_payload: dict[str, Any] | None = None,
+    ) -> bool:
+        """A worker was selected but the task could not be delivered to it (its
+        node has no live dispatch subscriber, or publish failed). This is an
+        infrastructure fault, not a task-execution failure: keep retrying without
+        spending the task's ``max_attempts`` budget, then fail with a descriptive
+        error after ``no_worker_grace_sec``. Uses its own clock so the eligibility
+        path resetting ``no_eligible_since`` does not clobber the timer."""
+        self._runtime.release_merge(task_id)
+        record.last_error = message
+        now = time.time()
+        if record.no_dispatch_since is None:
+            record.no_dispatch_since = now
+        waited = now - record.no_dispatch_since
+        if waited >= self._no_worker_grace_sec:
+            self._logger.warning(
+                "Task %s undeliverable after %.0fs (%s); failing",
+                task_id,
+                waited,
+                reason,
+            )
+            payload = {"reason": reason}
+            if extra_payload:
+                payload.update(extra_payload)
+            self._fail_task(
+                task_id, message, worker_id=record.last_failed_worker, payload=payload
+            )
+            return False
+        self._requeue_task(task_id, reason=reason, front=True, count_retry=False)
+        return False
+
     def _grace_then_fail_exhausted(
         self, task_id: str, record: TaskRecord, failed_ids: set[str]
     ) -> bool:
@@ -447,17 +484,37 @@ def dispatch_once(self, task_id: str) -> bool:
             receivers = self._worker_registry.publish_task(worker, message)
         except Exception as exc:
             self._logger.warning("Failed to publish task %s: %s", task_id, exc)
-            self._requeue_task(task_id, reason="publish_failed", front=True)
-            return False
+            return self._grace_then_fail_undeliverable(
+                task_id,
+                record,
+                reason="publish_failed",
+                message=f"Failed to publish task to worker {worker.id}: {exc}",
+                extra_payload={"worker_id": worker.id, "node_id": worker.node_id},
+            )
 
         if receivers <= 0:
             self._logger.info(
-                "No subscribers on tasks channel; delaying task %s", task_id
+                "Node %s dispatch channel has no live subscriber; delaying task %s "
+                "(orphaned worker %s)",
+                worker.node_id,
+                task_id,
+                worker.id,
+            )
+            return self._grace_then_fail_undeliverable(
+                task_id,
+                record,
+                reason="no_dispatch_subscriber",
+                message=(
+                    f"Selected worker {worker.id} on node {worker.node_id} has no "
+                    "live dispatch subscriber (orphaned/stale worker registration); "
+                    "the worker heartbeats but its supervisor is not consuming the "
+                    "node dispatch channel"
+                ),
+                extra_payload={"worker_id": worker.id, "node_id": worker.node_id},
             )
-            self._requeue_task(task_id, reason="no_subscribers", front=True)
-            return False
 
         # 9. Mark dispatched
+        record.no_dispatch_since = None
         self._runtime.mark_dispatched(task_id, worker)
         if merged_children:
             try:

src/server/task/models.py

@@ -126,6 +126,12 @@ class TaskRecord(BaseModel):
         description="Epoch seconds when no eligible worker was first observed.",
         exclude=True,
     )
+    no_dispatch_since: float | None = Field(
+        default=None,
+        description="Epoch seconds when a selected worker was first found "
+        "undeliverable (no live dispatch subscriber / publish failure).",
+        exclude=True,
+    )
     local_name: str | None = Field(default=None, description="Workflow stage name.")
     graph_node_name: str | None = Field(default=None, description="Graph node name.")
     load: int = Field(default=0, description="Load score.")

tests/server/dispatcher/test_dispatch_undeliverable.py

@@ -0,0 +1,67 @@
+"""Undeliverable-dispatch grace behavior.
+
+When a worker is selected but the task cannot be delivered to it (its node has
+no live dispatch subscriber, or publish raises), the dispatcher must treat this
+as an infrastructure fault: retry without spending the task's ``max_attempts``
+budget, then fail with a descriptive error after the no-worker grace.
+"""
+
+import time
+from types import SimpleNamespace
+
+from .helpers import make_capturing_dispatcher
+
+
+def _record() -> SimpleNamespace:
+    return SimpleNamespace(
+        no_dispatch_since=None,
+        last_error=None,
+        last_failed_worker=None,
+    )
+
+
+def test_undeliverable_within_grace_requeues_without_retry() -> None:
+    disp = make_capturing_dispatcher(grace_sec=60)
+    disp._runtime.release_merge = lambda task_id: None
+    record = _record()
+
+    result = disp._grace_then_fail_undeliverable(
+        "tsk-1", record, reason="no_dispatch_subscriber", message="orphaned"
+    )
+
+    assert result is False
+    assert not disp.failed
+    assert len(disp.requeued) == 1
+    _, kwargs = disp.requeued[0]
+    # Infra fault must NOT consume the execution retry budget.
+    assert kwargs["count_retry"] is False
+    assert kwargs["reason"] == "no_dispatch_subscriber"
+    assert record.no_dispatch_since is not None
+    assert record.last_error == "orphaned"
+
+
+def test_undeliverable_after_grace_fails_with_message() -> None:
+    disp = make_capturing_dispatcher(grace_sec=60)
+    disp._runtime.release_merge = lambda task_id: None
+    record = _record()
+    # Pretend the condition has persisted longer than the grace window.
+    record.no_dispatch_since = time.time() - 120
+
+    result = disp._grace_then_fail_undeliverable(
+        "tsk-1",
+        record,
+        reason="no_dispatch_subscriber",
+        message="orphaned worker w1 on node n1",
+        extra_payload={"worker_id": "w1", "node_id": "n1"},
+    )
+
+    assert result is False
+    assert not disp.requeued
+    assert len(disp.failed) == 1
+    task_id, error_message, kwargs = disp.failed[0]
+    assert task_id == "tsk-1"
+    assert error_message == "orphaned worker w1 on node n1"
+    payload = kwargs["payload"]
+    assert payload["reason"] == "no_dispatch_subscriber"
+    assert payload["worker_id"] == "w1"
+    assert payload["node_id"] == "n1"