Mega 2.0: live LLM worker in full-pipeline

.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

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`

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_"}

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,
+        ]

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:

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"})

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
 

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}/

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"

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:

services/worker-manager/tests/unit/test_build_logic.py

@@ -239,3 +239,49 @@ 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()

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."""