From 87626520a9554a61d0937ef4599a2e446ebfc7bf Mon Sep 17 00:00:00 2001 From: vladmesh <16962535+vladmesh@users.noreply.github.com> Date: Sat, 18 Jul 2026 02:50:40 +0300 Subject: [PATCH 1/4] Extend mega pipeline real-worker coverage --- docs/CHANGELOG.md | 9 ++ .../src/worker_wrapper/runners/factory.py | 13 +- .../src/worker_wrapper/wrapper.py | 3 +- .../tests/unit/test_agent_env.py | 20 +++ .../worker-wrapper/tests/unit/test_runners.py | 3 + .../worker-manager/src/container_config.py | 8 +- services/worker-manager/src/manager.py | 5 + .../tests/unit/test_build_logic.py | 48 ++++++ .../tests/unit/test_container_config.py | 23 +++ tests/live/pipeline_helpers.py | 106 +++++++++++-- tests/live/test_full_pipeline.py | 139 +++++++++++++++--- tests/live/test_harness_contract.py | 57 +++++++ tests/live/test_worker_cleanup.py | 23 +++ 13 files changed, 418 insertions(+), 39 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0f5ddd17..a6fab009 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## 2026-07-18 + +- 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 + entries, then gates on deploy success, `/health` 200, and non-LLM QA passed. Real LLM worker + containers now get a 4GB memory limit; noop workers keep the smaller 2GB limit. Factory worker + startup was also hardened to forward its API key and request Droid's non-interactive edit mode. + ## 2026-07-16 - Give post-deploy QA one source of truth for acceptance criteria. `Repository.acceptance_criteria` diff --git a/packages/worker-wrapper/src/worker_wrapper/runners/factory.py b/packages/worker-wrapper/src/worker_wrapper/runners/factory.py index 89adf874..7bf6221f 100644 --- a/packages/worker-wrapper/src/worker_wrapper/runners/factory.py +++ b/packages/worker-wrapper/src/worker_wrapper/runners/factory.py @@ -8,6 +8,13 @@ class FactoryRunner(AgentRunner): """Runner for Factory Droid agent.""" def build_command(self, prompt: str) -> list[str]: - # factory.ai is the CLI tool name, command is droid exec - # Based on specs: droid exec -o json "prompt" - return ["droid", "exec", "-o", "json", prompt] + return [ + "droid", + "exec", + "--skip-permissions-unsafe", + "--cwd", + "/workspace", + "-o", + "json", + prompt, + ] diff --git a/packages/worker-wrapper/src/worker_wrapper/wrapper.py b/packages/worker-wrapper/src/worker_wrapper/wrapper.py index 0dfc7732..d8be83db 100644 --- a/packages/worker-wrapper/src/worker_wrapper/wrapper.py +++ b/packages/worker-wrapper/src/worker_wrapper/wrapper.py @@ -661,7 +661,8 @@ async def execute_agent(self, data: dict[str, Any]) -> None: logger.error( "agent_process_failed", stderr=stderr, stdout=stdout, exit_code=proc.returncode ) - raise RuntimeError(f"Agent process failed with code {proc.returncode}: {stderr}") + detail = stderr or stdout or "no output" + raise RuntimeError(f"Agent process failed with code {proc.returncode}: {detail}") # Capture session_id from Claude CLI JSON output if self.config.agent_type == AgentType.CLAUDE and not session_id: diff --git a/packages/worker-wrapper/tests/unit/test_agent_env.py b/packages/worker-wrapper/tests/unit/test_agent_env.py index 03b02e91..f84b72cf 100644 --- a/packages/worker-wrapper/tests/unit/test_agent_env.py +++ b/packages/worker-wrapper/tests/unit/test_agent_env.py @@ -139,3 +139,23 @@ async def test_subprocess_removes_pythonpath_when_only_app(self): env = captured["env"] assert "PYTHONPATH" not in env or env["PYTHONPATH"] == "" + + @pytest.mark.asyncio + async def test_failure_error_uses_stdout_when_stderr_empty(self): + """Some CLIs emit structured failures on stdout, not stderr.""" + wrapper = _make_wrapper() + + async def fake_exec(*args, **kwargs): + proc = AsyncMock() + proc.communicate = AsyncMock(return_value=(b'{"result":"auth failed"}', b"")) + proc.returncode = 1 + proc.kill = AsyncMock() + proc.wait = AsyncMock() + return proc + + with ( + patch("worker_wrapper.wrapper.asyncio.create_subprocess_exec", side_effect=fake_exec), + patch.object(wrapper, "_resolve_prompt", return_value="do stuff"), + ): + with pytest.raises(RuntimeError, match="auth failed"): + await wrapper.execute_agent({"prompt": "test"}) diff --git a/packages/worker-wrapper/tests/unit/test_runners.py b/packages/worker-wrapper/tests/unit/test_runners.py index 47cfba9b..dcf62818 100644 --- a/packages/worker-wrapper/tests/unit/test_runners.py +++ b/packages/worker-wrapper/tests/unit/test_runners.py @@ -49,6 +49,9 @@ def test_build_command_uses_droid_exec(self): assert "exec" in cmd # Check flags + assert "--skip-permissions-unsafe" in cmd + assert "--cwd" in cmd + assert "/workspace" in cmd assert "-o" in cmd assert "json" in cmd diff --git a/services/worker-manager/src/container_config.py b/services/worker-manager/src/container_config.py index 75ece16c..290cda07 100644 --- a/services/worker-manager/src/container_config.py +++ b/services/worker-manager/src/container_config.py @@ -88,7 +88,7 @@ def to_docker_run_kwargs(self, network_name: Optional[str] = None) -> Dict[str, "name": f"worker-{self.worker_id}", "hostname": f"worker-{self.worker_id}", # Standard limits - "mem_limit": "2g", + "mem_limit": self._mem_limit(), "cpu_period": 100000, "cpu_quota": 100000, # 1 CPU } @@ -101,3 +101,9 @@ def to_docker_run_kwargs(self, network_name: Optional[str] = None) -> Dict[str, kwargs["network_mode"] = "host" return kwargs + + def _mem_limit(self) -> str: + """Return the Docker memory limit for this worker agent.""" + if self.agent_type in {AgentType.CLAUDE, AgentType.FACTORY}: + return "4g" + return "2g" diff --git a/services/worker-manager/src/manager.py b/services/worker-manager/src/manager.py index 329c1acf..47da70f8 100644 --- a/services/worker-manager/src/manager.py +++ b/services/worker-manager/src/manager.py @@ -427,6 +427,11 @@ async def create_worker_with_capabilities( worker_manager_url=settings.WORKER_MANAGER_URL, ) container_env.update(env_vars) + if agent_type == AgentType.FACTORY and "FACTORY_API_KEY" not in container_env: + factory_api_key = os.getenv("FACTORY_API_KEY") + if not factory_api_key: + raise RuntimeError("FACTORY_API_KEY is not set") + container_env["FACTORY_API_KEY"] = factory_api_key github_token = env_vars.get("GITHUB_TOKEN") if github_token: diff --git a/services/worker-manager/tests/unit/test_build_logic.py b/services/worker-manager/tests/unit/test_build_logic.py index ce2b34d2..c749c4c5 100644 --- a/services/worker-manager/tests/unit/test_build_logic.py +++ b/services/worker-manager/tests/unit/test_build_logic.py @@ -239,3 +239,51 @@ async def test_create_worker_with_capabilities(self, mock_workspace, mock_redis, call_kwargs = mock_docker.run_container.call_args[1] expected_hash = compute_image_hash(["GIT", "CURL"]) assert expected_hash in call_kwargs["image"] + + @pytest.mark.asyncio + @patch("src.manager.workspace_mod") + async def test_factory_worker_forwards_manager_api_key( + self, mock_workspace, mock_redis, mock_docker, monkeypatch + ): + """Factory child containers need FACTORY_API_KEY even in host-session mode.""" + from pathlib import Path + + monkeypatch.setenv("FACTORY_API_KEY", "fk-test") + mock_workspace.get_scaffolded_workspace.return_value = (Path("/data/workspaces/repo-1"), True) + manager = WorkerManager(redis=mock_redis, docker_client=mock_docker) + + await manager.create_worker_with_capabilities( + worker_id="factory-worker-1", + capabilities=["GIT"], + base_image="worker-base:latest", + agent_type="factory", + repo_id="repo-1", + env_vars={"GITHUB_TOKEN": "tok", "REPO_NAME": "org/repo"}, + ) + + call_kwargs = mock_docker.run_container.call_args[1] + assert call_kwargs["environment"]["FACTORY_API_KEY"] == "fk-test" + + @pytest.mark.asyncio + @patch("src.manager.workspace_mod") + async def test_factory_worker_fails_fast_without_api_key( + self, mock_workspace, mock_redis, mock_docker, monkeypatch + ): + """Missing Factory credentials should fail before launching a dead worker.""" + from pathlib import Path + + monkeypatch.delenv("FACTORY_API_KEY", raising=False) + mock_workspace.get_scaffolded_workspace.return_value = (Path("/data/workspaces/repo-1"), True) + manager = WorkerManager(redis=mock_redis, docker_client=mock_docker) + + with pytest.raises(RuntimeError, match="FACTORY_API_KEY is not set"): + await manager.create_worker_with_capabilities( + worker_id="factory-worker-1", + capabilities=["GIT"], + base_image="worker-base:latest", + agent_type="factory", + repo_id="repo-1", + env_vars={"GITHUB_TOKEN": "tok", "REPO_NAME": "org/repo"}, + ) + + mock_docker.run_container.assert_not_awaited() diff --git a/services/worker-manager/tests/unit/test_container_config.py b/services/worker-manager/tests/unit/test_container_config.py index 53f300a8..941dc9d3 100644 --- a/services/worker-manager/tests/unit/test_container_config.py +++ b/services/worker-manager/tests/unit/test_container_config.py @@ -59,6 +59,29 @@ def test_to_docker_run_kwargs_defaults_to_host_network(self): kwargs = config.to_docker_run_kwargs() assert kwargs["network_mode"] == "host" assert "network" not in kwargs + assert kwargs["mem_limit"] == "4g" + + def test_noop_worker_keeps_lower_memory_limit(self): + """Noop workers do not need the real-agent memory budget.""" + config = WorkerContainerConfig( + worker_id="test-1", + worker_type="developer", + agent_type="noop", + capabilities=[], + ) + kwargs = config.to_docker_run_kwargs() + assert kwargs["mem_limit"] == "2g" + + def test_factory_worker_gets_real_agent_memory_limit(self): + """Factory workers need the same memory budget as other real LLM agents.""" + config = WorkerContainerConfig( + worker_id="test-1", + worker_type="developer", + agent_type="factory", + capabilities=[], + ) + kwargs = config.to_docker_run_kwargs() + assert kwargs["mem_limit"] == "4g" def test_to_docker_run_kwargs_with_network_name(self): """With network_name, should attach to that network.""" diff --git a/tests/live/pipeline_helpers.py b/tests/live/pipeline_helpers.py index ae629859..b30047bc 100644 --- a/tests/live/pipeline_helpers.py +++ b/tests/live/pipeline_helpers.py @@ -43,6 +43,7 @@ # Timeouts (seconds) SCAFFOLD_TIMEOUT = 120 ENGINEERING_TIMEOUT = 420 # 7 min (worker spawn + noop + CI) +LLM_ENGINEERING_TIMEOUT = 1800 # 30 min (worker spawn + LLM edits + CI-fix loop) DEPLOY_TIMEOUT = 420 # 7 min (deploy.yml + smoke test) # Deploy allocates a port per module: the web module plus these infra ports # (mirror of deploy._DEPLOY_INFRA_PORT_SERVICES). Only the web module serves @@ -81,6 +82,32 @@ # its payload. ENV_CONTRACT_PROBE_MARKER = "ENV_CONTRACT_PROBE:" +NOOP_PROJECT_DESCRIPTION = "Pipeline E2E test - noop" +NOOP_TASK_TITLE = "Noop implementation task" +NOOP_TASK_DESCRIPTION = "Empty commit via NoopRunner - pipeline test" + +LLM_BACKEND_PROJECT_DESCRIPTION = ( + "Backend-only live LLM pipeline test. Build a minimal HTTP API that can deploy " + "without any user-provided secrets." +) +LLM_BACKEND_DETAILED_SPEC = """Implement a backend-only service. + +Requirements: +- Keep the project backend-only. Do not add frontend, Telegram, notification, or bot modules. +- Do not require any user-provided secrets or environment variables. +- Keep the existing deploy contract limited to generated, computed, or literal values. +- Ensure GET /health returns HTTP 200 with a small JSON payload. +- Add or update a focused backend test for the health endpoint if the scaffold does not already + cover it. +- Run the repository's normal formatting, linting, and unit tests before committing. +""" +LLM_BACKEND_TASK_TITLE = "Implement backend health API" +LLM_BACKEND_TASK_DESCRIPTION = ( + "Use the scaffolded backend service and make the smallest code change needed to implement " + "a production-safe GET /health endpoint that returns HTTP 200 JSON. The app must deploy " + "with backend-only modules and no user-required secrets." +) + # ── Low-level helpers ──────────────────────────────────────────────────── @@ -145,11 +172,27 @@ async def poll_status( # ── Pipeline phase helpers ─────────────────────────────────────────────── -async def create_noop_project(api: httpx.AsyncClient) -> dict: - """Create project + repository for noop pipeline testing. Returns ctx dict.""" +async def create_pipeline_project( + api: httpx.AsyncClient, + *, + project_prefix: str, + description: str, + agent_type: str, + task_title: str, + task_description: str, + detailed_spec: str | None = None, +) -> dict: + """Create project + repository for one live pipeline variant. Returns ctx dict.""" suffix = secrets.token_hex(4) - project_name = f"live-test-{suffix}" + project_name = f"{project_prefix}-{suffix}" project_id = str(uuid.uuid4()) + config = { + "description": description, + "modules": ["backend"], + "agent_type": agent_type, + } + if detailed_spec: + config["detailed_spec"] = detailed_spec resp = await api.post( "/api/projects/", @@ -157,11 +200,7 @@ async def create_noop_project(api: httpx.AsyncClient) -> dict: "id": project_id, "name": project_name, "status": ProjectStatus.DRAFT, - "config": { - "description": "Pipeline E2E test — noop", - "modules": ["backend"], - "agent_type": "noop", - }, + "config": config, }, ) assert resp.status_code == 201, f"Create project failed: {resp.text}" @@ -173,6 +212,10 @@ async def create_noop_project(api: httpx.AsyncClient) -> dict: "project_name": project_name, "repo_name": project_name, "manifest": manifest, + "agent_type": agent_type, + "scaffold_task_description": task_description, + "task_title": task_title, + "task_description": task_description, } async with cleanup_on_error(lambda: cleanup_all(api, None, ctx)): @@ -191,6 +234,31 @@ async def create_noop_project(api: httpx.AsyncClient) -> dict: return ctx +async def create_noop_project(api: httpx.AsyncClient) -> dict: + """Create project + repository for noop pipeline testing. Returns ctx dict.""" + return await create_pipeline_project( + api, + project_prefix="live-test", + description=NOOP_PROJECT_DESCRIPTION, + agent_type="noop", + task_title=NOOP_TASK_TITLE, + task_description=NOOP_TASK_DESCRIPTION, + ) + + +async def create_llm_backend_project(api: httpx.AsyncClient) -> dict: + """Create project + repository for the live LLM backend pipeline.""" + return await create_pipeline_project( + api, + project_prefix="live-test-llm", + description=LLM_BACKEND_PROJECT_DESCRIPTION, + detailed_spec=LLM_BACKEND_DETAILED_SPEC, + agent_type="claude", + task_title=LLM_BACKEND_TASK_TITLE, + task_description=LLM_BACKEND_TASK_DESCRIPTION, + ) + + def trigger_scaffold(ctx: dict) -> None: """Publish scaffold message to Redis stream.""" ctx["manifest"].own("github_repository", f"{GITHUB_ORG}/{ctx['repo_name']}") @@ -204,7 +272,7 @@ def trigger_scaffold(ctx: dict) -> None: "template_ref": TEMPLATE_REF, "project_name": ctx["project_name"], "modules": "backend", - "task_description": "Pipeline E2E test project", + "task_description": ctx.get("scaffold_task_description", "Pipeline E2E test project"), } result = subprocess.run( [ @@ -271,8 +339,8 @@ async def create_story_and_task(api: httpx.AsyncClient, ctx: dict) -> None: "project_id": ctx["project_id"], "story_id": ctx["story_id"], "type": "create", - "title": "Noop implementation task", - "description": "Empty commit via NoopRunner — pipeline test", + "title": ctx.get("task_title", NOOP_TASK_TITLE), + "description": ctx.get("task_description", NOOP_TASK_DESCRIPTION), "status": TaskStatus.BACKLOG, }, ) @@ -483,8 +551,14 @@ def build_env_contract_probe_script( " if content is None:\n" " raise RuntimeError(f'contract fragment disappeared: {path}')\n" " fragments.append(yaml.safe_load(content))\n" - " entries = (\n" - " sorted(merge_env_contract_fragments(fragments).entries) if fragments else []\n" + " contract = merge_env_contract_fragments(fragments) if fragments else None\n" + " entries = sorted(contract.entries) if contract else []\n" + " user_secret_entries = (\n" + " sorted(\n" + " key for key, entry in contract.entries.items()\n" + " if getattr(entry, 'source', None) == 'user_secret'\n" + " )\n" + " if contract else []\n" " )\n" " merged_into_main = None\n" " if verify_merged:\n" @@ -498,7 +572,8 @@ def build_env_contract_probe_script( " resp.raise_for_status()\n" " merged_into_main = resp.json()['status'] in ('identical', 'behind')\n" " payload = {'ref': ref, 'fragment_paths': fragment_paths,\n" - " 'entries': entries, 'merged_into_main': merged_into_main}\n" + " 'entries': entries, 'user_secret_entries': user_secret_entries,\n" + " 'merged_into_main': merged_into_main}\n" f" print({ENV_CONTRACT_PROBE_MARKER!r} + json.dumps(payload))\n" "asyncio.run(probe())\n" ) @@ -1090,7 +1165,8 @@ def _wait_for_container_absence( cwd=ORCHESTRATOR_ROOT, ) if verify.returncode != 0: - if "No such container" in verify.stderr: + inspect_error = f"{verify.stderr}\n{verify.stdout}".lower() + if any(marker in inspect_error for marker in ("no such container", "no such object")): return None return "Docker inspect failed" if time.monotonic() >= deadline: diff --git a/tests/live/test_full_pipeline.py b/tests/live/test_full_pipeline.py index 904506d5..67923262 100644 --- a/tests/live/test_full_pipeline.py +++ b/tests/live/test_full_pipeline.py @@ -1,17 +1,18 @@ -"""Pipeline test: Full deploy (~7-10 min) — THE MEGA TEST. +"""Pipeline test: Full deploy - THE MEGA TEST. Exercises the entire path from project creation to a live /health response: - 1. API: create project (agent_type=noop) + repo + 1. API: create project + repo 2. scaffold:queue → scaffolder → GitHub repo (+ branch protection on main) 3. API: create story (in_progress) + task (todo) - 4. task_dispatcher → engineering:queue → noop worker → empty commit + push to story branch + 4. task_dispatcher → engineering:queue → worker → commit + push to story branch 5. All tasks done → dispatcher creates PR story/{id} → main (auto-merge enabled) 6. CI runs on PR → green → auto-merge → webhook → deploy:queue 7. deploy consumer → DevOps subgraph → GitHub Actions deploy.yml 8. smoke test: GET /health → 200 -No LLM. Fully deterministic. Real queues, real GitHub, real server. +The noop path stays deterministic. The LLM path exercises the product route where a +real developer worker changes code before CI, merge, deploy, health, and QA. """ import asyncio @@ -26,9 +27,11 @@ DEPLOY_TIMEOUT, ENGINEERING_TIMEOUT, EXPECTED_ENV_CONTRACT_FRAGMENTS, + LLM_ENGINEERING_TIMEOUT, QA_RUN_TIMEOUT, SCAFFOLD_TIMEOUT, cleanup_all, + create_llm_backend_project, create_noop_project, create_story_and_task, dump_debug, @@ -54,8 +57,7 @@ pytestmark = pytest.mark.asyncio(loop_scope="module") -@pytest_asyncio.fixture(loop_scope="module", scope="module") -async def pipeline(): +async def _pipeline_run(create_project, *, engineering_timeout: int, debug_prefix: str): """Full pipeline: scaffold → engineering → deploy. Yields context for assertions.""" async with httpx.AsyncClient(base_url=API_URL, timeout=10, headers=AUTH_HEADERS) as api: await ensure_test_user(api) @@ -68,7 +70,7 @@ async def pipeline(): base_url=API_URL, timeout=10, headers=internal_headers() ) as api_internal, ): - ctx = await create_noop_project(api) + ctx = await create_project(api) async with cleanup_guard( lambda: cleanup_all(api, api_no_auth, ctx), manifest=ctx["manifest"] ): @@ -77,32 +79,32 @@ async def pipeline(): await wait_scaffold(api, ctx, timeout=SCAFFOLD_TIMEOUT) if ctx.get("scaffold_status") != ProjectStatus.ACTIVE: yield ctx - dump_debug(ctx, "full-scaffold") + dump_debug(ctx, f"{debug_prefix}-scaffold") return # The scaffolded repo must already carry the contract deploy # resolves its environment from. if not record_env_contract(ctx, "main", phase="scaffold"): yield ctx - dump_debug(ctx, "full-env-contract-scaffold") + dump_debug(ctx, f"{debug_prefix}-env-contract-scaffold") return # Phase 2: Engineering await create_story_and_task(api, ctx) - await wait_engineering(api, ctx, timeout=ENGINEERING_TIMEOUT) + await wait_engineering(api, ctx, timeout=engineering_timeout) if ctx.get("task_status") != TaskStatus.DONE: yield ctx - dump_debug(ctx, "full-engineering") + dump_debug(ctx, f"{debug_prefix}-engineering") return # Phase 3: Deploy. The story branch merges into main and only then - # does a deploy run appear carrying the merged head SHA — the ref + # does a deploy run appear carrying the merged head SHA. The ref # deploy reads the contract at. Re-check the contract there: the # scaffolded tree proves nothing about what engineering merged. deploy_run = await wait_deploy_run(api_internal, ctx, timeout=DEPLOY_RUN_TIMEOUT) if deploy_run is None: yield ctx - dump_debug(ctx, "full-deploy-run") + dump_debug(ctx, f"{debug_prefix}-deploy-run") return if not record_env_contract( ctx, @@ -111,7 +113,7 @@ async def pipeline(): verify_merged_into_main=True, ): yield ctx - dump_debug(ctx, "full-env-contract-merged") + dump_debug(ctx, f"{debug_prefix}-env-contract-merged") return await wait_deploy(api, api_no_auth, ctx, timeout=DEPLOY_TIMEOUT) @@ -132,7 +134,29 @@ async def pipeline(): ctx.get("final_app_status") != ApplicationStatus.RUNNING.value or ctx.get("deploy_outcome") != DeployOutcome.SUCCESS.value ): - dump_debug(ctx, "full-deploy") + dump_debug(ctx, f"{debug_prefix}-deploy") + + +@pytest_asyncio.fixture(loop_scope="module", scope="module") +async def pipeline(): + """Full noop pipeline: scaffold → engineering → deploy.""" + async for ctx in _pipeline_run( + create_noop_project, + engineering_timeout=ENGINEERING_TIMEOUT, + debug_prefix="full-noop", + ): + yield ctx + + +@pytest_asyncio.fixture(loop_scope="module", scope="module") +async def llm_pipeline(): + """Full LLM pipeline: scaffold → real worker → deploy.""" + async for ctx in _pipeline_run( + create_llm_backend_project, + engineering_timeout=LLM_ENGINEERING_TIMEOUT, + debug_prefix="full-llm", + ): + yield ctx class TestFullPipeline: @@ -141,13 +165,13 @@ class TestFullPipeline: async def test_project_active(self, pipeline): """Project status should be 'active' after successful scaffold + deploy.""" assert pipeline.get("scaffold_status") == ProjectStatus.ACTIVE, ( - f"Scaffold failed — status: {pipeline.get('scaffold_status')}" + f"Scaffold failed, status: {pipeline.get('scaffold_status')}" ) assert pipeline.get("task_status") == TaskStatus.DONE, ( - f"Engineering failed — task status: {pipeline.get('task_status')}" + f"Engineering failed, task status: {pipeline.get('task_status')}" ) assert pipeline.get("final_app_status") == ApplicationStatus.RUNNING.value, ( - f"Deploy failed — app_status: {pipeline.get('final_app_status')}" + f"Deploy failed, app_status: {pipeline.get('final_app_status')}" ) async def test_env_contract_committed_by_scaffold(self, pipeline): @@ -203,7 +227,7 @@ async def test_health_endpoint(self, pipeline): """GET /health on deployed service returns 200.""" if pipeline.get("final_app_status") != ApplicationStatus.RUNNING.value: pytest.skip("deploy failed") - assert "deployed_url" in pipeline, "No deployed_url — port allocation missing?" + assert "deployed_url" in pipeline, "No deployed_url, port allocation missing?" url = pipeline["deployed_url"] async with httpx.AsyncClient(timeout=30) as client: @@ -235,3 +259,80 @@ async def test_non_llm_qa_passed(self, pipeline): "status": "completed", "qa_outcome": "passed", } + + +class TestFullPipelineLLM: + """THE MEGA TEST with a real developer worker.""" + + async def test_project_active(self, llm_pipeline): + """Project status should be 'active' after successful scaffold + deploy.""" + assert llm_pipeline.get("agent_type") == "claude" + assert llm_pipeline.get("scaffold_status") == ProjectStatus.ACTIVE, ( + f"Scaffold failed, status: {llm_pipeline.get('scaffold_status')}" + ) + assert llm_pipeline.get("task_status") == TaskStatus.DONE, ( + f"Engineering failed, task status: {llm_pipeline.get('task_status')}" + ) + assert llm_pipeline.get("final_app_status") == ApplicationStatus.RUNNING.value, ( + f"Deploy failed, app_status: {llm_pipeline.get('final_app_status')}" + ) + + async def test_no_user_secrets_required(self, llm_pipeline): + """The backend-only LLM project must not trip the user-secret deploy path.""" + if llm_pipeline.get("task_status") != TaskStatus.DONE: + pytest.skip("engineering failed") + errors = llm_pipeline.get("env_contract_errors") or {} + assert "merged" not in errors, errors.get("merged") + probe = llm_pipeline["env_contract_probes"]["merged"] + assert probe["user_secret_entries"] == [] + + async def test_deploy_run_outcome_success(self, llm_pipeline): + """The deploy run this mega triggered must conclude deploy_outcome=success.""" + if llm_pipeline.get("task_status") != TaskStatus.DONE: + pytest.skip("engineering failed") + assert llm_pipeline.get("deploy_run_error") is None, llm_pipeline["deploy_run_error"] + assert llm_pipeline.get("deploy_outcome_error") is None, llm_pipeline[ + "deploy_outcome_error" + ] + assert llm_pipeline.get("deploy_outcome") == DeployOutcome.SUCCESS.value, ( + f"Deploy run {llm_pipeline.get('deploy_run_id')} ended " + f"deploy_outcome={llm_pipeline.get('deploy_outcome')} " + f"({llm_pipeline.get('deploy_error_details')})" + ) + + async def test_health_endpoint(self, llm_pipeline): + """GET /health on deployed service returns 200.""" + if llm_pipeline.get("final_app_status") != ApplicationStatus.RUNNING.value: + pytest.skip("deploy failed") + assert "deployed_url" in llm_pipeline, "No deployed_url, port allocation missing?" + + url = llm_pipeline["deployed_url"] + async with httpx.AsyncClient(timeout=30) as client: + for _attempt in range(5): + try: + resp = await client.get(f"{url}/health") + if resp.status_code == 200: + break + resp = await client.get(f"{url}/v1/health") + if resp.status_code == 200: + break + except httpx.ConnectError: + pass + await asyncio.sleep(5) + else: + pytest.fail(f"Health endpoint not reachable at {url}/health after 5 attempts") + + assert resp.status_code == 200, f"Health check failed: {resp.status_code} {resp.text[:200]}" + + async def test_non_llm_qa_passed(self, llm_pipeline): + """A separate post-deploy QA run must terminate as passed.""" + if ( + llm_pipeline.get("final_app_status") != ApplicationStatus.RUNNING.value + or llm_pipeline.get("deploy_outcome") != DeployOutcome.SUCCESS.value + ): + pytest.skip("deploy failed") + assert llm_pipeline.get("qa_result") == { + "run_id": llm_pipeline["qa_result"]["run_id"], + "status": "completed", + "qa_outcome": "passed", + } diff --git a/tests/live/test_harness_contract.py b/tests/live/test_harness_contract.py index cdf2aa00..13daf81f 100644 --- a/tests/live/test_harness_contract.py +++ b/tests/live/test_harness_contract.py @@ -711,6 +711,63 @@ async def handler(request: httpx.Request) -> httpx.Response: assert written["resources"] == [{"identifier": manifest.run_id, "kind": "project"}] +@pytest.mark.asyncio +async def test_llm_backend_project_uses_real_worker_backend_only_config(monkeypatch, tmp_path): + requests = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.url.path, json.loads(request.content))) + if request.url.path == "/api/projects/": + return httpx.Response(201, json={"id": "project"}) + return httpx.Response(201, json={"id": "repo-1"}) + + 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) + + project_payload = requests[0][1] + config = project_payload["config"] + assert project_payload["name"].startswith("live-test-llm-") + assert config["modules"] == ["backend"] + assert config["agent_type"] == "claude" + assert "user-provided secrets" in config["detailed_spec"] + assert "secrets" not in config + assert "env_hints" not in config + assert ctx["repo_id"] == "repo-1" + assert ctx["task_title"] == pipeline_helpers.LLM_BACKEND_TASK_TITLE + assert ctx["task_description"] == pipeline_helpers.LLM_BACKEND_TASK_DESCRIPTION + + +@pytest.mark.asyncio +async def test_create_story_and_task_uses_context_task_description(): + requests = [] + + async def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) if request.content else {} + requests.append((request.url.path, body)) + if request.url.path == "/api/stories/": + return httpx.Response(201, json={"id": "story-1"}) + if request.url.path == "/api/tasks/": + return httpx.Response(201, json={"id": "task-1"}) + return httpx.Response(200, json={}) + + ctx = { + "project_id": "project-1", + "task_title": "Implement backend health API", + "task_description": "Create a real backend health endpoint.", + } + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as api: + await pipeline_helpers.create_story_and_task(api, ctx) + + task_payload = [body for path, body in requests if path == "/api/tasks/"][0] + assert task_payload["title"] == "Implement backend health API" + assert task_payload["description"] == "Create a real backend health endpoint." + assert ctx["story_id"] == "story-1" + assert ctx["task_id"] == "task-1" + + @pytest.mark.asyncio async def test_common_live_project_gets_persisted_manifest(monkeypatch, tmp_path): async def handler(request: httpx.Request) -> httpx.Response: diff --git a/tests/live/test_worker_cleanup.py b/tests/live/test_worker_cleanup.py index 440daedc..e4f7cc8a 100644 --- a/tests/live/test_worker_cleanup.py +++ b/tests/live/test_worker_cleanup.py @@ -79,6 +79,29 @@ def run(cmd, **kwargs): assert any("redis-cli" in cmd and "DEL" in cmd for cmd in calls) +def test_cleanup_accepts_docker_no_such_object_as_verified_absence(monkeypatch): + def run(cmd, **kwargs): + if cmd[:3] == ["docker", "rm", "-f"]: + return SimpleNamespace(returncode=0, stdout="", stderr="") + if cmd[:2] == ["docker", "inspect"]: + return SimpleNamespace( + returncode=1, + stdout="[]", + stderr="Error: No such object: custom-prefix-abc", + ) + if "EXISTS" in cmd: + return SimpleNamespace(returncode=0, stdout="0\n", stderr="") + return SimpleNamespace(returncode=0, stdout="1\n", stderr="") + + monkeypatch.setattr("pipeline_helpers.subprocess.run", run) + monkeypatch.setattr("pipeline_helpers.capture_owned_workers", lambda ctx: None) + errors = [] + + cleanup_owned_workers(_context(), errors, timeout=0, poll_interval=0) + + assert errors == [] + + def test_cleanup_stays_red_when_container_remains_and_still_cleans_redis(monkeypatch): calls = [] From e0a3ee0b0210f3a9b26931f2590b58ff01db9f25 Mon Sep 17 00:00:00 2001 From: vladmesh <16962535+vladmesh@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:16:53 +0300 Subject: [PATCH 2/4] =?UTF-8?q?fix:=20make=20live=20LLM=20mega=20green=20?= =?UTF-8?q?=E2=80=94=20required-user-secret=20assert=20+=20worker=20timeou?= =?UTF-8?q?t=20default?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live LLM mega passes end to end (scaffold -> live Claude worker -> deploy -> /health -> QA). Two fixes close the last gaps a real worker exposed: - test_no_user_secrets_required asserted user_secret_entries == [], counting optional overrides. The backend template declares DATABASE_URL/ASYNC_DATABASE_URL as source: user_secret, required: false, resolved from the allocated postgres. Only required user secrets dead-end deploy (WAITING_FOR_USER_SECRET), so the contract probe now reports required_user_secret_entries and the test checks that. - Raise the worker subprocess timeout default 300 -> 900 (worker-manager and worker-wrapper) so a live LLM agent process is not killed mid-edit; stays within the harness LLM_ENGINEERING_TIMEOUT. The noop runner keeps its own short timeout. Document COMPOSE_PROJECT_NAME and HOST_CLAUDE_DIR (isolated worker session copy) in .env.example so live tests and the stack agree regardless of checkout dir. --- .env.example | 12 ++++++++++++ .../worker-wrapper/src/worker_wrapper/config.py | 4 +++- services/worker-manager/src/config.py | 6 ++++-- tests/live/pipeline_helpers.py | 9 +++++++++ tests/live/test_full_pipeline.py | 13 +++++++++++-- 5 files changed, 39 insertions(+), 5 deletions(-) diff --git a/.env.example b/.env.example index 88e867aa..e237e9fe 100644 --- a/.env.example +++ b/.env.example @@ -147,3 +147,15 @@ LOKI_PUSH_PASSWORD=change_me_in_production # Bcrypt hash for Caddy basic_auth — same escaping rules as REGISTRY_PASSWORD_HASH # Generate: docker run --rm caddy:2-alpine caddy hash-password --plaintext 'your_password' LOKI_PUSH_PASSWORD_HASH=$$2a$$14$$replace_with_actual_hash + +# ----- Worker agent runtime ----- +# Pin the compose project name so live tests and the running stack agree on which +# containers to target, regardless of the checkout directory name. +COMPOSE_PROJECT_NAME=codegen_orchestrator +# Host path to a Claude Code session dir mounted into spawned worker containers. +# Use an ISOLATED COPY of ~/.claude, not your live session — workers get rw access. +# mkdir -p ~/.claude-worker && cp ~/.claude/.credentials.json ~/.claude-worker/ +# HOST_CLAUDE_DIR=/home/youruser/.claude-worker +# Per-agent-process timeout inside the worker (seconds). Code default is 900, +# which fits live LLM engineering; override here only to tune. +# WORKER_SUBPROCESS_TIMEOUT_SECONDS=900 diff --git a/packages/worker-wrapper/src/worker_wrapper/config.py b/packages/worker-wrapper/src/worker_wrapper/config.py index 3f1396e9..862be279 100644 --- a/packages/worker-wrapper/src/worker_wrapper/config.py +++ b/packages/worker-wrapper/src/worker_wrapper/config.py @@ -16,7 +16,9 @@ class WorkerWrapperConfig(BaseSettings): # Optional execution settings poll_interval_ms: int = 500 - subprocess_timeout_seconds: int = 300 + # Fallback when WORKER_SUBPROCESS_TIMEOUT_SECONDS is unset; worker-manager + # normally forwards its own value. Sized for live LLM agents, not noop. + subprocess_timeout_seconds: int = 900 http_server_port: int = 9090 model_config = {"env_prefix": "WORKER_"} diff --git a/services/worker-manager/src/config.py b/services/worker-manager/src/config.py index cb6dddc5..9cec9bac 100644 --- a/services/worker-manager/src/config.py +++ b/services/worker-manager/src/config.py @@ -24,8 +24,10 @@ class WorkerManagerSettings(BaseSettings): # Host path to .claude directory (for mounting into workers) HOST_CLAUDE_DIR: str | None = None - # Worker subprocess timeout (seconds) - WORKER_SUBPROCESS_TIMEOUT_SECONDS: int = 300 + # Worker subprocess timeout (seconds). Live LLM agents (Claude/Factory) need + # well over the noop budget to write and iterate on real code; keep within the + # harness LLM_ENGINEERING_TIMEOUT. The noop runner uses its own short timeout. + WORKER_SUBPROCESS_TIMEOUT_SECONDS: int = 900 # Path to pre-scaffolded workspaces (created by scaffolder service) # All workspaces live here, keyed by repo_id: /data/workspaces/{repo_id}/ diff --git a/tests/live/pipeline_helpers.py b/tests/live/pipeline_helpers.py index b30047bc..900839d5 100644 --- a/tests/live/pipeline_helpers.py +++ b/tests/live/pipeline_helpers.py @@ -560,6 +560,14 @@ def build_env_contract_probe_script( " )\n" " if contract else []\n" " )\n" + " required_user_secret_entries = (\n" + " sorted(\n" + " key for key, entry in contract.entries.items()\n" + " if getattr(entry, 'source', None) == 'user_secret'\n" + " and getattr(entry, 'required', False)\n" + " )\n" + " if contract else []\n" + " )\n" " merged_into_main = None\n" " if verify_merged:\n" " token = await gh.get_token(owner, repo)\n" @@ -573,6 +581,7 @@ def build_env_contract_probe_script( " merged_into_main = resp.json()['status'] in ('identical', 'behind')\n" " payload = {'ref': ref, 'fragment_paths': fragment_paths,\n" " 'entries': entries, 'user_secret_entries': user_secret_entries,\n" + " 'required_user_secret_entries': required_user_secret_entries,\n" " 'merged_into_main': merged_into_main}\n" f" print({ENV_CONTRACT_PROBE_MARKER!r} + json.dumps(payload))\n" "asyncio.run(probe())\n" diff --git a/tests/live/test_full_pipeline.py b/tests/live/test_full_pipeline.py index 67923262..ecaad8f6 100644 --- a/tests/live/test_full_pipeline.py +++ b/tests/live/test_full_pipeline.py @@ -278,13 +278,22 @@ async def test_project_active(self, llm_pipeline): ) async def test_no_user_secrets_required(self, llm_pipeline): - """The backend-only LLM project must not trip the user-secret deploy path.""" + """The backend-only LLM project must not trip the user-secret deploy path. + + Only *required* user secrets dead-end the deploy (DeployOutcome + WAITING_FOR_USER_SECRET). Optional ``user_secret`` overrides such as the + template's ``DATABASE_URL`` (``required: false``) are resolved from the + allocated infrastructure and must not fail this project. + """ if llm_pipeline.get("task_status") != TaskStatus.DONE: pytest.skip("engineering failed") errors = llm_pipeline.get("env_contract_errors") or {} assert "merged" not in errors, errors.get("merged") probe = llm_pipeline["env_contract_probes"]["merged"] - assert probe["user_secret_entries"] == [] + assert probe["required_user_secret_entries"] == [], ( + "required user secrets would dead-end deploy: " + f"{probe['required_user_secret_entries']}" + ) async def test_deploy_run_outcome_success(self, llm_pipeline): """The deploy run this mega triggered must conclude deploy_outcome=success.""" From 0a4816b399a21dc91da092fa29c69b8d1c1cecc1 Mon Sep 17 00:00:00 2001 From: vladmesh <16962535+vladmesh@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:26:52 +0300 Subject: [PATCH 3/4] style: ruff format test_full_pipeline assertion --- tests/live/test_full_pipeline.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/live/test_full_pipeline.py b/tests/live/test_full_pipeline.py index ecaad8f6..d3e0f8f3 100644 --- a/tests/live/test_full_pipeline.py +++ b/tests/live/test_full_pipeline.py @@ -291,8 +291,7 @@ async def test_no_user_secrets_required(self, llm_pipeline): assert "merged" not in errors, errors.get("merged") probe = llm_pipeline["env_contract_probes"]["merged"] assert probe["required_user_secret_entries"] == [], ( - "required user secrets would dead-end deploy: " - f"{probe['required_user_secret_entries']}" + f"required user secrets would dead-end deploy: {probe['required_user_secret_entries']}" ) async def test_deploy_run_outcome_success(self, llm_pipeline): From e2ccd75cc25bba41d51689c443be2b89cd6969c7 Mon Sep 17 00:00:00 2001 From: vladmesh <16962535+vladmesh@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:28:26 +0300 Subject: [PATCH 4/4] style: ruff format test_build_logic --- services/worker-manager/tests/unit/test_build_logic.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/services/worker-manager/tests/unit/test_build_logic.py b/services/worker-manager/tests/unit/test_build_logic.py index c749c4c5..4d9d0a12 100644 --- a/services/worker-manager/tests/unit/test_build_logic.py +++ b/services/worker-manager/tests/unit/test_build_logic.py @@ -242,9 +242,7 @@ async def test_create_worker_with_capabilities(self, mock_workspace, mock_redis, @pytest.mark.asyncio @patch("src.manager.workspace_mod") - async def test_factory_worker_forwards_manager_api_key( - self, mock_workspace, mock_redis, mock_docker, monkeypatch - ): + async def test_factory_worker_forwards_manager_api_key(self, mock_workspace, mock_redis, mock_docker, monkeypatch): """Factory child containers need FACTORY_API_KEY even in host-session mode.""" from pathlib import Path