Method _spawn_testing_agent

parallel_orchestrator.py:953–1043

Spawn a testing agent subprocess for batch regression testing. Selects a prioritized batch of passing features using weighted scoring (via _get_test_batch) and passes them as --testing-feature-ids to the subprocess. Falls back to single --testing-feature-id for batches of on

(self)

        return True, f"Started batch [{ids_str}]"

    def _spawn_testing_agent(self) -> tuple[bool, str]:
        """Spawn a testing agent subprocess for batch regression testing.

        Selects a prioritized batch of passing features using weighted scoring
        (via _get_test_batch) and passes them as --testing-feature-ids to the
        subprocess. Falls back to single --testing-feature-id for batches of one.

        Multiple testing agents can test the same feature concurrently - this is
        intentional and simplifies the architecture by removing claim coordination.
        """
        # Check limits first (under lock)
        with self._lock:
            current_testing_count = len(self.running_testing_agents)
            if current_testing_count >= self.max_concurrency:
                debug_log.log("TESTING", f"Skipped spawn - at max testing agents ({current_testing_count}/{self.max_concurrency})")
                return False, f"At max testing agents ({current_testing_count})"
            total_agents = len(self.running_coding_agents) + len(self.running_testing_agents)
            if total_agents >= MAX_TOTAL_AGENTS:
                debug_log.log("TESTING", f"Skipped spawn - at max total agents ({total_agents}/{MAX_TOTAL_AGENTS})")
                return False, f"At max total agents ({total_agents})"

        # Select a weighted batch of passing features for regression testing
        batch = self._get_test_batch(self.testing_batch_size)
        if not batch:
            debug_log.log("TESTING", "No features available for testing")
            return False, "No features available for testing"

        # Use the first feature ID as the representative for logging/tracking
        primary_feature_id = batch[0]
        batch_str = ",".join(str(fid) for fid in batch)
        debug_log.log("TESTING", f"Selected batch for testing: [{batch_str}]")

        # Spawn the testing agent
        with self._lock:
            # Re-check limits in case another thread spawned while we were selecting
            current_testing_count = len(self.running_testing_agents)
            if current_testing_count >= self.max_concurrency:
                return False, f"At max testing agents ({current_testing_count})"

            cmd = [
                sys.executable,
                "-u",
                str(AUTOFORGE_ROOT / "autonomous_agent_demo.py"),
                "--project-dir", str(self.project_dir),
                "--max-iterations", "1",
                "--agent-type", "testing",
                "--testing-feature-ids", batch_str,
            ]
            if self.model:
                cmd.extend(["--model", self.model])

            try:
                # CREATE_NO_WINDOW on Windows prevents console window pop-ups
                # stdin=DEVNULL prevents blocking on stdin reads
                # encoding="utf-8" and errors="replace" fix Windows CP1252 issues
                popen_kwargs: dict[str, Any] = {
                    "stdin": subprocess.DEVNULL,
                    "stdout": subprocess.PIPE,