feat(self-healing): API-key preflight + cost_preflight claim-guard + launchd KeepAlive

claude_swarm/conductors/__init__.py

@@ -12,12 +12,15 @@
 from __future__ import annotations
 
 from .api import ApiConductor
+from .env import MissingAnthropicApiKeyError, require_anthropic_api_key
 from .factory import DEFAULT_CONDUCTOR, build_conductor
 from .sdk import SDKConductor
 
 __all__ = [
-    "ApiConductor",
     "DEFAULT_CONDUCTOR",
+    "ApiConductor",
+    "MissingAnthropicApiKeyError",
     "SDKConductor",
     "build_conductor",
+    "require_anthropic_api_key",
 ]

claude_swarm/conductors/api.py

@@ -27,6 +27,7 @@
 from ..heads import Head
 from ..kanban import Task, TaskStatus
 from ..supervisor import DispatchResult
+from .env import MissingAnthropicApiKeyError, require_anthropic_api_key
 
 log = logging.getLogger(__name__)
 
@@ -118,6 +119,10 @@ class ApiConductor:
 
     def dispatch(self, *, head: Head, task: Task) -> DispatchResult:
         """Run *head* against *task* via the Anthropic Messages API (synchronous)."""
+        preflight = self.preflight(head=head, task=task)
+        if preflight is not None:
+            return preflight
+
         import anthropic  # lazy — optional dep; mypy override in pyproject.toml
 
         client = anthropic.Anthropic()
@@ -292,6 +297,25 @@ def dispatch(self, *, head: Head, task: Task) -> DispatchResult:
             cost_usd=running_cost,
         )
 
+    def preflight(self, *, head: Head, task: Task) -> DispatchResult | None:
+        """Return a failure result when the API conductor is not ready.
+
+        Supervisors call this before claiming a task; direct ``dispatch()``
+        callers get the same fail-fast behavior before any client is built.
+        """
+
+        del head, task
+        try:
+            require_anthropic_api_key()
+        except MissingAnthropicApiKeyError as exc:
+            log.error("api-dispatch refused before client construction: %s", exc)
+            return DispatchResult(
+                status=TaskStatus.FAILED,
+                error=str(exc),
+                cost_usd=0.0,
+            )
+        return None
+
     def _tools(self) -> list[dict[str, Any]]:
         """Return the tool-spec list for the API call.
 

claude_swarm/conductors/env.py

@@ -0,0 +1,33 @@
+"""Environment preflights for conductor backends."""
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+
+class MissingAnthropicApiKeyError(RuntimeError):
+    """Raised when an API-backed conductor cannot run without credentials."""
+
+
+def require_anthropic_api_key(*, environ: Mapping[str, str] | None = None) -> None:
+    """Fail fast unless the API key is present and non-blank.
+
+    The key value is intentionally not returned so callers cannot accidentally
+    log, persist, or pass it through surfaces that only need readiness proof.
+    """
+
+    env = os.environ if environ is None else environ
+    value = env.get("ANTHROPIC_API_KEY")
+    if value is None:
+        raise MissingAnthropicApiKeyError(
+            "ANTHROPIC_API_KEY is not set; conductor='api' requires a key in the "
+            "process environment before any task is claimed."
+        )
+    if not value.strip():
+        raise MissingAnthropicApiKeyError(
+            "ANTHROPIC_API_KEY is blank; conductor='api' requires a non-empty key "
+            "in the process environment before any task is claimed."
+        )
+
+
+__all__ = ["MissingAnthropicApiKeyError", "require_anthropic_api_key"]

claude_swarm/conductors/factory.py

@@ -22,6 +22,7 @@
 from ..conductor import ClaudeCLIConductor
 from ..supervisor import Conductor, StubConductor
 from .api import ApiConductor
+from .env import MissingAnthropicApiKeyError, require_anthropic_api_key
 from .sdk import SDKConductor
 
 log = logging.getLogger(__name__)
@@ -65,15 +66,16 @@ def build_conductor(
         For an unrecognised *name* (shouldn't happen when called from a
         ``click.Choice``-validated CLI option, but guards programmatic use).
     """
-    import os
-
     if name == "stub":
         return StubConductor(demo_delay_s=demo_delay_s)
     if name == "claude":
+        import os
+
         if not os.environ.get("CLAUDE_SWARM_ALLOW_CLI_CONDUCTOR"):
             warnings.warn(_CLI_DEPRECATION_MSG, DeprecationWarning, stacklevel=2)
         return ClaudeCLIConductor(model_override=model_override)
     if name == "api":
+        require_anthropic_api_key()
         return ApiConductor(model_override=model_override)
     if name == "sdk":
         return SDKConductor(model_override=model_override)
@@ -82,4 +84,9 @@ def build_conductor(
     )
 
 
