codegen_orchestrator-549: pipeline/codegen_orchestrator-549

docs/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 2026-07-18
 
+- Make live harness API reads fail loud before body parsing. Parsed `tests/live` API responses now
+  call `raise_for_status()` first, including scaffold/project polling and auth-gated server,
+  ssh-key and port-allocation checks, with a mock-transport regression for a rejected polling call.
 - Route live teardown run discovery through an internal-only API client. Cleanup now sees unowned
   deploy and QA runs, selects only records whose `project_id` matches the teardown manifest,
   cancels active records, and proves their terminal status before deleting external resources.

tests/live/conftest.py

@@ -30,7 +30,7 @@ async def api():
     """Async httpx client with test user auth header."""
     headers = {"X-Telegram-ID": str(TEST_TELEGRAM_ID)}
     async with httpx.AsyncClient(base_url=API_URL, timeout=10, headers=headers) as client:
-        await client.post(
+        resp = await client.post(
             "/api/users/upsert",
             json={
                 "telegram_id": TEST_TELEGRAM_ID,
@@ -39,6 +39,7 @@ async def api():
                 "last_name": "Test",
             },
         )
+        resp.raise_for_status()
         yield client
 
 
@@ -117,6 +118,7 @@ async def create_test_project_context(api):
             "config": {"description": "live test project"},
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201, resp.text
     data = resp.json()
     manifest = OwnershipManifest(project_id)

tests/live/pipeline_helpers.py

@@ -139,7 +139,7 @@ def docker_exec(service: str, script: str, timeout: int = 30) -> subprocess.Comp
 
 async def ensure_test_user(api: httpx.AsyncClient) -> None:
     """Create test user if not exists."""
-    await api.post(
+    resp = await api.post(
         "/api/users/upsert",
         json={
             "telegram_id": TEST_TELEGRAM_ID,
@@ -148,6 +148,7 @@ async def ensure_test_user(api: httpx.AsyncClient) -> None:
             "last_name": "Test",
         },
     )
+    resp.raise_for_status()
 
 
 async def poll_status(
@@ -161,8 +162,7 @@ async def poll_status(
     for _ in range(timeout // 3):
         await asyncio.sleep(3)
         resp = await api.get(endpoint)
-        if resp.status_code != 200:
-            continue
+        resp.raise_for_status()
         status = resp.json().get("status")
         if status in target_statuses:
             return status
@@ -204,6 +204,7 @@ async def create_pipeline_project(
             "config": config,
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201, f"Create project failed: {resp.text}"
 
     manifest = OwnershipManifest(run_id=project_id)
@@ -229,6 +230,7 @@ async def create_pipeline_project(
                 "git_url": f"https://github.com/{GITHUB_ORG}/{project_name}",
             },
         )
+        resp.raise_for_status()
         assert resp.status_code == 201, f"Create repository failed: {resp.text}"
         ctx["repo_id"] = resp.json()["id"]
 
@@ -329,13 +331,15 @@ async def create_story_and_task(api: httpx.AsyncClient, ctx: dict) -> None:
             "type": "technical",
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201, f"Create story failed: {resp.text}"
     ctx["story_id"] = resp.json()["id"]
 
     resp = await api.post(
         f"/api/stories/{ctx['story_id']}/start",
         json={"actor": "live-test"},
     )
+    resp.raise_for_status()
     assert resp.status_code == 200, f"Story start failed: {resp.text}"
 
     resp = await api.post(
@@ -349,6 +353,7 @@ async def create_story_and_task(api: httpx.AsyncClient, ctx: dict) -> None:
             "status": TaskStatus.BACKLOG,
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201, f"Create task failed: {resp.text}"
     ctx["task_id"] = resp.json()["id"]
 
@@ -357,6 +362,7 @@ async def create_story_and_task(api: httpx.AsyncClient, ctx: dict) -> None:
         params={"to_status": TaskStatus.TODO},
         json={"actor": "live-test"},
     )
+    resp.raise_for_status()
     assert resp.status_code == 200, f"Task transition to todo failed: {resp.text}"
 
 
@@ -371,6 +377,7 @@ async def wait_engineering(
         await asyncio.sleep(5)
         elapsed += 5
         resp = await api.get(f"/api/tasks/{ctx['task_id']}")
+        resp.raise_for_status()
         status = resp.json().get("status")
         if status in done_statuses:
             break
@@ -384,20 +391,20 @@ async def wait_engineering(
         for _ in range(20):  # up to 60s
             await asyncio.sleep(3)
             resp = await api.get(f"/api/stories/{ctx['story_id']}")
-            if resp.status_code == 200:
-                story_status = resp.json().get("status")
-                ctx["story_status"] = story_status
-                if story_status in {
-                    StoryStatus.PR_REVIEW,
-                    StoryStatus.DEPLOYING,
-                    StoryStatus.COMPLETED,
-                    StoryStatus.FAILED,
-                }:
-                    break
+            resp.raise_for_status()
+            story_status = resp.json().get("status")
+            ctx["story_status"] = story_status
+            if story_status in {
+                StoryStatus.PR_REVIEW,
+                StoryStatus.DEPLOYING,
+                StoryStatus.COMPLETED,
+                StoryStatus.FAILED,
+            }:
+                break
     elif "story_id" in ctx:
         resp = await api.get(f"/api/stories/{ctx['story_id']}")
-        if resp.status_code == 200:
-            ctx["story_status"] = resp.json().get("status")
+        resp.raise_for_status()
+        ctx["story_status"] = resp.json().get("status")
 
 
 async def poll_field(
@@ -412,8 +419,7 @@ async def poll_field(
     for _ in range(timeout // 3):
         await asyncio.sleep(3)
         resp = await api.get(endpoint)
-        if resp.status_code != 200:
-            continue
+        resp.raise_for_status()
         value = resp.json().get(field)
         if value in target_values:
             return value
@@ -440,8 +446,10 @@ async def wait_deploy(
     deadline = asyncio.get_event_loop().time() + timeout
     while asyncio.get_event_loop().time() < deadline:
         repos_resp = await api.get("/api/repositories/", params={"project_id": ctx["project_id"]})
+        repos_resp.raise_for_status()
         for repo in repos_resp.json():
             apps_resp = await api.get("/api/applications/", params={"repo_id": repo["id"]})
+            apps_resp.raise_for_status()
             for app in apps_resp.json():
                 if app["status"] in {s.value for s in terminal}:
                     app_status = app["status"]
@@ -457,15 +465,15 @@ async def wait_deploy(
 
     if story_id := ctx.get("story_id"):
         tasks_resp = await api.get("/api/tasks/", params={"story_id": story_id})
-        if tasks_resp.status_code == 200:
-            ctx["ci_failure_evidence"] = [
-                {
-                    "fix_task_id": task["id"],
-                    **task["failure_metadata"]["ci_failure"],
-                }
-                for task in tasks_resp.json()
-                if (task.get("failure_metadata") or {}).get("ci_failure")
-            ]
+        tasks_resp.raise_for_status()
+        ctx["ci_failure_evidence"] = [
+            {
+                "fix_task_id": task["id"],
+                **task["failure_metadata"]["ci_failure"],
+            }
+            for task in tasks_resp.json()
+            if (task.get("failure_metadata") or {}).get("ci_failure")
+        ]
 
     if app_status != ApplicationStatus.RUNNING.value:
         return

tests/live/test_api_crud.py

@@ -20,18 +20,21 @@ async def test_create_and_get_project(api):
         "/api/projects/",
         json={"id": project_id, "name": name, "status": ProjectStatus.DRAFT, "config": {}},
     )
+    resp.raise_for_status()
     assert resp.status_code == 201
     data = resp.json()
     assert data["id"] == project_id
     assert data["name"] == name
 
     # GET
     resp = await api.get(f"/api/projects/{project_id}")
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["name"] == name
 
     # cleanup
-    await api.delete(f"/api/projects/{project_id}")
+    resp = await api.delete(f"/api/projects/{project_id}")
+    resp.raise_for_status()
 
 
 @pytest.mark.asyncio
@@ -45,13 +48,15 @@ async def test_create_story_for_project(api, test_project):
             "description": "Automated live test",
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201
     story = resp.json()
     assert story["status"] == StoryStatus.CREATED
     assert story["project_id"] == test_project["id"]
 
     # verify list
     resp = await api.get(f"/api/stories/?project_id={test_project['id']}")
+    resp.raise_for_status()
     assert resp.status_code == 200
     stories = resp.json()
     assert any(s["id"] == story["id"] for s in stories)
@@ -68,12 +73,14 @@ async def test_create_and_list_tasks(api, test_project):
             "description": "Test task for live tests",
         },
     )
+    task_resp.raise_for_status()
     assert task_resp.status_code == 201
     task = task_resp.json()
     assert task["status"] == TaskStatus.BACKLOG
 
     # list by project
     resp = await api.get(f"/api/tasks/?project_id={test_project['id']}")
+    resp.raise_for_status()
     assert resp.status_code == 200
     tasks = resp.json()
     assert any(t["id"] == task["id"] for t in tasks)
@@ -89,21 +96,25 @@ async def test_task_transitions(api, test_project):
             "title": "Transition test task",
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 201
     task_id = resp.json()["id"]
 
     # backlog → todo
     resp = await api.post(f"/api/tasks/{task_id}/transition?to_status={TaskStatus.TODO}")
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["status"] == TaskStatus.TODO
 
     # todo → in_dev (via start)
     resp = await api.post(f"/api/tasks/{task_id}/start")
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["status"] == TaskStatus.IN_DEV
 
     # in_dev → failed (via fail)
     resp = await api.post(f"/api/tasks/{task_id}/fail")
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["status"] == TaskStatus.FAILED
 
@@ -121,6 +132,7 @@ async def test_upsert_user(api):
             "last_name": "Test",
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 200
     user = resp.json()
     assert user["telegram_id"] == tg_id
@@ -133,5 +145,6 @@ async def test_upsert_user(api):
             "username": "live_test_user_updated",
         },
     )
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["username"] == "live_test_user_updated"

tests/live/test_deploy_infra.py

@@ -35,6 +35,7 @@
 async def managed_server(api_internal):
     """Get first managed server with active-ish status, or skip."""
     resp = await api_internal.get("/api/servers/")
+    resp.raise_for_status()
     assert resp.status_code == 200, f"List servers failed: {resp.text}"
     servers = resp.json()
 
@@ -52,6 +53,7 @@ class TestManagedServer:
     async def test_managed_server_exists(self, api_internal):
         """At least one managed server with operational status in DB."""
         resp = await api_internal.get("/api/servers/")
+        resp.raise_for_status()
         assert resp.status_code == 200
         servers = resp.json()
 
@@ -68,6 +70,7 @@ async def test_server_has_ssh_key(self, api_internal, managed_server):
         """Managed server has a decryptable SSH key (deployer fetches it at step 5)."""
         handle = managed_server["handle"]
         resp = await api_internal.get(f"/api/servers/{handle}/ssh-key")
+        resp.raise_for_status()
         assert resp.status_code == 200, f"SSH key endpoint failed: {resp.text}"
         body = resp.json()
         key = body.get("ssh_key", "")
@@ -95,9 +98,7 @@ def test_server_reachable_via_ssh(self, compose_exec, managed_server):
             "        base_url=api_url, timeout=10, headers=headers\n"
             "    ) as client:\n"
             f"        resp = await client.get('/api/servers/{handle}/ssh-key')\n"
-            "        if resp.status_code != 200:\n"
-            "            print(f'FAILED:ssh-key-fetch:{resp.status_code}')\n"
-            "            return\n"
+            "        resp.raise_for_status()\n"
             "        key = resp.json().get('ssh_key', '')\n"
             "    if not key.endswith('\\n'):\n"
             "        key += '\\n'\n"
@@ -147,6 +148,7 @@ async def test_allocate_and_release_port(self, api_internal, managed_server):
             f"/api/servers/{handle}/ports/allocate-next",
             json={"service_name": "live-test-deploy-infra", "start_port": 19000},
         )
+        resp.raise_for_status()
         assert resp.status_code in (200, 201), f"Port allocation failed: {resp.text}"
         allocation = resp.json()
         alloc_id = allocation["id"]
@@ -156,12 +158,14 @@ async def test_allocate_and_release_port(self, api_internal, managed_server):
         try:
             # Verify it shows up in listings
             resp = await api_internal.get(f"/api/servers/{handle}/ports")
+            resp.raise_for_status()
             assert resp.status_code == 200
             allocated_ports = [p["port"] for p in resp.json()]
             assert port in allocated_ports, f"Port {port} not found in server ports list"
         finally:
             # Always release — this test owns only the allocation it just created.
-            await api_internal.delete(f"/api/allocations/{alloc_id}")
+            resp = await api_internal.delete(f"/api/allocations/{alloc_id}")
+            resp.raise_for_status()
 
 
 class TestDeploySecrets:

tests/live/test_harness_contract.py

@@ -699,7 +699,7 @@ async def handler(request: httpx.Request) -> httpx.Response:
     monkeypatch.setattr(pipeline_helpers, "cleanup_all", cleanup)
     transport = httpx.MockTransport(handler)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as api:
-        with pytest.raises(AssertionError, match="Create repository failed"):
+        with pytest.raises(httpx.HTTPStatusError):
             await pipeline_helpers.create_noop_project(api, api)
 
     assert len(cleanup_contexts) == 1
@@ -768,6 +768,22 @@ async def handler(request: httpx.Request) -> httpx.Response:
     assert ctx["task_id"] == "task-1"
 
 
+@pytest.mark.asyncio
+async def test_poll_status_fails_loudly_before_parsing_error_bodies():
+    async def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(403, json={"detail": "internal key required"})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as api:
+        with pytest.raises(httpx.HTTPStatusError):
+            await pipeline_helpers.poll_status(
+                api,
+                "/api/projects/project-1",
+                {"active"},
+                timeout=3,
+            )
+
+
 @pytest.mark.asyncio
 async def test_common_live_project_gets_persisted_manifest(monkeypatch, tmp_path):
     async def handler(request: httpx.Request) -> httpx.Response:

tests/live/test_health.py

@@ -9,6 +9,7 @@
 async def test_api_health(api_no_auth):
     """API /health returns 200."""
     resp = await api_no_auth.get("/health")
+    resp.raise_for_status()
     assert resp.status_code == 200
     assert resp.json()["status"] == "ok"