fix: gate SSH dispatch to Docker-capable workers (#74)

Author: kaiitunnz

docs/ARCHITECTURE.md

@@ -127,6 +127,12 @@ scripts/dev/            compile_protos, sync_requirements, check_env_examples
   `WorkerHardware`. The dispatcher's `_cached_worker_candidates` filters
   to workers whose cache covers the task's references; entries older
   than `WORKER_CACHE_TTL_SEC` are ignored.
+- **Worker capabilities.** Beyond hardware fit, each worker advertises a
+  `WorkerCapabilities` describing the task kinds it can service. The dispatcher
+  routes a task that requires a capability only to workers advertising it, and
+  excludes a worker that advertises none from those tasks rather than failing
+  them on it. SSH is one such capability — a worker advertises it when it can
+  launch session containers.
 - **Cursor pagination.** List endpoints accept `limit` and `before` /
   `after` cursors. The cursor is an opaque base64 of `(timestamp, id)`;
   do not parse client-side.

pyproject.toml

@@ -23,7 +23,7 @@ all = [
 # Runtime/container dependencies.
 runtime-server = [
     "aiohttp>=3.14.0",
-    "cryptography>=46.0.7",
+    "cryptography>=48.0.1",
     "docker>=7.1.0",
     "fastapi>=0.135.0",
     "flowmesh-hook",
@@ -32,7 +32,7 @@ runtime-server = [
     "lumid-hooks>=0.2.0",
     "protobuf>=5.29.6",
     "pydantic>=2.12.3",
-    "python-multipart>=0.0.26",
+    "python-multipart>=0.0.31",
     "pyyaml>=6.0.2",
     "redis>=7.0.1",
     "requests>=2.33.0",
@@ -83,7 +83,7 @@ runtime-agent = [
     "arxiv>=2.2.0",
     "beautifulsoup4>=4.14.3",
     "chunkr-ai>=0.3.7",
-    "crawl4ai>=0.8.0",
+    "crawl4ai>=0.8.9",
     "ddgs>=9.10.0",
     "google-genai>=1.47.0",
     "hydra-core>=1.3.2",

sdk/src/flowmesh/models/__init__.py

@@ -42,6 +42,7 @@
     SSHLimits,
     StorageInfo,
     Worker,
+    WorkerCapabilities,
     WorkerHardware,
     WorkerInfo,
 )
@@ -91,6 +92,7 @@
     "TaskType",
     "TaskUsage",
     "Worker",
+    "WorkerCapabilities",
     "WorkerHardware",
     "WorkerInfo",
     "WorkerRegisterResponse",

sdk/src/flowmesh/models/workers.py

@@ -79,6 +79,10 @@ class SSHLimits(BaseModel):
     max_pids: int | None = None
 
 
+class WorkerCapabilities(BaseModel):
+    ssh: bool = False
+
+
 class Worker(BaseModel):
     id: str
     alias: str | None = None
@@ -91,6 +95,7 @@ class Worker(BaseModel):
     pid: int | None = None
     env: dict[str, Any] = Field(default_factory=dict)
     hardware: WorkerHardware | None = None
+    capabilities: WorkerCapabilities = Field(default_factory=WorkerCapabilities)
     ssh_limits: SSHLimits | None = None
     tags: list[str] = Field(default_factory=list)
     last_seen: str | None = None

src/server/dispatcher/base.py

@@ -178,7 +178,8 @@ def dispatch_once(self, task_id: str) -> bool:
                     task_id,
                     record,
                     reason="no_eligible_worker",
-                    message="No worker satisfies the task hardware requirements",
+                    message="No worker satisfies the task hardware and capability "
+                    "requirements",
                 )
             if not (eligible - failed_ids):
                 return self._grace_then_fail_exhausted(task_id, record, failed_ids)

src/server/registries/worker.py

@@ -9,7 +9,7 @@
     StopMessage,
     TaskMessage,
 )
-from shared.schemas.worker import SSHLimits
+from shared.schemas.worker import SSHLimits, WorkerCapabilities
 from shared.tasks import TaskEnvelope
 from shared.tasks.components.resources import GPURequirements
 from shared.tasks.specs import SSHSpecStrict, SSHSpecTemplate
@@ -55,6 +55,10 @@ class Worker(BaseModel):
     hardware: WorkerHardware | None = Field(
         default=None, description="Hardware metadata."
     )
+    capabilities: WorkerCapabilities = Field(
+        default_factory=WorkerCapabilities,
+        description="Task capabilities the worker advertises.",
+    )
     ssh_limits: SSHLimits | None = Field(
         default=None, description="Configured ceiling on SSH session resources."
     )
@@ -372,18 +376,18 @@ def idle_satisfying_pool(self, task: TaskEnvelope) -> list[Worker]:
                 continue
             if self.is_worker_stale(worker.id):
                 continue
-            if hw_satisfies(worker, task):
+            if hw_satisfies(worker, task) and capability_satisfies(worker, task):
                 available.append(worker)
         return self.sort_workers(available)
 
     def satisfying_workers(self, task: TaskEnvelope) -> list[Worker]:
-        """Non-stale workers whose hardware satisfies the task, any status."""
+        """Non-stale workers whose hardware and capabilities satisfy the task."""
         available: list[Worker] = []
         for worker_id in self.get_worker_ids():
             worker = self.get_worker(worker_id)
             if not worker or self.is_worker_stale(worker.id):
                 continue
-            if hw_satisfies(worker, task):
+            if hw_satisfies(worker, task) and capability_satisfies(worker, task):
                 available.append(worker)
         return self.sort_workers(available)
 
@@ -545,6 +549,12 @@ def hw_satisfies(worker: Worker, task: TaskEnvelope) -> bool:
     return True
 
 
+def capability_satisfies(worker: Worker, task: TaskEnvelope) -> bool:
+    if isinstance(task.spec, (SSHSpecStrict, SSHSpecTemplate)):
+        return worker.capabilities.ssh
+    return True
+
+
 def _gpu_meets_requirements(hw: WorkerHardware, gpu_req: GPURequirements) -> bool:
     required_count = gpu_req.count
     if required_count is not None:
@@ -642,6 +652,12 @@ def _ensure_str_list(items: Any) -> list[str]:
         if hardware_json is None
         else WorkerHardware.model_validate_json(hardware_json)
     )
+    capabilities_json = value.get("capabilities_json")
+    capabilities = (
+        WorkerCapabilities()
+        if capabilities_json is None
+        else WorkerCapabilities.model_validate_json(capabilities_json)
+    )
     ssh_limits_json = value.get("ssh_limits_json")
     ssh_limits = (
         None
@@ -677,6 +693,7 @@ def _ensure_str_list(items: Any) -> list[str]:
         pid=pid,
         env=env,
         hardware=hardware,
+        capabilities=capabilities,
         ssh_limits=ssh_limits,
         tags=tags,
         last_seen=value.get("last_seen"),

src/server/requirements.txt

@@ -1,15 +1,15 @@
 # AUTO-GENERATED by scripts/dev/sync_requirements.py — do not edit by hand.
 # Regenerate with: uv run scripts/dev/sync_requirements.py --write
 aiohttp==3.14.1
-cryptography==47.0.0
+cryptography==49.0.0
 docker==7.1.0
 fastapi==0.136.1
 grpcio==1.76.0
 httpx==0.28.1
 lumid-hooks==0.2.0
 protobuf==5.29.6
 pydantic==2.12.3
-python-multipart==0.0.27
+python-multipart==0.0.32
 pyyaml==6.0.2
 redis==7.0.1
 requests==2.33.1

src/shared/schemas/worker.py

@@ -30,4 +30,12 @@ class SSHLimits(BaseModel):
     )
 
 
-__all__ = ["SSHLimits", "WorkerStatus"]
+class WorkerCapabilities(BaseModel):
+    """Task capabilities a worker advertises to the dispatcher."""
+
+    ssh: bool = Field(
+        default=False, description="Whether the worker can run SSH session tasks."
+    )
+
+
+__all__ = ["SSHLimits", "WorkerCapabilities", "WorkerStatus"]

src/worker/README.md

@@ -119,6 +119,12 @@ The SSH session container is labelled with `flowmesh.ssh.task_id` and
 `flowmesh.ssh.worker_id` so the server can clean it up if the worker
 container is stopped externally.
 
+At startup the worker initializes the SSH executor only if the Docker daemon is
+reachable, and advertises its `ssh` capability accordingly; the dispatcher then
+selects it for SSH tasks only when it can actually run them. (The supervisor
+mounts the Docker socket only when SSH is enabled for the node; see
+`ENABLE_SSH_BY_DEFAULT` below.)
+
 Relevant env vars for SSH tasks:
 
 | Variable | Default | Description |

src/worker/executors/__init__.py

@@ -1,55 +1,60 @@
 import importlib
 
+from .base_executor import Executor
+
 _IMPORT_ERRORS: dict[str, str] = {}
 
 
-def _safe_import(name: str, module: str) -> type | None:
+def _import_executor(name: str, module: str) -> type[Executor] | None:
     try:
         pkg = importlib.import_module(module, package=__package__)
-        return getattr(pkg, name)
+        if issubclass(cls := getattr(pkg, name), Executor):
+            return cls
+        error = f"{name} is not a subclass of Executor"
     except Exception as exc:
-        _IMPORT_ERRORS[name] = str(exc)
-        return None
+        error = str(exc)
+    _IMPORT_ERRORS[name] = error
+    return None
 
 
-VLLMExecutor = _safe_import("VLLMExecutor", ".vllm_executor")
-VLLMLoRAExecutor = _safe_import("VLLMLoRAExecutor", ".vllm_lora_executor")
-PPOExecutor = _safe_import("PPOExecutor", ".ppo_executor")
-DPOExecutor = _safe_import("DPOExecutor", ".dpo_executor")
-SFTExecutor = _safe_import("SFTExecutor", ".sft_executor")
-LoRASFTExecutor = _safe_import("LoRASFTExecutor", ".lora_sft_executor")
-ImageClassificationTrainingExecutor = _safe_import(
+VLLMExecutor = _import_executor("VLLMExecutor", ".vllm_executor")
+VLLMLoRAExecutor = _import_executor("VLLMLoRAExecutor", ".vllm_lora_executor")
+PPOExecutor = _import_executor("PPOExecutor", ".ppo_executor")
+DPOExecutor = _import_executor("DPOExecutor", ".dpo_executor")
+SFTExecutor = _import_executor("SFTExecutor", ".sft_executor")
+LoRASFTExecutor = _import_executor("LoRASFTExecutor", ".lora_sft_executor")
+ImageClassificationTrainingExecutor = _import_executor(
     "ImageClassificationTrainingExecutor", ".image_classification_executor"
 )
-HFTransformersExecutor = _safe_import(
+HFTransformersExecutor = _import_executor(
     "HFTransformersExecutor", ".transformers_executor"
 )
-RAGExecutor = _safe_import("RAGExecutor", ".rag_executor")
-AgentExecutor = _safe_import("AgentExecutor", ".agent_executor")
-EchoExecutor = _safe_import("EchoExecutor", ".echo_executor")
-DataProfilingExecutor = _safe_import(
+RAGExecutor = _import_executor("RAGExecutor", ".rag_executor")
+AgentExecutor = _import_executor("AgentExecutor", ".agent_executor")
+EchoExecutor = _import_executor("EchoExecutor", ".echo_executor")
+DataProfilingExecutor = _import_executor(
     "DataProfilingExecutor", ".data_profiling_executor"
 )
-DataRetrievalExecutor = _safe_import(
+DataRetrievalExecutor = _import_executor(
     "DataRetrievalExecutor", ".data_retrieval_executor"
 )
-DiffusersExecutor = _safe_import("DiffusersExecutor", ".diffusers_executor")
-APIExecutor = _safe_import("APIExecutor", ".api_executor")
-SSHExecutor = _safe_import("SSHExecutor", ".ssh_executor")
-OmniText2ImageExecutor = _safe_import(
+DiffusersExecutor = _import_executor("DiffusersExecutor", ".diffusers_executor")
+APIExecutor = _import_executor("APIExecutor", ".api_executor")
+SSHExecutor = _import_executor("SSHExecutor", ".ssh_executor")
+OmniText2ImageExecutor = _import_executor(
     "OmniText2ImageExecutor", ".omni_text2image_executor"
 )
-OmniText2SpeechExecutor = _safe_import(
+OmniText2SpeechExecutor = _import_executor(
     "OmniText2SpeechExecutor", ".omni_text2speech_executor"
 )
-OmniText2AudioExecutor = _safe_import(
+OmniText2AudioExecutor = _import_executor(
     "OmniText2AudioExecutor", ".omni_text2audio_executor"
 )
-OmniText2GeneralExecutor = _safe_import(
+OmniText2GeneralExecutor = _import_executor(
     "OmniText2GeneralExecutor", ".omni_text2general_executor"
 )
 
-EXECUTOR_REGISTRY: dict[str, type | None] = {
+EXECUTOR_REGISTRY: dict[str, type[Executor] | None] = {
     "vllm": VLLMExecutor,
     "vllm_lora": VLLMLoRAExecutor,
     "ppo": PPOExecutor,