-__all__ = ["DEFAULT_CONDUCTOR", "build_conductor"]
+__all__ = [
+    "DEFAULT_CONDUCTOR",
+    "MissingAnthropicApiKeyError",
+    "build_conductor",
+    "require_anthropic_api_key",
+]

claude_swarm/perpetual.py

@@ -122,7 +122,20 @@ def _drive_claimed_task(*, supervisor: Supervisor, **_extra: Any) -> Task | None
     ready = kb.unblocked(limit=1)
     if not ready:
         return None
-    required_head = ready[0].required_head
+    task = ready[0]
+    required_head = task.required_head
+    head = supervisor._pick_head(task)
+    if head is None:
+        log.warning("no head matches required=%r for task %s", task.required_head, task.id)
+        kb.update(
+            task.id,
+            status=TaskStatus.FAILED,
+            error="no matching head",
+            completed_at=time.time(),
+        )
+        return None
+    if not supervisor._preflight_before_claim(head=head, task=task):
+        return None
     claimed = kb.claim_one(
         worker_id=f"{teammate}:{required_head}",
         required_head=required_head,

claude_swarm/supervisor.py

@@ -21,14 +21,15 @@
 import logging
 import time
 from collections.abc import Callable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Any, Protocol
 
 from .abort import AbortMarker, AbortRequested
 from .heads import Head, default_roster
 from .kanban import Kanban, Task, TaskStatus
 from .messaging import MessageBus
+from .meta_supervisor import cost_preflight
 from .reviewer_checkpoint import ReviewerCheckpoint
 
 log = logging.getLogger(__name__)
@@ -126,6 +127,56 @@ def _pick_head(self, task: Task) -> Head | None:
             return h
         return self.roster.get("builder")
 
+    def _preflight_before_claim(self, *, head: Head, task: Task) -> bool:
+        """Return True when *task* may be claimed."""
+
+        preflight_task = task
+        if "cost_cap_usd" not in task.metadata:
+            preflight_task = replace(
+                task,
+                metadata={**task.metadata, "cost_cap_usd": self.config.cost_cap_usd},
+            )
+        verdict = cost_preflight(preflight_task, head_name=head.name)
+        if verdict.verdict == "REJECT":
+            log.warning(
+                "cost preflight rejected task=%s head=%s reason=%s",
+                task.id,
+                head.name,
+                verdict.reason,
+            )
+            self.kanban.update(
+                task.id,
+                status=TaskStatus.FAILED,
+                error=f"cost_preflight_rejected: {verdict.reason}",
+                completed_at=time.time(),
+                reason="cost_preflight:reject",
+            )
+            return False
+        if verdict.verdict == "HOLD":
+            log.warning(
+                "cost preflight hold admitted task=%s head=%s reason=%s",
+                task.id,
+                head.name,
+                verdict.reason,
+            )
+
+        conductor_preflight = getattr(self.conductor, "preflight", None)
+        if callable(conductor_preflight):
+            outcome = conductor_preflight(head=head, task=task)
+            if outcome is not None:
+                self.kanban.update(
+                    task.id,
+                    status=outcome.status,
+                    cost_usd=outcome.cost_usd,
+                    result=outcome.result,
+                    error=outcome.error,
+                    pr_path=outcome.pr_path,
+                    completed_at=time.time(),
+                    reason="conductor_preflight",
+                )
+                return False
+        return True
+
     def step(self) -> Task | None:
         """Run a single supervisor iteration. Returns the dispatched task."""
         if self._abort is not None:
@@ -138,6 +189,8 @@ def step(self) -> Task | None:
         if head is None:
             log.warning("no head matches required=%r for task %s", task.required_head, task.id)
             return None
+        if not self._preflight_before_claim(head=head, task=task):
+            return None
         claimed = self.kanban.claim_one(
             worker_id=f"{self.config.teammate_name}:{head.name}",
             required_head=task.required_head,
@@ -244,6 +297,8 @@ def _run_parallel(self, *, on_idle: Callable[[], None] | None = None) -> None:
                         # Mark failed so we don't loop forever
                         self.kanban.update(task.id, status=TaskStatus.FAILED, error="no matching head")
                         continue
+                    if not self._preflight_before_claim(head=head, task=task):
+                        continue
                     claimed = self.kanban.claim_one(
                         worker_id=f"{self.config.teammate_name}:{head.name}",
                         required_head=task.required_head,

docs/launchd/README.md

@@ -0,0 +1,33 @@
+# Launchd KeepAlive Template
+
+`com.kushal.claude-swarm-perpetual.plist.template` is a source template only.
+It is not an installed LaunchAgent.
+
+Fill the placeholders before any approved install:
+
+- `__CLAUDE_SWARM_BIN__`: absolute path to the `claude-swarm` executable.
+- `__CLAUDE_SWARM_HOME__`: approved swarm state directory.
+- `__AGENT_SWARM_REPO__`: absolute path to the repo checkout to run from.
+- `__LOG_DIR__`: approved writable log directory.
+
+Do not put provider secrets in the plist. The API conductor now fails fast unless
+`ANTHROPIC_API_KEY` is present in the process environment, but the key must come
+from an operator-approved secret path outside the committed template, such as an
+approved per-user environment setup or a Keychain-backed wrapper.
+
+Example live-enablement shape, held until explicit Operator Review approval:
+
+```sh
+launchctl setenv ANTHROPIC_API_KEY '<approved-secret-from-Keychain-or-operator>'
+```
+
+Source-only checks that are safe before installation:
+
+```sh
+plutil -lint docs/launchd/com.kushal.claude-swarm-perpetual.plist.template
+python -m pytest tests/test_launchd_template.py -q
+```
+
+Installing, loading, starting, stopping, or inspecting live launchd state remains
+a runtime/process action and needs explicit approval with target path, overwrite
+policy, secret scope, evidence target, stop plan, and rollback plan.

docs/launchd/com.kushal.claude-swarm-perpetual.plist.template

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
+  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+  <key>Label</key>
+  <string>com.kushal.claude-swarm-perpetual</string>
+
+  <key>ProgramArguments</key>
+  <array>
+    <string>__CLAUDE_SWARM_BIN__</string>
+    <string>perpetual</string>
+    <string>--home</string>
+    <string>__CLAUDE_SWARM_HOME__</string>
+    <string>--count</string>
+    <string>1</string>
+    <string>--conductor</string>
+    <string>api</string>
+  </array>
+
+  <key>WorkingDirectory</key>
+  <string>__AGENT_SWARM_REPO__</string>
+
+  <key>EnvironmentVariables</key>
+  <dict>
+    <key>CLAUDE_SWARM_HOME</key>
+    <string>__CLAUDE_SWARM_HOME__</string>
+    <key>PATH</key>
+    <string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
+  </dict>
+
+  <key>RunAtLoad</key>
+  <true/>
+  <key>KeepAlive</key>
+  <true/>
+
+  <key>StandardOutPath</key>
+  <string>__LOG_DIR__/claude-swarm-perpetual.out</string>
+  <key>StandardErrorPath</key>
+  <string>__LOG_DIR__/claude-swarm-perpetual.err</string>
+</dict>
+</plist>

tests/test_conductor_api.py

@@ -19,6 +19,12 @@
 from claude_swarm.heads import Builder
 from claude_swarm.kanban import Task, TaskStatus
 
+
+@pytest.fixture(autouse=True)
+def fake_anthropic_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "test-only-not-a-real-key")
+
+
 # ---------------------------------------------------------------------------
 # Helpers — fake Anthropic response objects
 # ---------------------------------------------------------------------------
@@ -333,6 +339,23 @@ def test_cache_creation_tokens_split_breakdown_preferred(self) -> None:
 
 
 class TestApiConductorErrorPropagation:
+    def test_missing_api_key_fails_before_client_construction(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        fake_anthropic = _make_fake_anthropic([_make_response("end_turn", "unused")])
+
+        with patch.dict("sys.modules", {"anthropic": fake_anthropic}):
+            result = ApiConductor(model_override="claude-haiku-4-5").dispatch(
+                head=Builder(),
+                task=Task(title="t", prompt="x"),
+            )
+
+        assert result.status is TaskStatus.FAILED
+        assert result.error is not None
+        assert "ANTHROPIC_API_KEY" in result.error
+        fake_anthropic.Anthropic.assert_not_called()
+
     def test_api_error_propagates(self) -> None:
         """If messages.create() raises, the exception propagates to the supervisor."""
         fake_anthropic = _make_fake_anthropic(ValueError("API error"))

tests/test_conductor_factory.py

@@ -12,6 +12,11 @@
 from claude_swarm.supervisor import StubConductor
 
 
+@pytest.fixture(autouse=True)
+def fake_anthropic_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "test-only-not-a-real-key")
+
+
 class TestDefaultConductorIsAPIBased:
     """A: DEFAULT_CONDUCTOR must resolve to the Anthropic Messages API backend.
 
@@ -119,6 +124,16 @@ def test_api_model_override_forwarded(self) -> None:
         assert isinstance(cond, ApiConductor)
         assert cond.model_override == "claude-sonnet-4-6"
 
+    def test_api_requires_anthropic_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        with pytest.raises(RuntimeError, match="ANTHROPIC_API_KEY"):
+            build_conductor("api", model_override=None)
+
+    def test_api_rejects_blank_anthropic_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("ANTHROPIC_API_KEY", "  ")
+        with pytest.raises(RuntimeError, match="blank"):
+            build_conductor("api", model_override=None)
+
     def test_sdk_returns_sdk_conductor(self) -> None:
         cond = build_conductor("sdk", model_override=None)
         assert isinstance(cond, SDKConductor)