fix: keep interrupted ingest out of execution slot

apps/api/src/mutiny_api/app.py

@@ -329,9 +329,10 @@ def health() -> HealthResponse:
         llm_ok = bool(cfg.configured)
         model = cfg.model if llm_ok else f"unconfigured:{DEFAULT_MUTATION_MODEL}"
         mutator_mode = "featherless" if llm_ok else "template"
+        max_concurrent_campaigns = 1
         running = len(repo.list_running_campaigns())
         status = "ok" if db_ok else "degraded"
-        if running > 1:
+        if running >= max_concurrent_campaigns:
             status = "degraded"
         return HealthResponse(
             status=status,
@@ -343,7 +344,7 @@ def health() -> HealthResponse:
             llm_configured=llm_ok,
             db_latency_ms=latency_ms,
             schema_version=SCHEMA_VERSION,
-            max_concurrent_campaigns=1,
+            max_concurrent_campaigns=max_concurrent_campaigns,
             running_campaigns=running,
             adapter_loading="trusted_demo_only",
         )

apps/api/src/mutiny_api/ingest.py

@@ -184,7 +184,10 @@ def open_campaign(self, body: IngestCampaignOpenRequest) -> dict[str, Any]:
             body.campaign_id,
             stored_config,
             project_id=project_id,
-            status=body.status,
+            # Ingest campaigns are observe-only. Normalize older CLI clients
+            # that still send ``running`` so an interrupted upload cannot
+            # occupy the supervisor's single execution slot indefinitely.
+            status="created",
         )
         return {"campaign": camp, "created": True}
 

apps/api/src/mutiny_api/supervisor.py

@@ -294,6 +294,8 @@ def start_campaign(
             raise RuntimeError(
                 f"cannot start campaign in status={campaign['status']}"
             )
+        if campaign["config"].get("execution_mode") == "local_cli":
+            raise RuntimeError("cannot start observe-only ingest campaign")
         running = self.repo.list_running_campaigns()
         if running:
             raise RuntimeError(

packages/mutiny_cli/src/mutiny_cli/hosted_sync.py

@@ -373,7 +373,9 @@ def build_campaign_open_payload(bundle: LocalRunBundle) -> dict[str, Any]:
         "project_name": bundle.project_root.name,
         "project_label": bundle.project_root.name,
         "adapter": adapter,
-        "status": "running",
+        # Observe-only ingest must not consume the Hosted execution slot if the
+        # upload is interrupted before the terminal completion request.
+        "status": "created",
         "attestation": True,
         "started_at": bundle.started_at or _utcnow(),
         "config": cfg,

tests/integration/test_api.py

@@ -73,6 +73,18 @@ def test_start_rejects_invalid_persisted_elite_config(client):
     assert client.app.state.repo.get_campaign("invalid-elites")["status"] == "created"
 
 
+def test_health_degrades_when_execution_slot_is_full(client: TestClient):
+    client.app.state.repo.create_campaign(
+        "health-running-campaign", {}, status="running"
+    )
+
+    body = client.get("/api/health").json()
+
+    assert body["max_concurrent_campaigns"] == 1
+    assert body["running_campaigns"] == 1
+    assert body["status"] == "degraded"
+
+
 def test_health_queries_sqlite_only_under_repository_lock(client: TestClient):
     """The health endpoint must not bypass the repository mutex."""
     repo = client.app.state.repo

tests/integration/test_api_cli_hosted_sync.py

@@ -91,10 +91,40 @@ def test_cli_built_payloads_ingest_end_to_end(client: TestClient, tmp_path: Path
     complete_body = build_complete_payload(bundle)
 
     assert open_body["execution_mode"] == "local_cli"
+    assert open_body["status"] == "created"
     assert "project_path" not in open_body
 
-    r1 = client.post("/api/ingest/v1/campaigns", headers=_auth(), json=open_body)
+    # Legacy clients sent ``running``. Hosted must normalize that value so an
+    # interrupted observe-only upload cannot block the real execution slot.
+    legacy_open_body = {**open_body, "status": "running"}
+    r1 = client.post(
+        "/api/ingest/v1/campaigns", headers=_auth(), json=legacy_open_body
+    )
     assert r1.status_code in (200, 201), r1.text
+    assert r1.json()["campaign"]["status"] == "created"
+
+    health = client.get("/api/health", headers=_auth())
+    assert health.status_code == 200
+    assert health.json()["running_campaigns"] == 0
+
+    demo = client.post(
+        "/api/campaigns",
+        headers=_auth(),
+        json={
+            "population_size": 2,
+            "elite_count": 1,
+            "max_generations": 1,
+            "rng_seed": 0,
+            "target": "in_process_demo",
+        },
+    )
+    assert demo.status_code == 201, demo.text
+    started = client.post(
+        f"/api/campaigns/{demo.json()['id']}/start",
+        headers=_auth(),
+        json={"attestation": True},
+    )
+    assert started.status_code == 200, started.text
 
     r2 = client.post(
         f"/api/ingest/v1/campaigns/{bundle.campaign_id}/batch",
@@ -123,7 +153,9 @@ def test_cli_built_payloads_ingest_end_to_end(client: TestClient, tmp_path: Path
 
     # Idempotent retry with identical CLI payloads.
     assert (
-        client.post("/api/ingest/v1/campaigns", headers=_auth(), json=open_body).status_code
+        client.post(
+            "/api/ingest/v1/campaigns", headers=_auth(), json=legacy_open_body
+        ).status_code
         in (200, 201)
     )
     assert (

tests/integration/test_api_hosted_ingest.py

@@ -407,14 +407,37 @@ def test_10_first_campaign_ingestion(client: TestClient) -> None:
     assert r.status_code == 201
     camp = r.json()["campaign"]
     assert camp["id"] == cid
-    assert camp["status"] == "running"
+    assert camp["status"] == "created"
     assert camp["config"]["execution_mode"] == "local_cli"
     assert camp["config"]["local_project_key"] == "unique-key-10"
     assert camp["project_id"]
     # Opaque path prefix — not a real FS mount.
     assert camp["project"]["path"].startswith(INGEST_LOCAL_KEY_PREFIX)
 
 
+def test_ingest_campaign_cannot_be_started(client: TestClient) -> None:
+    response, cid, _ = _open_campaign(client)
+    assert response.status_code == 201
+    assert response.json()["campaign"]["status"] == "created"
+
+    started = client.post(
+        f"/api/campaigns/{cid}/start",
+        headers=_auth(),
+        json={"attestation": True},
+    )
+    assert started.status_code == 409
+    assert started.json()["error"]["code"] == "campaign_conflict"
+
+    campaign = client.get(f"/api/campaigns/{cid}", headers=_auth())
+    assert campaign.status_code == 200
+    assert campaign.json()["status"] == "created"
+
+    health = client.get("/api/health", headers=_auth())
+    assert health.status_code == 200
+    assert health.json()["status"] == "ok"
+    assert health.json()["running_campaigns"] == 0
+
+
 def test_11_duplicate_campaign_ingestion(client: TestClient) -> None:
     r1, cid, payload = _open_campaign(client, local_project_key="dup-key")
     assert r1.status_code == 201