From d3f11e9c5b8040251c724974035b1ac9a8e89391 Mon Sep 17 00:00:00 2001 From: Aditya Datta Date: Thu, 8 Oct 2026 07:58:43 +0530 Subject: [PATCH 1/2] fix: keep interrupted ingest out of execution slot --- apps/api/src/mutiny_api/app.py | 5 +-- apps/api/src/mutiny_api/ingest.py | 5 ++- .../mutiny_cli/src/mutiny_cli/hosted_sync.py | 4 ++- tests/integration/test_api.py | 12 +++++++ tests/integration/test_api_cli_hosted_sync.py | 36 +++++++++++++++++-- tests/integration/test_api_hosted_ingest.py | 2 +- 6 files changed, 57 insertions(+), 7 deletions(-) diff --git a/apps/api/src/mutiny_api/app.py b/apps/api/src/mutiny_api/app.py index 3544a17..d5aeb44 100644 --- a/apps/api/src/mutiny_api/app.py +++ b/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", ) diff --git a/apps/api/src/mutiny_api/ingest.py b/apps/api/src/mutiny_api/ingest.py index 7984382..3372d66 100644 --- a/apps/api/src/mutiny_api/ingest.py +++ b/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} diff --git a/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py b/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py index f3cde07..c6b9f2e 100644 --- a/packages/mutiny_cli/src/mutiny_cli/hosted_sync.py +++ b/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, diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index 64c1063..2260ca1 100644 --- a/tests/integration/test_api.py +++ b/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 diff --git a/tests/integration/test_api_cli_hosted_sync.py b/tests/integration/test_api_cli_hosted_sync.py index 1c76f81..ca4178c 100644 --- a/tests/integration/test_api_cli_hosted_sync.py +++ b/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, + "max_generations": 1, + "rng_seed": 0, + "use_boundary_seeds": True, + "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 ( diff --git a/tests/integration/test_api_hosted_ingest.py b/tests/integration/test_api_hosted_ingest.py index de250aa..a45a207 100644 --- a/tests/integration/test_api_hosted_ingest.py +++ b/tests/integration/test_api_hosted_ingest.py @@ -407,7 +407,7 @@ 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"] From cceb2a152683c6124913e06ea9703f8e83b14c15 Mon Sep 17 00:00:00 2001 From: Aditya Datta Date: Sun, 11 Oct 2026 19:30:03 +0530 Subject: [PATCH 2/2] fix: keep ingest campaigns observe-only --- apps/api/src/mutiny_api/supervisor.py | 2 ++ tests/integration/test_api_cli_hosted_sync.py | 2 +- tests/integration/test_api_hosted_ingest.py | 23 +++++++++++++++++++ 3 files changed, 26 insertions(+), 1 deletion(-) diff --git a/apps/api/src/mutiny_api/supervisor.py b/apps/api/src/mutiny_api/supervisor.py index a089a09..b2cfa8c 100644 --- a/apps/api/src/mutiny_api/supervisor.py +++ b/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( diff --git a/tests/integration/test_api_cli_hosted_sync.py b/tests/integration/test_api_cli_hosted_sync.py index ca4178c..2b411b3 100644 --- a/tests/integration/test_api_cli_hosted_sync.py +++ b/tests/integration/test_api_cli_hosted_sync.py @@ -112,9 +112,9 @@ def test_cli_built_payloads_ingest_end_to_end(client: TestClient, tmp_path: Path headers=_auth(), json={ "population_size": 2, + "elite_count": 1, "max_generations": 1, "rng_seed": 0, - "use_boundary_seeds": True, "target": "in_process_demo", }, ) diff --git a/tests/integration/test_api_hosted_ingest.py b/tests/integration/test_api_hosted_ingest.py index a45a207..10fde53 100644 --- a/tests/integration/test_api_hosted_ingest.py +++ b/tests/integration/test_api_hosted_ingest.py @@ -415,6 +415,29 @@ def test_10_first_campaign_ingestion(client: TestClient) -> None: 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