codegen_orchestrator-618: pipeline/codegen_orchestrator-618

docs/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 2026-07-18
 
+- 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.
 - Extend the live mega pipeline with a real Claude developer worker variant. The noop route still
   proves the deterministic plumbing, while the LLM route now scaffolds a backend-only project,
   runs engineering with a longer timeout, verifies the merged env contract has no user-secret

tests/live/conftest.py

@@ -96,10 +96,12 @@ def redis():
 
 
 @pytest.fixture
-async def test_project(api):
+async def test_project(api, api_internal):
     """Create a manifest-owned project and prove its teardown."""
     data, ctx = await create_test_project_context(api)
-    async with cleanup_guard(lambda: cleanup_all(api, None, ctx), manifest=ctx["manifest"]):
+    async with cleanup_guard(
+        lambda: cleanup_all(api_internal, None, ctx), manifest=ctx["manifest"]
+    ):
         yield data
 
 

tests/live/pipeline_helpers.py

@@ -174,6 +174,7 @@ async def poll_status(
 
 async def create_pipeline_project(
     api: httpx.AsyncClient,
+    api_internal: httpx.AsyncClient,
     *,
     project_prefix: str,
     description: str,
@@ -218,7 +219,7 @@ async def create_pipeline_project(
         "task_description": task_description,
     }
 
-    async with cleanup_on_error(lambda: cleanup_all(api, None, ctx)):
+    async with cleanup_on_error(lambda: cleanup_all(api_internal, None, ctx)):
         manifest.write(ORCHESTRATOR_ROOT / ".live-manifests" / f"{project_id}.json")
         resp = await api.post(
             "/api/repositories/",
@@ -234,10 +235,11 @@ async def create_pipeline_project(
     return ctx
 
 
-async def create_noop_project(api: httpx.AsyncClient) -> dict:
+async def create_noop_project(api: httpx.AsyncClient, api_internal: httpx.AsyncClient) -> dict:
     """Create project + repository for noop pipeline testing. Returns ctx dict."""
     return await create_pipeline_project(
         api,
+        api_internal,
         project_prefix="live-test",
         description=NOOP_PROJECT_DESCRIPTION,
         agent_type="noop",
@@ -246,10 +248,13 @@ async def create_noop_project(api: httpx.AsyncClient) -> dict:
     )
 
 
-async def create_llm_backend_project(api: httpx.AsyncClient) -> dict:
+async def create_llm_backend_project(
+    api: httpx.AsyncClient, api_internal: httpx.AsyncClient
+) -> dict:
     """Create project + repository for the live LLM backend pipeline."""
     return await create_pipeline_project(
         api,
+        api_internal,
         project_prefix="live-test-llm",
         description=LLM_BACKEND_PROJECT_DESCRIPTION,
         detailed_spec=LLM_BACKEND_DETAILED_SPEC,
@@ -671,18 +676,18 @@ def require_unscoped_run_observer(api_internal: httpx.AsyncClient) -> None:
 
     list_runs narrows its result to ``Run.user_id == caller`` for every non-admin
     ``X-Telegram-ID`` it sees, and a valid internal key does not lift that
-    narrowing. pr_poller creates deploy runs with no user_id, so a user-scoped
-    client is answered `[]` for them no matter which filter it passes.
+    narrowing. Deploy and QA producers can create runs with no user_id, so a
+    user-scoped client is answered `[]` for them no matter which filter it passes.
 
     On 2026-07-16 that silently cost the mega a 420s wait for the already
     successful deploy `deploy-poll-ea0bed35`, so this is a loud crash rather than
     a blind poll.
     """
     if USER_AUTH_HEADER in api_internal.headers:
         raise RuntimeError(
-            f"deploy runs must be observed without {USER_AUTH_HEADER}: list_runs "
-            "narrows its result to runs the non-admin harness user owns, and "
-            "pr_poller's deploy run has no user_id"
+            f"runs must be observed without {USER_AUTH_HEADER}: list_runs narrows "
+            "its result to runs the non-admin harness user owns, while internal "
+            "deploy and QA runs can have no user_id"
         )
 
 
@@ -952,18 +957,25 @@ def record(message: CapabilityMessage) -> None:
     )
 
 
-async def cancel_owned_runs(api: httpx.AsyncClient, ctx: dict) -> list[str]:
+async def cancel_owned_runs(api_internal: httpx.AsyncClient, ctx: dict) -> list[str]:
     """Cancel every active run owned by this project before resource teardown."""
+    require_unscoped_run_observer(api_internal)
     project_id = ctx.get("project_id")
     if not project_id:
         return []
-    response = await api.get("/api/runs/", params={"project_id": project_id})
+    response = await api_internal.get("/api/runs/", params={"project_id": project_id})
     response.raise_for_status()
     run_ids = [
-        str(run["id"]) for run in response.json() if run.get("status") in _ACTIVE_RUN_STATUSES
+        str(run["id"])
+        for run in response.json()
+        if str(run.get("project_id")) == str(project_id)
+        and run.get("status") in _ACTIVE_RUN_STATUSES
     ]
     for run_id in run_ids:
-        response = await api.patch(f"/api/runs/{run_id}", json={"status": "cancelled"})
+        response = await api_internal.patch(
+            f"/api/runs/{run_id}",
+            json={"status": "cancelled"},
+        )
         response.raise_for_status()
         ctx["manifest"].own("run", run_id)
     if run_ids:
@@ -974,23 +986,28 @@ async def cancel_owned_runs(api: httpx.AsyncClient, ctx: dict) -> list[str]:
 
 
 async def wait_for_owned_runs(
-    api: httpx.AsyncClient,
+    api_internal: httpx.AsyncClient,
     ctx: dict,
     *,
     timeout: float = RUN_CANCELLATION_TIMEOUT,
     poll_interval: float = RUN_CANCELLATION_POLL_INTERVAL,
 ) -> None:
     """Wait until the run records owned by teardown are terminal."""
+    require_unscoped_run_observer(api_internal)
     run_ids = {
         resource.identifier for resource in ctx["manifest"].resources if resource.kind == "run"
     }
     if not run_ids:
         return
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:
-        response = await api.get("/api/runs/", params={"project_id": ctx["project_id"]})
+        response = await api_internal.get("/api/runs/", params={"project_id": ctx["project_id"]})
         response.raise_for_status()
-        statuses = {str(run["id"]): run.get("status") for run in response.json()}
+        statuses = {
+            str(run["id"]): run.get("status")
+            for run in response.json()
+            if str(run.get("project_id")) == str(ctx["project_id"])
+        }
         if all(statuses.get(run_id) in _TERMINAL_RUN_STATUSES for run_id in run_ids):
             return
         await asyncio.sleep(poll_interval)
@@ -1431,19 +1448,20 @@ def cleanup_server_container(ctx: dict) -> None:
 
 
 async def cleanup_all(
-    api: httpx.AsyncClient,
+    api_internal: httpx.AsyncClient,
     api_no_auth: httpx.AsyncClient | None,
     ctx: dict,
 ) -> None:
-    """Delete only resources owned by this run and prove absence."""
+    """Delete owned resources using an unscoped internal run observer."""
     errors: list[str] = []
 
     # XDEL cannot cancel a claimed message. Fence the consumer and wait for any
     # active scaffold job before deleting or verifying external resources.
     try:
+        require_unscoped_run_observer(api_internal)
         cancel_owned_scaffold(ctx)
-        await cancel_owned_runs(api, ctx)
-        await wait_for_owned_runs(api, ctx)
+        await cancel_owned_runs(api_internal, ctx)
+        await wait_for_owned_runs(api_internal, ctx)
         cancel_owned_active_work(ctx)
         cleanup_owned_capability_work(ctx)
     except Exception as exc:
@@ -1539,7 +1557,7 @@ async def cleanup_all(
             errors.append(f"Redis entry {resource.identifier} still exists or cannot be verified")
 
     if "project_id" in ctx:
-        verify = await api.get(f"/api/projects/{ctx['project_id']}")
+        verify = await api_internal.get(f"/api/projects/{ctx['project_id']}")
         if verify.status_code != 404:
             errors.append(f"project {ctx['project_id']} still exists")
     if errors:

tests/live/test_full_pipeline.py

@@ -70,9 +70,9 @@ async def _pipeline_run(create_project, *, engineering_timeout: int, debug_prefi
                 base_url=API_URL, timeout=10, headers=internal_headers()
             ) as api_internal,
         ):
-            ctx = await create_project(api)
+            ctx = await create_project(api, api_internal)
             async with cleanup_guard(
-                lambda: cleanup_all(api, api_no_auth, ctx), manifest=ctx["manifest"]
+                lambda: cleanup_all(api_internal, api_no_auth, ctx), manifest=ctx["manifest"]
             ):
                 # Phase 1: Scaffold
                 trigger_scaffold(ctx)

tests/live/test_harness_contract.py

@@ -700,7 +700,7 @@ async def handler(request: httpx.Request) -> httpx.Response:
     transport = httpx.MockTransport(handler)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as api:
         with pytest.raises(AssertionError, match="Create repository failed"):
-            await pipeline_helpers.create_noop_project(api)
+            await pipeline_helpers.create_noop_project(api, api)
 
     assert len(cleanup_contexts) == 1
     manifest = cleanup_contexts[0]["manifest"]
@@ -724,7 +724,7 @@ async def handler(request: httpx.Request) -> httpx.Response:
     monkeypatch.setattr(pipeline_helpers, "ORCHESTRATOR_ROOT", tmp_path)
     transport = httpx.MockTransport(handler)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as api:
-        ctx = await pipeline_helpers.create_llm_backend_project(api)
+        ctx = await pipeline_helpers.create_llm_backend_project(api, api)
 
     project_payload = requests[0][1]
     config = project_payload["config"]
@@ -798,7 +798,7 @@ async def handler(request: httpx.Request) -> httpx.Response:
     monkeypatch.setattr(live_conftest, "cleanup_all", cleanup)
     transport = httpx.MockTransport(handler)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as api:
-        fixture = live_conftest.test_project.__wrapped__(api)
+        fixture = live_conftest.test_project.__wrapped__(api, api)
         assert await anext(fixture) == {"id": "common-project"}
         with pytest.raises(StopAsyncIteration):
             await anext(fixture)
@@ -1015,7 +1015,14 @@ async def handler(request: httpx.Request) -> httpx.Response:
             events.append("list-runs")
             return httpx.Response(
                 200,
-                json=[{"id": "deploy-1", "status": "running", "type": "deploy"}],
+                json=[
+                    {
+                        "id": "deploy-1",
+                        "project_id": "project-1",
+                        "status": "running",
+                        "type": "deploy",
+                    }
+                ],
             )
         if request.method == "PATCH" and request.url.path == "/api/runs/deploy-1":
             events.append("cancel-run")
@@ -1371,6 +1378,118 @@ def _runs_client(runs: list[dict], *, as_user: bool = False) -> httpx.AsyncClien
     )
 
 
+@pytest.mark.asyncio
+async def test_cleanup_cancels_unowned_project_runs_through_internal_client(monkeypatch, tmp_path):
+    """Cleanup must see unowned deploy/QA runs without trusting API narrowing."""
+    monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key")
+    monkeypatch.setattr(pipeline_helpers, "ORCHESTRATOR_ROOT", tmp_path)
+    runs = [
+        {
+            "id": "deploy-owned",
+            "type": "deploy",
+            "project_id": "project-1",
+            "story_id": "story-1",
+            "user_id": None,
+            "status": "running",
+        },
+        {
+            "id": "qa-owned",
+            "type": "qa",
+            "project_id": "project-1",
+            "story_id": "story-1",
+            "user_id": None,
+            "status": "queued",
+        },
+        {
+            "id": "deploy-foreign",
+            "type": "deploy",
+            "project_id": "project-2",
+            "story_id": "story-2",
+            "user_id": HARNESS_USER_ID,
+            "status": "running",
+        },
+    ]
+    cancelled = []
+    cancellation_requested = set()
+    internal_run_headers = []
+    external_teardown = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "GET" and request.url.path == "/api/runs/":
+            # Deliberately ignore project_id: the harness must enforce ownership
+            # even if an API implementation returns more than it was asked for.
+            selected = list(runs)
+            if pipeline_helpers.USER_AUTH_HEADER in request.headers:
+                selected = [run for run in selected if run["user_id"] == HARNESS_USER_ID]
+            else:
+                internal_run_headers.append(request.headers)
+                for run in selected:
+                    if run["id"] in cancellation_requested:
+                        run["status"] = "cancelled"
+            return httpx.Response(200, json=selected)
+        if request.method == "PATCH" and request.url.path.startswith("/api/runs/"):
+            internal_run_headers.append(request.headers)
+            run_id = request.url.path.rsplit("/", 1)[-1]
+            cancelled.append(run_id)
+            cancellation_requested.add(run_id)
+            return httpx.Response(200, json=next(run for run in runs if run["id"] == run_id))
+        if request.method == "GET" and request.url.path == "/api/projects/project-1":
+            return httpx.Response(404)
+        raise AssertionError(f"unexpected request: {request.method} {request.url}")
+
+    monkeypatch.setattr(pipeline_helpers, "cancel_owned_scaffold", lambda ctx: None)
+    monkeypatch.setattr(pipeline_helpers, "cancel_owned_active_work", lambda ctx: None)
+    monkeypatch.setattr(pipeline_helpers, "cleanup_owned_capability_work", lambda ctx: None)
+
+    def cleanup_server(ctx):
+        owned_statuses = {run["status"] for run in runs if run["project_id"] == ctx["project_id"]}
+        assert owned_statuses == {"cancelled"}
+        external_teardown.append("server")
+
+    monkeypatch.setattr(pipeline_helpers, "cleanup_server_container", cleanup_server)
+    monkeypatch.setattr(pipeline_helpers, "cleanup_owned_workers", lambda ctx, errors: None)
+    monkeypatch.setattr(pipeline_helpers, "cleanup_registry_resources", lambda ctx, errors: None)
+    monkeypatch.setattr(pipeline_helpers, "_cleanup_db", lambda project_id: None)
+
+    manifest = OwnershipManifest("project-1")
+    manifest.own("project", "project-1")
+    transport = httpx.MockTransport(handler)
+    async with (
+        httpx.AsyncClient(
+            base_url="http://test",
+            transport=transport,
+            headers={**pipeline_helpers.internal_headers(), **pipeline_helpers.AUTH_HEADERS},
+        ) as api_as_user,
+        httpx.AsyncClient(
+            base_url="http://test",
+            transport=transport,
+            headers=pipeline_helpers.internal_headers(),
+        ) as api_internal,
+    ):
+        blind = await api_as_user.get("/api/runs/", params={"project_id": "project-1"})
+        assert [run["id"] for run in blind.json()] == ["deploy-foreign"]
+
+        with pytest.raises(CleanupError, match=pipeline_helpers.USER_AUTH_HEADER):
+            await pipeline_helpers.cleanup_all(
+                api_as_user,
+                None,
+                {"project_id": "project-1", "manifest": manifest},
+            )
+        assert cancelled == []
+        assert external_teardown == []
+
+        await pipeline_helpers.cleanup_all(
+            api_internal,
+            None,
+            {"project_id": "project-1", "manifest": manifest},
+        )
+
+    assert cancelled == ["deploy-owned", "qa-owned"]
+    assert external_teardown == ["server"]
+    assert all(headers["X-Internal-Key"] == "test-internal-key" for headers in internal_run_headers)
+    assert all(pipeline_helpers.USER_AUTH_HEADER not in headers for headers in internal_run_headers)
+
+
 @pytest.mark.asyncio
 async def test_wait_deploy_run_selects_the_run_carrying_the_merged_sha(monkeypatch):
     """Only the pr_poller run records head_sha; the engineering-triggered deploy

tests/live/test_pipeline_engineering.py

@@ -18,6 +18,7 @@
     create_story_and_task,
     dump_debug,
     ensure_test_user,
+    internal_headers,
     trigger_scaffold,
     wait_engineering,
     wait_scaffold,
@@ -35,10 +36,15 @@
 @pytest_asyncio.fixture(loop_scope="module", scope="module")
 async def engineering_ctx():
     """Engineering pipeline: scaffold → story/task → noop worker → CI → done."""
-    async with httpx.AsyncClient(base_url=API_URL, timeout=10, headers=AUTH_HEADERS) as api:
+    async with (
+        httpx.AsyncClient(base_url=API_URL, timeout=10, headers=AUTH_HEADERS) as api,
+        httpx.AsyncClient(base_url=API_URL, timeout=10, headers=internal_headers()) as api_internal,
+    ):
         await ensure_test_user(api)
-        ctx = await create_noop_project(api)
-        async with cleanup_guard(lambda: cleanup_all(api, None, ctx), manifest=ctx["manifest"]):
+        ctx = await create_noop_project(api, api_internal)
+        async with cleanup_guard(
+            lambda: cleanup_all(api_internal, None, ctx), manifest=ctx["manifest"]
+        ):
             # Phase 1: Scaffold
             trigger_scaffold(ctx)
             await wait_scaffold(api, ctx, timeout=SCAFFOLD_TIMEOUT)