diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index a6fab009..431d00cc 100644 --- a/docs/CHANGELOG.md +++ b/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 diff --git a/tests/live/conftest.py b/tests/live/conftest.py index ad7689c0..79cd3cb6 100644 --- a/tests/live/conftest.py +++ b/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 diff --git a/tests/live/pipeline_helpers.py b/tests/live/pipeline_helpers.py index 900839d5..16b01bd3 100644 --- a/tests/live/pipeline_helpers.py +++ b/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,8 +676,8 @@ 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 @@ -680,9 +685,9 @@ def require_unscoped_run_observer(api_internal: httpx.AsyncClient) -> None: """ 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,13 +986,14 @@ 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" } @@ -988,9 +1001,13 @@ async def wait_for_owned_runs( 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: diff --git a/tests/live/test_full_pipeline.py b/tests/live/test_full_pipeline.py index d3e0f8f3..ec2867ce 100644 --- a/tests/live/test_full_pipeline.py +++ b/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) diff --git a/tests/live/test_harness_contract.py b/tests/live/test_harness_contract.py index 13daf81f..0dfe2afb 100644 --- a/tests/live/test_harness_contract.py +++ b/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 diff --git a/tests/live/test_pipeline_engineering.py b/tests/live/test_pipeline_engineering.py index 20ae5e18..11955f43 100644 --- a/tests/live/test_pipeline_engineering.py +++ b/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) diff --git a/tests/live/test_pipeline_scaffold.py b/tests/live/test_pipeline_scaffold.py index 80cb3547..55600cf3 100644 --- a/tests/live/test_pipeline_scaffold.py +++ b/tests/live/test_pipeline_scaffold.py @@ -16,6 +16,7 @@ docker_exec, dump_debug, ensure_test_user, + internal_headers, trigger_scaffold, wait_scaffold, ) @@ -30,10 +31,15 @@ @pytest_asyncio.fixture(loop_scope="module", scope="module") async def scaffold_ctx(): """Scaffold pipeline: create project + repo, trigger scaffold, wait.""" - 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"] + ): trigger_scaffold(ctx) await wait_scaffold(api, ctx, timeout=SCAFFOLD_TIMEOUT) diff --git a/tests/live/test_scaffold_result.py b/tests/live/test_scaffold_result.py index 26b0fb23..7c7b191b 100644 --- a/tests/live/test_scaffold_result.py +++ b/tests/live/test_scaffold_result.py @@ -33,7 +33,7 @@ @pytest.fixture -async def scaffolded_project(api, compose_exec): +async def scaffolded_project(api, api_internal, compose_exec): """Create project + repo, scaffold it, yield (project, repo_name), cleanup.""" suffix = secrets.token_hex(4) project_name = f"live-test-{suffix}" @@ -61,7 +61,7 @@ async def scaffolded_project(api, compose_exec): } # 2. Create repository record - 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/", @@ -74,7 +74,9 @@ async def scaffolded_project(api, compose_exec): assert resp.status_code == 201, f"Create repository failed: {resp.text}" repo_id = resp.json()["id"] - 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"] + ): # 3. Publish ScaffoldMessage to scaffold:queue via redis manifest.own("github_repository", f"{GITHUB_ORG}/{repo_name}") manifest.write(ORCHESTRATOR_ROOT / ".live-manifests" / f"{project_id}.json")