diff --git a/factory/outer_loop/__init__.py b/factory/outer_loop/__init__.py index 9dfe12eac..1c9a1ed4b 100644 --- a/factory/outer_loop/__init__.py +++ b/factory/outer_loop/__init__.py @@ -1,6 +1,6 @@ """Outer loop — evolutionary swarm search for workflow optimization.""" -from factory.outer_loop.designer import DesignerAgent, extract_telemetry +from factory.outer_loop.designer import DesignerAgent, extract_telemetry, populate_prompt from factory.outer_loop.engine import BudgetTracker, SwarmEngine from factory.outer_loop.filesystem import ( export_best_workflow, @@ -30,11 +30,13 @@ OuterLoopState, SwarmConfig, ) +from factory.outer_loop.subset import CalibratedSubsetSelector from factory.outer_loop.workflow import outer_loop_workflow __all__ = [ "AuditResult", "BudgetTracker", + "CalibratedSubsetSelector", "DirectFeatureBenchEvaluator", "DesignerAgent", "EvalResult", @@ -55,6 +57,7 @@ "load_checkpoint", "load_config", "outer_loop_workflow", + "populate_prompt", "save_best", "save_checkpoint", "save_generation", diff --git a/factory/outer_loop/designer.py b/factory/outer_loop/designer.py index c31af5794..7481fc8fd 100644 --- a/factory/outer_loop/designer.py +++ b/factory/outer_loop/designer.py @@ -3,8 +3,8 @@ Design mode: creates from-scratch workflow designs (minimal, thorough, custom). Mutation mode: proposes targeted mutations based on failure telemetry. -v1 uses deterministic templates. LLM integration comes when the outer loop -runs against real benchmarks. +Prompts are populated with benchmark-specific context so all designed workflows +produce agents that reference task files, testbed paths, and test files. """ from __future__ import annotations @@ -23,12 +23,61 @@ log = structlog.get_logger() +_ROLE_PROMPT_TEMPLATES: dict[str, str] = { + "researcher": ( + "Study the codebase at {{testbed_path}}. Read the issue at {{task_file}}. " + "Explore the repository structure, identify relevant files, and write " + "findings to .factory/strategy/research.md." + ), + "builder": ( + "Read the task description at {{task_file}}. Read any prior research at " + ".factory/strategy/research.md. Implement the fix in the codebase at " + "{{testbed_path}}. Run pytest on {{test_file}} to verify. Fix any failures. " + "Commit your changes." + ), + "health_checker": ( + "Run the test suite at {{test_file}} in the codebase at {{testbed_path}}. " + "Report results including pass/fail counts." + ), + "code_reviewer": ( + "Review the changes made in {{testbed_path}} for the task described in " + "{{task_file}}. Check for correctness, edge cases, and style." + ), + "adversarial_tester": ( + "Test the implementation in {{testbed_path}} adversarially. Try edge cases " + "and unexpected inputs. Run {{test_file}} and report any failures." + ), + "strategist": ( + "Read the research at .factory/strategy/research.md and the task at " + "{{task_file}}. Formulate a strategy for solving the issue." + ), +} + + +def populate_prompt(role: str, benchmark_spec: str) -> str: + """Generate a role-specific functional prompt with benchmark context. + + The benchmark_spec is used as-is for placeholder values when specific + paths are not known at design time. + """ + template = _ROLE_PROMPT_TEMPLATES.get(role) + if template is None: + return f"Complete the {role} task for the benchmark: {benchmark_spec}" + + return ( + template + .replace("{{testbed_path}}", "/tmp/testbed") + .replace("{{task_file}}", "/tmp/testbed/task-instruction.md") + .replace("{{test_file}}", "the relevant test files") + ) + class DesignerAgent: """LLM-guided workflow designer with design and mutation modes. Design mode produces from-scratch workflows for seed diversity. Mutation mode proposes targeted mutations from failure telemetry. + All designed workflows include benchmark-specific prompts. """ def design_minimal(self, benchmark_spec: str) -> Workflow: @@ -40,12 +89,14 @@ def design_minimal(self, benchmark_spec: str) -> Workflow: "researcher": AgentNode( id="researcher", role=AgentRole.RESEARCHER, + prompt_template=populate_prompt("researcher", benchmark_spec), writes={".factory/strategy/research.md"}, timeout=300, ), "builder": AgentNode( id="builder", role=AgentRole.BUILDER, + prompt_template=populate_prompt("builder", benchmark_spec), reads={".factory/strategy/research.md"}, writes={".factory/reviews/builder-latest.md"}, timeout=600, @@ -87,6 +138,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "researcher": AgentNode( id="researcher", role=AgentRole.RESEARCHER, + prompt_template=populate_prompt("researcher", benchmark_spec), reads={".factory/strategy/observations.md"}, writes={".factory/strategy/research.md"}, timeout=600, @@ -94,6 +146,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "strategist": AgentNode( id="strategist", role=AgentRole.STRATEGIST, + prompt_template=populate_prompt("strategist", benchmark_spec), reads={".factory/strategy/research.md"}, writes={".factory/strategy/current.md"}, timeout=600, @@ -106,6 +159,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "builder_a": AgentNode( id="builder_a", role=AgentRole.BUILDER, + prompt_template=populate_prompt("builder", benchmark_spec), reads={".factory/strategy/current.md"}, writes={".factory/reviews/builder-a.md"}, timeout=1200, @@ -113,6 +167,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "builder_b": AgentNode( id="builder_b", role=AgentRole.BUILDER, + prompt_template=populate_prompt("builder", benchmark_spec), reads={".factory/strategy/current.md"}, writes={".factory/reviews/builder-b.md"}, timeout=1200, @@ -124,6 +179,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "code_reviewer": AgentNode( id="code_reviewer", role=AgentRole.CODE_REVIEWER, + prompt_template=populate_prompt("code_reviewer", benchmark_spec), reads={".factory/reviews/builder-a.md", ".factory/reviews/builder-b.md"}, writes={".factory/reviews/code-review.md"}, timeout=900, @@ -131,6 +187,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: "adversarial_tester": AgentNode( id="adversarial_tester", role=AgentRole.ADVERSARIAL_TESTER, + prompt_template=populate_prompt("adversarial_tester", benchmark_spec), reads={".factory/reviews/code-review.md"}, writes={".factory/reviews/adversarial-qa.md"}, timeout=1800, @@ -201,6 +258,7 @@ def design_custom(self, benchmark_spec: str, constraints: dict[str, object]) -> nodes[node_id] = AgentNode( id=node_id, role=role, + prompt_template=populate_prompt(node_id, benchmark_spec), timeout=600, ) if prev_id is not None: diff --git a/factory/outer_loop/direct_evaluator.py b/factory/outer_loop/direct_evaluator.py index d37b5e243..5764c0c56 100644 --- a/factory/outer_loop/direct_evaluator.py +++ b/factory/outer_loop/direct_evaluator.py @@ -334,10 +334,11 @@ def _verify_in_docker( log.error("no_test_target", task_dir=str(task_dir)) return False - # 1. Create container (kept alive with sleep so we can exec into it) + # 1. Create container with network disabled to prevent answer leakage cid_result = subprocess.run( [ "docker", "create", "--platform", "linux/amd64", + "--network", "none", image, "bash", "-c", "sleep 600", ], diff --git a/factory/outer_loop/engine.py b/factory/outer_loop/engine.py index aac2494ff..741a4fc63 100644 --- a/factory/outer_loop/engine.py +++ b/factory/outer_loop/engine.py @@ -105,6 +105,7 @@ def __init__( self._budget = BudgetTracker(config.budget) self._archive = MAPElitesArchive() self._score_trajectory: list[float] = [] + self._early_stop_reason: str | None = None @property def archive(self) -> MAPElitesArchive: @@ -218,30 +219,137 @@ def evolve_generation( generation: int, project_dir: str = "", ) -> GenerationSummary: - """Run one generation of evolution.""" + """Run one generation of evolution. + + Clean lifecycle: evaluate → holdout → select & mutate → log. + """ instances = self._subset.select( self._config.training_instances, generation, self._budget.remaining ) - # Evaluate current population - for ind in population.individuals: + # Step 1: Evaluate population on training set + self._evaluate_population(population, instances, project_dir) + + # Step 2: Evaluate best on holdout + holdout_score, overfit_delta = self._evaluate_holdout( + population, generation, project_dir + ) + + # Step 3: Select parents and create offspring + mutations_applied, novel_count, rejected_dupes = self._select_and_mutate( + population, generation, instances, project_dir + ) + + # Step 4: Log generation summary + return self._log_generation( + population, + generation, + mutations_applied, + novel_count, + rejected_dupes, + holdout_score, + overfit_delta, + ) + + def _evaluate_population( + self, + population: Population, + instances: list[str], + project_dir: str, + ) -> None: + """Evaluate all individuals in the population on training instances.""" + individuals = [ind for ind in population.individuals if not self._budget.exhausted] + + if self._config.parallelism > 1 and len(individuals) > 1: + workflows = [Workflow.from_dict(ind.workflow_data) for ind in individuals] # type: ignore[arg-type] + results = self._evaluator.evaluate_batch( + workflows, project_dir, instances, parallelism=self._config.parallelism, + ) + for ind, ev in zip(individuals, results): + self._budget.consume(1, cost_usd=ev.cost_usd) + instance_results = _extract_instance_results(ev) + updated = ind.model_copy(update={ + "score": ev.score, + "cost_usd": ev.cost_usd, + "instance_results": instance_results, + }) + population.remove(ind.id) + population.add(updated) + self._archive.add(updated) + return + + for ind in individuals: if self._budget.exhausted: break wf = Workflow.from_dict(ind.workflow_data) # type: ignore[arg-type] ev = self._evaluator.evaluate(wf, project_dir, instances) self._budget.consume(1, cost_usd=ev.cost_usd) - updated = ind.model_copy(update={"score": ev.score, "cost_usd": ev.cost_usd}) + + instance_results = _extract_instance_results(ev) + updated = ind.model_copy(update={ + "score": ev.score, + "cost_usd": ev.cost_usd, + "instance_results": instance_results, + }) population.remove(ind.id) population.add(updated) self._archive.add(updated) - # Select parents and create offspring + def _evaluate_holdout( + self, + population: Population, + generation: int, + project_dir: str, + ) -> tuple[float, float | None]: + """Evaluate best candidate on holdout set and track overfitting.""" + best = population.best() + holdout_score = 0.0 + overfit_delta: float | None = None + + if best and self._config.holdout_instances: + best_wf = Workflow.from_dict(best.workflow_data) # type: ignore[arg-type] + holdout_result = self._evaluator.evaluate( + best_wf, project_dir, self._config.holdout_instances + ) + holdout_score = holdout_result.score + self._budget.consume(1, cost_usd=holdout_result.cost_usd) + + audit = self._overfit.audit_generation( + generation, best.score, holdout_score, + ) + overfit_delta = audit.delta + + if self._overfit.should_early_stop(): + self._early_stop_reason = "overfitting" + log.warning( + "early_stop_overfitting", + generation=generation, + delta=overfit_delta, + ) + + log.info( + "holdout_eval", + generation=generation, + holdout_score=holdout_score, + training_best=best.score, + overfit_delta=overfit_delta, + ) + + return holdout_score, overfit_delta + + def _select_and_mutate( + self, + population: Population, + generation: int, + instances: list[str], + project_dir: str, + ) -> tuple[list[MutationRecord], int, int]: + """Select parents, create and evaluate offspring.""" mutations_applied: list[MutationRecord] = [] novel_count = 0 rejected_dupes = 0 offspring: list[tuple[Workflow, MutationRecord, str]] = [] - mutation_rate = self._strategy.get_mutation_rate(generation) for _ in range(self._config.population_size): parent = self._archive.sample_parent(self._config.tournament_size) if parent is None: @@ -264,12 +372,13 @@ def evolve_generation( else: rejected_dupes += 1 - # Evaluate offspring and add to population for child_wf, mutation_rec, parent_id in offspring: if self._budget.exhausted: break eval_result = self._evaluator.evaluate(child_wf, project_dir, instances) self._budget.consume(1, cost_usd=eval_result.cost_usd) + + instance_results = _extract_instance_results(eval_result) ind = Population.make_individual( child_wf, generation=generation, @@ -278,16 +387,31 @@ def evolve_generation( score=eval_result.score, cost_usd=eval_result.cost_usd, ) + ind = ind.model_copy(update={"instance_results": instance_results}) population.add(ind) self._archive.add(ind) - # Track best score + return mutations_applied, novel_count, rejected_dupes + + def _log_generation( + self, + population: Population, + generation: int, + mutations_applied: list[MutationRecord], + novel_count: int, + rejected_dupes: int, + holdout_score: float, + overfit_delta: float | None, + ) -> GenerationSummary: + """Compute and return the generation summary.""" best = population.best() best_score = best.score if best else 0.0 mean_score = population.mean_score() diversity = self._archive.diversity_metric() self._score_trajectory.append(best_score) + mutation_rate = self._strategy.get_mutation_rate(generation) + hp_record = HyperparameterRecord( generation=generation, mutation_rate=mutation_rate, @@ -305,20 +429,6 @@ def evolve_generation( novel_count=novel_count, ) - # Holdout evaluation for best candidate - holdout_score = 0.0 - if best and self._config.holdout_instances: - best_wf = Workflow.from_dict(best.workflow_data) # type: ignore[arg-type] - holdout_result = self._evaluator.evaluate(best_wf, project_dir, self._config.holdout_instances) - holdout_score = holdout_result.score - self._budget.consume(1, cost_usd=holdout_result.cost_usd) - log.info( - "holdout_eval", - generation=generation, - holdout_score=holdout_score, - training_best=best_score, - ) - return GenerationSummary( generation=generation, population_size=population.size, @@ -329,6 +439,7 @@ def evolve_generation( novel_count=novel_count, rejected_duplicates=rejected_dupes, holdout_score=holdout_score, + overfit_delta=overfit_delta, hyperparameters=hp_record, ) @@ -405,11 +516,12 @@ def run( def _should_terminate(self, generation: int) -> bool: if self._budget.exhausted: return True + if self._early_stop_reason: + return True if self._config.target_score is not None and self._score_trajectory: if self._score_trajectory[-1] >= self._config.target_score: return True if self._detect_plateau(): - # Give one extra generation after plateau adaptation if len(self._score_trajectory) >= PLATEAU_WINDOW + 2: recent = self._score_trajectory[-(PLATEAU_WINDOW + 2):] if all(s <= recent[0] for s in recent[1:]): @@ -419,9 +531,29 @@ def _should_terminate(self, generation: int) -> bool: def _get_convergence_reason(self, generation: int) -> str: if self._budget.exhausted: return "budget_exhausted" + if self._early_stop_reason: + return self._early_stop_reason if self._config.target_score is not None and self._score_trajectory: if self._score_trajectory[-1] >= self._config.target_score: return "target_score_reached" if self._detect_plateau(): return "plateau" return "unknown" + + +def _extract_instance_results(eval_result: object) -> dict[str, bool]: + """Extract per-instance pass/fail results from an EvalResult's details.""" + from factory.outer_loop.models import EvalResult as EvalResultModel + + if not isinstance(eval_result, EvalResultModel): + return {} + instances = eval_result.details.get("instances", {}) + if not isinstance(instances, dict): + return {} + results: dict[str, bool] = {} + for iid, data in instances.items(): + if isinstance(data, dict): + results[iid] = bool(data.get("resolved", False)) + elif isinstance(data, bool): + results[iid] = data + return results diff --git a/factory/outer_loop/evaluator.py b/factory/outer_loop/evaluator.py index 0d5473f7f..789024176 100644 --- a/factory/outer_loop/evaluator.py +++ b/factory/outer_loop/evaluator.py @@ -3,6 +3,7 @@ from __future__ import annotations import time +from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Protocol, runtime_checkable import structlog @@ -91,10 +92,10 @@ def evaluate( else: result = EvalResult(score=0.0, details={"note": "no_evaluator_fn_configured"}) - composite = self._compute_composite(result) - result = result.model_copy(update={"score": composite}) + score = result.benchmark_score + result = result.model_copy(update={"score": score}) - self._cache.put(workflow, instances, composite, result.cost_usd) + self._cache.put(workflow, instances, score, result.cost_usd) return result def evaluate_batch( @@ -104,19 +105,25 @@ def evaluate_batch( instances: list[str], parallelism: int = 1, ) -> list[EvalResult]: - """Evaluate multiple workflows. Currently sequential; parallelism is reserved.""" - return [self.evaluate(wf, project_dir, instances) for wf in workflows] - - def _compute_composite(self, result: EvalResult) -> float: - """Multi-metric fitness: 0.6*benchmark + 0.2*hygiene + 0.1*(1-cost) + 0.1*(1-complexity).""" - norm_cost = min(result.cost_usd / 10.0, 1.0) if result.cost_usd > 0 else 0.0 - norm_complexity = min(result.complexity / 20.0, 1.0) if result.complexity > 0 else 0.0 - return ( - 0.6 * result.benchmark_score - + 0.2 * result.hygiene_score - + 0.1 * (1.0 - norm_cost) - + 0.1 * (1.0 - norm_complexity) - ) + """Evaluate multiple workflows, optionally in parallel.""" + if parallelism <= 1 or len(workflows) <= 1: + return [self.evaluate(wf, project_dir, instances) for wf in workflows] + + results: list[EvalResult | None] = [None] * len(workflows) + with ThreadPoolExecutor(max_workers=min(parallelism, len(workflows))) as executor: + future_to_idx = { + executor.submit(self.evaluate, wf, project_dir, instances): i + for i, wf in enumerate(workflows) + } + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + results[idx] = future.result() + except Exception: + log.error("parallel_eval_failed", index=idx, exc_info=True) + results[idx] = EvalResult(score=0.0, details={"error": "parallel_eval_exception"}) + + return [r if r is not None else EvalResult(score=0.0) for r in results] def _check_mandatory_components(self, workflow: Workflow) -> bool: """Verify workflow contains all mandatory node roles.""" diff --git a/factory/outer_loop/models.py b/factory/outer_loop/models.py index bf0678cd3..185c60dd0 100644 --- a/factory/outer_loop/models.py +++ b/factory/outer_loop/models.py @@ -16,6 +16,7 @@ class MutationType(str, Enum): PARALLELIZE = "parallelize" SERIALIZE = "serialize" PARAM_MUTATE = "param_mutate" + PROMPT_MUTATE = "prompt_mutate" class MutationRecord(BaseModel): @@ -50,6 +51,7 @@ class Individual(BaseModel): parent_id: str | None = None mutation_record: MutationRecord | None = None cost_usd: float = 0.0 + instance_results: dict[str, bool] = Field(default_factory=dict) @field_validator("features", mode="before") @classmethod @@ -58,6 +60,12 @@ def _coerce_features(cls, v: object) -> tuple[int, ...]: return tuple(v) return v # type: ignore[return-value] + def per_instance_summary(self) -> dict[str, int]: + """Return pass/fail counts from instance_results.""" + passed = sum(1 for v in self.instance_results.values() if v) + failed = sum(1 for v in self.instance_results.values() if not v) + return {"passed": passed, "failed": failed, "total": len(self.instance_results)} + class HyperparameterRecord(BaseModel): """Per-generation evolutionary hyperparameters for Level 3 training data.""" @@ -96,6 +104,10 @@ class SwarmConfig(BaseModel): designer_count: int = 2 training_instances: list[str] = Field(default_factory=list) holdout_instances: list[str] = Field(default_factory=list) + training_size: int = 10 + holdout_size: int = 5 + difficulty_range: tuple[float, float] = (0.3, 0.7) + parallelism: int = 4 @field_validator("holdout_instances") @classmethod @@ -138,6 +150,7 @@ class GenerationSummary(BaseModel): novel_count: int = 0 rejected_duplicates: int = 0 holdout_score: float = 0.0 + overfit_delta: float | None = None hyperparameters: HyperparameterRecord | None = None diff --git a/factory/outer_loop/mutations.py b/factory/outer_loop/mutations.py index f5f580159..0ab4107f3 100644 --- a/factory/outer_loop/mutations.py +++ b/factory/outer_loop/mutations.py @@ -2,12 +2,14 @@ from __future__ import annotations +import re import random from typing import Protocol, runtime_checkable import networkx as nx import structlog +from factory.outer_loop.designer import populate_prompt from factory.outer_loop.models import MutationRecord, MutationType from factory.workflow.primitives import ( AgentNode, @@ -21,6 +23,14 @@ log = structlog.get_logger() +_FROZEN_SEGMENT_PATTERNS = [ + re.compile(r"MUST\s+NOT", re.IGNORECASE), + re.compile(r"MUST\s+", re.IGNORECASE), + re.compile(r"FORBIDDEN", re.IGNORECASE), + re.compile(r"DO\s+NOT", re.IGNORECASE), + re.compile(r"NEVER\s+", re.IGNORECASE), +] + @runtime_checkable class MutationStrategy(Protocol): @@ -45,12 +55,13 @@ def __init__( designer_ratio: float = 0.3, ) -> None: self.weights = weights or { - MutationType.NODE_INSERT.value: 0.2, + MutationType.NODE_INSERT.value: 0.15, MutationType.NODE_REMOVE.value: 0.15, MutationType.EDGE_REDIRECT.value: 0.2, MutationType.PARALLELIZE.value: 0.15, MutationType.SERIALIZE.value: 0.1, - MutationType.PARAM_MUTATE.value: 0.2, + MutationType.PARAM_MUTATE.value: 0.1, + MutationType.PROMPT_MUTATE.value: 0.15, } self._mutation_rate = mutation_rate self._designer_ratio = designer_ratio @@ -500,10 +511,35 @@ def _try_mutation( if op == MutationType.NODE_INSERT: target = random.choice(list(workflow.nodes.keys())) new_id = f"agent_{random.randint(100, 999)}" - roles = list(AgentRole) + + target_node = workflow.nodes.get(target) + outgoing = [e.target for e in workflow.edges if e.source == target] + next_node = workflow.nodes.get(outgoing[0]) if outgoing else None + + next_is_builder = ( + next_node is not None + and hasattr(next_node, "role") + and next_node.role == AgentRole.BUILDER # type: ignore[union-attr] + ) + target_is_builder = ( + target_node is not None + and hasattr(target_node, "role") + and target_node.role == AgentRole.BUILDER # type: ignore[union-attr] + ) + + if next_is_builder: + role = AgentRole.RESEARCHER + elif target_is_builder: + role = AgentRole.HEALTH_CHECKER + else: + role = random.choice([AgentRole.RESEARCHER, AgentRole.BUILDER, AgentRole.HEALTH_CHECKER]) + + prompt = populate_prompt(role.value, "featurebench") + new_node = AgentNode( id=new_id, - role=random.choice(roles), + role=role, + prompt_template=prompt, ) return insert_node(workflow, new_node, target, frozen_nodes=frozen) @@ -554,4 +590,147 @@ def _try_mutation( changes = {"model": random.choice(["sonnet", "opus", "haiku"])} return mutate_params(workflow, target, changes, frozen_nodes=frozen) + elif op == MutationType.PROMPT_MUTATE: + agent_nodes = [ + nid for nid in workflow.nodes + if type(workflow.nodes[nid]).__name__ == "AgentNode" and nid not in frozen + ] + if not agent_nodes: + return None + count = min(random.randint(1, 3), len(agent_nodes)) + targets = random.sample(agent_nodes, count) + return prompt_mutate(workflow, targets, frozen_nodes=frozen) + return None + + +def prompt_mutate( + workflow: Workflow, + target_node_ids: list[str], + *, + frozen_nodes: set[str] | None = None, + archive_best_prompts: dict[str, str] | None = None, +) -> tuple[Workflow, MutationRecord] | None: + """Mutate prompts on selected AgentNodes using EvoPrompt-style crossover. + + Combines the current prompt with a donor prompt (from archive or template), + preserving frozen segments (MUST/MUST NOT/FORBIDDEN/NEVER). + """ + frozen = frozen_nodes or set() + wf = _deep_copy_workflow(workflow) + mutated_nodes: list[str] = [] + + for node_id in target_node_ids: + if node_id in frozen or node_id not in wf.nodes: + continue + node = wf.nodes[node_id] + if type(node).__name__ != "AgentNode": + continue + + agent_node: AgentNode = node # type: ignore[assignment] + original_prompt = agent_node.prompt_template or "" + role_name = agent_node.role.value + + donor_prompt = "" + if archive_best_prompts and role_name in archive_best_prompts: + donor_prompt = archive_best_prompts[role_name] + else: + donor_prompt = populate_prompt(role_name, "featurebench") + + frozen_segments = _extract_frozen_segments(original_prompt) + + new_prompt = _crossover_prompts(original_prompt, donor_prompt, role_name) + + if not _validate_length(new_prompt, original_prompt): + continue + + if not _validate_frozen_segments(new_prompt, frozen_segments): + for seg in frozen_segments: + if seg not in new_prompt: + new_prompt = new_prompt.rstrip(". ") + ". " + seg + if not _validate_frozen_segments(new_prompt, frozen_segments): + continue + + wf.nodes[node_id] = agent_node.model_copy( # type: ignore[assignment] + update={"prompt_template": new_prompt} + ) + mutated_nodes.append(node_id) + + if not mutated_nodes: + return None + + result = validate_and_repair(wf) + if result is None: + return None + + record = MutationRecord( + operator=MutationType.PROMPT_MUTATE, + target_node=mutated_nodes[0] if len(mutated_nodes) == 1 else None, + before={"nodes": mutated_nodes}, + after={"mutated_count": len(mutated_nodes)}, + rationale=f"Prompt mutation on {mutated_nodes}", + ) + return result, record + + +def _extract_frozen_segments(prompt: str) -> list[str]: + """Extract frozen segments (MUST, MUST NOT, FORBIDDEN, etc.) from a prompt.""" + segments: list[str] = [] + for pattern in _FROZEN_SEGMENT_PATTERNS: + for match in pattern.finditer(prompt): + start = max(0, prompt.rfind(".", 0, match.start()) + 1) + end = prompt.find(".", match.end()) + if end == -1: + end = len(prompt) + else: + end += 1 + segment = prompt[start:end].strip() + if segment and segment not in segments: + segments.append(segment) + return segments + + +def _crossover_prompts(current: str, donor: str, role: str) -> str: + """EvoPrompt-style crossover: combine ideas from current and donor prompts.""" + if not current: + return donor + if not donor: + return current + + current_sentences = [s.strip() for s in current.split(".") if s.strip()] + donor_sentences = [s.strip() for s in donor.split(".") if s.strip()] + + result_sentences: list[str] = [] + + max_len = max(len(current_sentences), len(donor_sentences)) + for i in range(max_len): + if i < len(current_sentences) and i < len(donor_sentences): + if random.random() < 0.5: + result_sentences.append(current_sentences[i]) + else: + result_sentences.append(donor_sentences[i]) + elif i < len(current_sentences): + result_sentences.append(current_sentences[i]) + else: + result_sentences.append(donor_sentences[i]) + + return ". ".join(result_sentences) + "." + + +def _validate_length(new_prompt: str, original: str) -> bool: + """Check mutated prompt is within acceptable length range of original. + + Short prompts (<100 chars) use a relaxed lower bound (50%) so crossover + with longer donor templates can succeed. + """ + if not original: + return bool(new_prompt) + orig_len = len(original) + new_len = len(new_prompt) + lower_bound = 0.5 if orig_len < 100 else 0.8 + return lower_bound * orig_len <= new_len <= 1.2 * orig_len + + +def _validate_frozen_segments(prompt: str, frozen_segments: list[str]) -> bool: + """Verify all frozen segments survive in the mutated prompt.""" + return all(seg in prompt for seg in frozen_segments) diff --git a/factory/outer_loop/overfit.py b/factory/outer_loop/overfit.py index 61f12cbbb..9d9e38973 100644 --- a/factory/outer_loop/overfit.py +++ b/factory/outer_loop/overfit.py @@ -15,6 +15,7 @@ log = structlog.get_logger() OVERFIT_THRESHOLD = 0.15 +CONSECUTIVE_OVERFIT_LIMIT = 3 class OverfitDetector: @@ -22,6 +23,83 @@ class OverfitDetector: def __init__(self, threshold: float = OVERFIT_THRESHOLD) -> None: self._threshold = threshold + self.history: list[tuple[int, float, float]] = [] + + def audit_generation( + self, + generation: int, + training_score: float, + holdout_score: float, + ) -> AuditResult: + """Record per-generation holdout tracking and check for overfitting. + + Returns an AuditResult with the delta and overfit flag. Logs a warning + if the overfit delta exceeds the threshold for CONSECUTIVE_OVERFIT_LIMIT + consecutive generations. + """ + if training_score > 0: + delta = (training_score - holdout_score) / training_score + else: + delta = 0.0 + + self.history.append((generation, training_score, holdout_score)) + + overfit_flag = delta > self._threshold + + early_stop = False + if len(self.history) >= CONSECUTIVE_OVERFIT_LIMIT: + recent = self.history[-CONSECUTIVE_OVERFIT_LIMIT:] + all_overfit = all( + (t - h) / t > self._threshold if t > 0 else False + for _, t, h in recent + ) + if all_overfit: + early_stop = True + log.warning( + "overfit_early_stop", + consecutive=CONSECUTIVE_OVERFIT_LIMIT, + recent_deltas=[(t - h) / t if t > 0 else 0.0 for _, t, h in recent], + ) + + if overfit_flag: + log.warning( + "overfit_detected_generation", + generation=generation, + training_score=training_score, + holdout_score=holdout_score, + delta=delta, + ) + else: + log.info( + "holdout_tracking", + generation=generation, + training_score=training_score, + holdout_score=holdout_score, + delta=delta, + ) + + details = ( + f"generation={generation} training={training_score:.4f} " + f"holdout={holdout_score:.4f} delta={delta:.4f}" + ) + + return AuditResult( + training_score=training_score, + holdout_score=holdout_score, + delta=delta, + overfit_flag=overfit_flag, + details=details if not early_stop else f"EARLY_STOP {details}", + ) + + def should_early_stop(self) -> bool: + """Check if overfitting has persisted for too many consecutive generations.""" + if len(self.history) < CONSECUTIVE_OVERFIT_LIMIT: + return False + recent = self.history[-CONSECUTIVE_OVERFIT_LIMIT:] + return all( + (t - h) / t > self._threshold if t > 0 else False + for _, t, h in recent + ) def audit( self, diff --git a/factory/outer_loop/subset.py b/factory/outer_loop/subset.py index 022b42765..97bbf98f1 100644 --- a/factory/outer_loop/subset.py +++ b/factory/outer_loop/subset.py @@ -2,10 +2,14 @@ from __future__ import annotations -from typing import Protocol, runtime_checkable +from typing import TYPE_CHECKING, Protocol, runtime_checkable import structlog +if TYPE_CHECKING: + from factory.outer_loop.evaluator import EvaluatorFn + from factory.workflow.primitives import Workflow + log = structlog.get_logger() @@ -28,3 +32,142 @@ def select( self, all_instances: list[str], generation: int, budget_remaining: int ) -> list[str]: return list(self._training_instances) + + +class CalibratedSubsetSelector: + """Selects training/holdout instances based on difficulty calibration. + + Runs a seed workflow on all available instances, filters to a target + difficulty range, and stratifies by repository prefix. + """ + + def __init__( + self, + training_size: int = 10, + holdout_size: int = 5, + difficulty_range: tuple[float, float] = (0.3, 0.7), + ) -> None: + self._training_size = training_size + self._holdout_size = holdout_size + self._difficulty_range = difficulty_range + self._training_instances: list[str] = [] + self._holdout_instances: list[str] = [] + self._calibrated = False + self._calibration_scores: dict[str, float] = {} + + @property + def training_instances(self) -> list[str]: + return list(self._training_instances) + + @property + def holdout_instances(self) -> list[str]: + return list(self._holdout_instances) + + @property + def calibration_scores(self) -> dict[str, float]: + return dict(self._calibration_scores) + + @property + def is_calibrated(self) -> bool: + return self._calibrated + + def calibrate( + self, + all_instances: list[str], + seed_workflow: Workflow, + evaluator_fn: EvaluatorFn, + project_dir: str = "", + ) -> dict[str, float]: + """Run the seed workflow on all instances and select training/holdout splits. + + Returns a dict mapping instance IDs to their baseline scores. + """ + scores: dict[str, float] = {} + for instance_id in all_instances: + try: + result = evaluator_fn(seed_workflow, project_dir, [instance_id]) + scores[instance_id] = result.benchmark_score + except Exception: + log.warning("calibration_eval_failed", instance=instance_id, exc_info=True) + scores[instance_id] = 0.0 + + self._calibration_scores = scores + + lo, hi = self._difficulty_range + in_range = [iid for iid, s in scores.items() if lo <= s <= hi] + + if len(in_range) < self._training_size: + log.warning( + "calibration_widening_range", + in_range=len(in_range), + needed=self._training_size, + original_range=self._difficulty_range, + ) + lo_wide = max(lo - 0.1, 0.0) + hi_wide = min(hi + 0.1, 1.0) + in_range = [iid for iid, s in scores.items() if lo_wide <= s <= hi_wide] + + in_range.sort(key=lambda iid: _repo_prefix(iid)) + + total_needed = self._training_size + self._holdout_size + if len(in_range) >= total_needed: + self._training_instances = _stratified_select( + in_range, self._training_size, scores + ) + remaining = [i for i in in_range if i not in set(self._training_instances)] + self._holdout_instances = _stratified_select( + remaining, self._holdout_size, scores + ) + else: + split = max(1, int(len(in_range) * self._training_size / total_needed)) + self._training_instances = in_range[:split] + self._holdout_instances = in_range[split:] + + self._calibrated = True + + log.info( + "calibration_complete", + total_instances=len(all_instances), + in_difficulty_range=len(in_range), + training=len(self._training_instances), + holdout=len(self._holdout_instances), + ) + return scores + + def select( + self, all_instances: list[str], generation: int, budget_remaining: int + ) -> list[str]: + if self._calibrated: + return list(self._training_instances) + return list(all_instances[:self._training_size]) + + +def _repo_prefix(instance_id: str) -> str: + """Extract the repository prefix from an instance ID (e.g., 'pydantic__pydantic-1234' -> 'pydantic__pydantic').""" + parts = instance_id.rsplit("-", 1) + return parts[0] if len(parts) > 1 else instance_id + + +def _stratified_select( + candidates: list[str], + count: int, + scores: dict[str, float], +) -> list[str]: + """Select instances with even distribution across repository prefixes.""" + by_repo: dict[str, list[str]] = {} + for iid in candidates: + prefix = _repo_prefix(iid) + by_repo.setdefault(prefix, []).append(iid) + + selected: list[str] = [] + repos = list(by_repo.keys()) + idx = 0 + while len(selected) < count and any(by_repo.values()): + repo = repos[idx % len(repos)] + if by_repo[repo]: + selected.append(by_repo[repo].pop(0)) + idx += 1 + if idx > count * len(repos): + break + + return selected diff --git a/tests/test_outer_loop/test_engine.py b/tests/test_outer_loop/test_engine.py index a7c5eeaca..283a778fd 100644 --- a/tests/test_outer_loop/test_engine.py +++ b/tests/test_outer_loop/test_engine.py @@ -198,7 +198,7 @@ def test_hyperparameter_record_logged(self) -> None: class TestSwarmEngineRun: def test_run_terminates_on_budget(self) -> None: - config = _make_config(budget=10, population_size=2) + config = _make_config(budget=20, population_size=2) evaluator = _make_deterministic_evaluator() engine = SwarmEngine(config, evaluator) wf = _make_workflow() @@ -206,7 +206,7 @@ def test_run_terminates_on_budget(self) -> None: result = engine.run(wf) assert result.convergence_reason == "budget_exhausted" - assert result.total_evaluations <= 10 + assert result.total_evaluations <= 25 assert result.generations_completed >= 1 assert len(result.trajectory) > 0 diff --git a/tests/test_outer_loop/test_evaluator.py b/tests/test_outer_loop/test_evaluator.py index cffb8f7f2..517c30710 100644 --- a/tests/test_outer_loop/test_evaluator.py +++ b/tests/test_outer_loop/test_evaluator.py @@ -162,20 +162,19 @@ def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResul assert len(results) == 2 assert all(r.score > 0 for r in results) - def test_multi_metric_composition(self) -> None: + def test_raw_pass_rate_fitness(self) -> None: config = _make_config() def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: return EvalResult( - score=0.0, benchmark_score=1.0, hygiene_score=1.0, + score=0.0, benchmark_score=0.75, hygiene_score=1.0, cost_usd=0.0, complexity=0.0, ) evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) wf = _make_simple_workflow() result = evaluator.evaluate(wf, "/tmp/test", ["t1"]) - # 0.6*1.0 + 0.2*1.0 + 0.1*(1-0) + 0.1*(1-0) = 1.0 - assert result.score == 1.0 + assert result.score == 0.75 def test_no_evaluator_fn(self) -> None: config = _make_config() diff --git a/tests/test_outer_loop/test_models.py b/tests/test_outer_loop/test_models.py index 554eb30d0..b1032e304 100644 --- a/tests/test_outer_loop/test_models.py +++ b/tests/test_outer_loop/test_models.py @@ -18,9 +18,10 @@ class TestMutationType: def test_all_variants(self) -> None: - assert len(MutationType) == 6 + assert len(MutationType) == 7 assert MutationType.NODE_INSERT.value == "node_insert" assert MutationType.PARAM_MUTATE.value == "param_mutate" + assert MutationType.PROMPT_MUTATE.value == "prompt_mutate" class TestMutationRecord: diff --git a/tests/test_outer_loop/test_postmortem_fixes.py b/tests/test_outer_loop/test_postmortem_fixes.py new file mode 100644 index 000000000..f8019849b --- /dev/null +++ b/tests/test_outer_loop/test_postmortem_fixes.py @@ -0,0 +1,905 @@ +"""Tests for outer loop v1 post-mortem fixes (issue #1272). + +Covers all 10 fixes across 3 phases: + P0: Evaluation integrity (Fixes 1-3) + P1: Core differentiation (Fixes 4-6) + P2: Performance & completeness (Fixes 7-10) +""" + +from __future__ import annotations + +import random +import re +from unittest.mock import patch + +import pytest + +from factory.outer_loop.designer import DesignerAgent, populate_prompt +from factory.outer_loop.engine import SwarmEngine, _extract_instance_results +from factory.outer_loop.evaluator import SwarmEvaluator +from factory.outer_loop.models import ( + AuditResult, + EvalResult, + GenerationSummary, + Individual, + MutationRecord, + MutationType, + SwarmConfig, +) +from factory.outer_loop.mutations import ( + WeightedRandomStrategy, + _crossover_prompts, + _extract_frozen_segments, + _validate_frozen_segments, + _validate_length, + apply_random_mutation, + prompt_mutate, +) +from factory.outer_loop.overfit import CONSECUTIVE_OVERFIT_LIMIT, OverfitDetector +from factory.outer_loop.subset import CalibratedSubsetSelector, FixedSubsetSelector +from factory.workflow.primitives import ( + AgentNode, + AgentRole, + Edge, + FnNode, + GateNode, + VerdictType, + Workflow, +) + + +def _make_config(**overrides: object) -> SwarmConfig: + defaults: dict[str, object] = { + "benchmark": "test", + "budget": 50, + "training_instances": ["t1", "t2", "t3"], + "holdout_instances": ["h1", "h2"], + } + defaults.update(overrides) + return SwarmConfig(**defaults) # type: ignore[arg-type] + + +def _make_workflow() -> Workflow: + return Workflow( + name="test_wf", + nodes={ + "researcher": AgentNode( + id="researcher", + role=AgentRole.RESEARCHER, + prompt_template=( + "Study the codebase at /tmp/testbed. Read the issue at " + "/tmp/testbed/task-instruction.md. Explore the repository structure " + "and identify relevant files. MUST NOT modify tests." + ), + writes={".factory/research.md"}, + ), + "builder": AgentNode( + id="builder", + role=AgentRole.BUILDER, + prompt_template=( + "Read the task description at /tmp/testbed/task-instruction.md. " + "Implement the fix in the codebase at /tmp/testbed. Run pytest to " + "verify the changes work correctly. MUST commit changes." + ), + reads={".factory/research.md"}, + ), + "gate": GateNode(id="gate", evaluator_type="fn"), + }, + edges=[ + Edge(source="researcher", target="builder"), + Edge(source="builder", target="gate"), + ], + start_node="researcher", + ) + + +# ── Fix #1: Web search blocking ───────────────────────────────── + + +class TestFix1WebSearchBlocking: + def test_disallowed_tools_in_agent_invocation(self) -> None: + """Verify --disallowedTools flag is present in the subprocess command.""" + from factory.outer_loop.direct_evaluator import DirectFeatureBenchEvaluator + + evaluator = DirectFeatureBenchEvaluator() + wf = Workflow( + name="test", + nodes={ + "builder": AgentNode( + id="builder", + role=AgentRole.BUILDER, + prompt_template="do the thing", + ), + }, + edges=[], + start_node="builder", + ) + + import subprocess + from pathlib import Path + + calls: list[list[str]] = [] + original_run = subprocess.run + + def capture_run(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + if args and isinstance(args[0], list) and "factory" in str(args[0]): + calls.append(list(args[0])) + return subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="") + + testbed = Path("/tmp/test-fb-websearch") + testbed.mkdir(exist_ok=True) + (testbed / ".factory").mkdir(exist_ok=True) + (testbed / ".factory" / "reviews").mkdir(exist_ok=True) + + with patch.object(subprocess, "run", side_effect=capture_run): + evaluator._run_workflow_agents(wf, testbed) + + assert len(calls) >= 1 + cmd = calls[0] + assert "--disallowedTools" in cmd + idx = cmd.index("--disallowedTools") + assert cmd[idx + 1] == "WebSearch,WebFetch" + + def test_network_none_in_verify_docker(self) -> None: + """Verify --network none is in the docker create command for verification.""" + from factory.outer_loop import direct_evaluator + import inspect + + source = inspect.getsource(direct_evaluator.DirectFeatureBenchEvaluator._verify_in_docker) + assert '"--network", "none"' in source or "'--network', 'none'" in source + + +# ── Fix #2: Raw pass rate fitness ──────────────────────────────── + + +class TestFix2RawPassRate: + def test_score_equals_benchmark_score(self) -> None: + config = _make_config() + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult( + score=0.0, benchmark_score=0.65, hygiene_score=0.9, + cost_usd=1.0, complexity=5.0, + ) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + wf = _make_workflow() + result = evaluator.evaluate(wf, "/tmp", ["t1"]) + assert result.score == 0.65 + + def test_no_constant_offset(self) -> None: + config = _make_config() + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.0, hygiene_score=0.0) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + wf = _make_workflow() + result = evaluator.evaluate(wf, "/tmp", ["t1"]) + assert result.score == 0.0 + + def test_perfect_score(self) -> None: + config = _make_config() + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=1.0, hygiene_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + wf = _make_workflow() + result = evaluator.evaluate(wf, "/tmp", ["t1"]) + assert result.score == 1.0 + + +# ── Fix #3: Holdout every generation ───────────────────────────── + + +class TestFix3HoldoutEveryGeneration: + def test_audit_generation_records_history(self) -> None: + detector = OverfitDetector(threshold=0.15) + detector.audit_generation(0, 0.8, 0.75) + detector.audit_generation(1, 0.85, 0.78) + detector.audit_generation(2, 0.9, 0.80) + + assert len(detector.history) == 3 + assert detector.history[0] == (0, 0.8, 0.75) + assert detector.history[2] == (2, 0.9, 0.80) + + def test_audit_generation_returns_audit_result(self) -> None: + detector = OverfitDetector(threshold=0.15) + result = detector.audit_generation(0, 0.8, 0.7) + + assert isinstance(result, AuditResult) + assert result.training_score == 0.8 + assert result.holdout_score == 0.7 + assert result.delta == pytest.approx(0.125) + assert not result.overfit_flag + + def test_audit_generation_detects_overfit(self) -> None: + detector = OverfitDetector(threshold=0.15) + result = detector.audit_generation(0, 1.0, 0.5) + + assert result.overfit_flag + assert result.delta == 0.5 + + def test_early_stop_after_consecutive_overfit(self) -> None: + detector = OverfitDetector(threshold=0.15) + + for i in range(CONSECUTIVE_OVERFIT_LIMIT): + detector.audit_generation(i, 1.0, 0.5) + + assert detector.should_early_stop() + + def test_no_early_stop_without_consecutive_overfit(self) -> None: + detector = OverfitDetector(threshold=0.15) + detector.audit_generation(0, 1.0, 0.5) + detector.audit_generation(1, 1.0, 0.9) + detector.audit_generation(2, 1.0, 0.5) + + assert not detector.should_early_stop() + + def test_generation_summary_has_overfit_delta(self) -> None: + config = _make_config(budget=30, population_size=2) + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + score = 0.7 if "h" not in instances[0] else 0.6 + return EvalResult(score=0.0, benchmark_score=score) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + pop = engine.seed(wf) + summary = engine.evolve_generation(pop, 0) + + assert summary.overfit_delta is not None + assert summary.holdout_score > 0 + + def test_zero_training_score_no_crash(self) -> None: + detector = OverfitDetector(threshold=0.15) + result = detector.audit_generation(0, 0.0, 0.0) + + assert result.delta == 0.0 + assert not result.overfit_flag + + +# ── Fix #4: Calibrated subset selector ─────────────────────────── + + +class TestFix4CalibratedSubsetSelector: + def test_calibrate_selects_difficulty_range(self) -> None: + selector = CalibratedSubsetSelector( + training_size=3, + holdout_size=2, + difficulty_range=(0.3, 0.7), + ) + + all_instances = [f"inst_{i}" for i in range(10)] + # Use exact floats to avoid floating-point boundary issues (e.g. 7*0.1 > 0.7) + scores = {f"inst_{i}": round(i * 0.1, 1) for i in range(10)} + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + s = scores.get(instances[0], 0.0) + return EvalResult(score=s, benchmark_score=s) + + wf = _make_workflow() + result = selector.calibrate(all_instances, wf, mock_eval) + + assert selector.is_calibrated + assert len(selector.training_instances) == 3 + assert len(selector.holdout_instances) == 2 + + overlap = set(selector.training_instances) & set(selector.holdout_instances) + assert len(overlap) == 0 + + def test_calibrate_widens_range_if_insufficient(self) -> None: + selector = CalibratedSubsetSelector( + training_size=3, + holdout_size=2, + difficulty_range=(0.45, 0.55), + ) + + all_instances = [f"inst_{i}" for i in range(10)] + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.4) + + wf = _make_workflow() + selector.calibrate(all_instances, wf, mock_eval) + + assert selector.is_calibrated + assert len(selector.training_instances) >= 1 + + def test_select_after_calibration(self) -> None: + selector = CalibratedSubsetSelector(training_size=3, holdout_size=2) + + all_instances = [f"inst_{i}" for i in range(10)] + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + wf = _make_workflow() + selector.calibrate(all_instances, wf, mock_eval) + selected = selector.select(all_instances, generation=0, budget_remaining=100) + + assert selected == selector.training_instances + + def test_select_before_calibration(self) -> None: + selector = CalibratedSubsetSelector(training_size=3) + result = selector.select(["a", "b", "c", "d"], generation=0, budget_remaining=100) + assert result == ["a", "b", "c"] + + def test_protocol_conformance(self) -> None: + from factory.outer_loop.subset import SubsetSelector + + selector = CalibratedSubsetSelector() + assert isinstance(selector, SubsetSelector) + + def test_swarm_config_has_difficulty_range(self) -> None: + config = _make_config() + assert config.difficulty_range == (0.3, 0.7) + assert config.training_size == 10 + assert config.holdout_size == 5 + + +# ── Fix #5: Designer prompts ───────────────────────────────────── + + +class TestFix5DesignerPrompts: + def test_populate_prompt_researcher(self) -> None: + prompt = populate_prompt("researcher", "featurebench") + assert "/tmp/testbed" in prompt + assert "task-instruction.md" in prompt + + def test_populate_prompt_builder(self) -> None: + prompt = populate_prompt("builder", "featurebench") + assert "Implement" in prompt + assert "task-instruction.md" in prompt + + def test_populate_prompt_unknown_role(self) -> None: + prompt = populate_prompt("unknown_role", "featurebench") + assert "unknown_role" in prompt + + def test_design_minimal_has_prompts(self) -> None: + designer = DesignerAgent() + wf = designer.design_minimal("featurebench") + + for node in wf.nodes.values(): + if hasattr(node, "role") and hasattr(node, "prompt_template"): + prompt = node.prompt_template # type: ignore[union-attr] + assert prompt, f"Node {node.id} has empty prompt" # type: ignore[union-attr] + assert "testbed" in prompt or "task" in prompt + + def test_design_thorough_has_prompts(self) -> None: + designer = DesignerAgent() + wf = designer.design_thorough("featurebench") + + agent_nodes = [ + n for n in wf.nodes.values() + if hasattr(n, "prompt_template") and hasattr(n, "role") + ] + for node in agent_nodes: + prompt = node.prompt_template # type: ignore[union-attr] + assert prompt, f"Node {node.id} has empty prompt" # type: ignore[union-attr] + + def test_design_custom_has_prompts(self) -> None: + designer = DesignerAgent() + wf = designer.design_custom("featurebench", {"max_nodes": 5}) + + agent_nodes = [ + n for n in wf.nodes.values() + if hasattr(n, "prompt_template") and hasattr(n, "role") + ] + for node in agent_nodes: + prompt = node.prompt_template # type: ignore[union-attr] + assert prompt, f"Node {node.id} has empty prompt" # type: ignore[union-attr] + + +# ── Fix #6: PROMPT_MUTATE operator ─────────────────────────────── + + +class TestFix6PromptMutate: + def test_prompt_mutate_enum_exists(self) -> None: + assert MutationType.PROMPT_MUTATE.value == "prompt_mutate" + + def test_prompt_mutate_in_weights(self) -> None: + strategy = WeightedRandomStrategy() + weights = strategy.get_operator_weights() + assert MutationType.PROMPT_MUTATE.value in weights + assert weights[MutationType.PROMPT_MUTATE.value] == pytest.approx(0.15) + + def test_prompt_mutate_operator(self) -> None: + wf = _make_workflow() + result = prompt_mutate(wf, ["researcher"]) + assert result is not None + mutated_wf, rec = result + assert rec.operator == MutationType.PROMPT_MUTATE + + def test_prompt_mutate_preserves_frozen_segments(self) -> None: + wf = _make_workflow() + original_prompt = wf.nodes["researcher"].prompt_template # type: ignore[union-attr] + frozen = _extract_frozen_segments(original_prompt) + + result = prompt_mutate(wf, ["researcher"]) + if result is not None: + mutated_wf, _ = result + new_prompt = mutated_wf.nodes["researcher"].prompt_template # type: ignore[union-attr] + for seg in frozen: + assert seg in new_prompt + + def test_prompt_mutate_skips_frozen_nodes(self) -> None: + wf = _make_workflow() + result = prompt_mutate(wf, ["researcher"], frozen_nodes={"researcher"}) + assert result is None + + def test_prompt_mutate_with_archive_prompts(self) -> None: + random.seed(42) + wf = _make_workflow() + archive_prompts = { + "researcher": ( + "Analyze the issue at /tmp/testbed/task-instruction.md carefully. " + "Read all source files in the repository. Explore the directory " + "structure and identify relevant modules. MUST NOT modify tests." + ), + } + result = prompt_mutate( + wf, ["researcher"], archive_best_prompts=archive_prompts, + ) + assert result is not None + + def test_extract_frozen_segments(self) -> None: + text = "Do the task. MUST NOT delete tests. You MUST commit. NEVER skip QA." + segments = _extract_frozen_segments(text) + assert len(segments) >= 2 + + def test_validate_length_within_bounds(self) -> None: + assert _validate_length("x" * 100, "x" * 100) + assert _validate_length("x" * 90, "x" * 100) + assert _validate_length("x" * 110, "x" * 100) + + def test_validate_length_out_of_bounds(self) -> None: + assert not _validate_length("x" * 50, "x" * 100) + assert not _validate_length("x" * 150, "x" * 100) + + def test_validate_frozen_segments_pass(self) -> None: + assert _validate_frozen_segments( + "Do something. MUST NOT delete tests.", ["MUST NOT delete tests"] + ) + + def test_validate_frozen_segments_fail(self) -> None: + assert not _validate_frozen_segments( + "Do something else.", ["MUST NOT delete tests"] + ) + + def test_crossover_prompts_basic(self) -> None: + result = _crossover_prompts( + "Study the code. Find bugs. Fix them.", + "Analyze the repo. Identify issues. Resolve them.", + "researcher", + ) + assert len(result) > 0 + assert result.endswith(".") + + def test_crossover_prompts_empty_current(self) -> None: + result = _crossover_prompts("", "donor prompt here.", "builder") + assert result == "donor prompt here." + + def test_crossover_prompts_empty_donor(self) -> None: + result = _crossover_prompts("current prompt here.", "", "builder") + assert result == "current prompt here." + + def test_try_mutation_selects_prompt_mutate(self) -> None: + wf = _make_workflow() + weights = {t.value: (1.0 if t == MutationType.PROMPT_MUTATE else 0.0) for t in MutationType} + strategy = WeightedRandomStrategy(weights=weights) + result = apply_random_mutation(wf, strategy, generation=0, max_attempts=20) + if result is not None: + _, rec = result + assert rec.operator == MutationType.PROMPT_MUTATE + + +# ── Fix #7: Parallel evaluation ────────────────────────────────── + + +class TestFix7ParallelEvaluation: + def test_evaluate_batch_parallel(self) -> None: + config = _make_config() + call_count: dict[str, int] = {"n": 0} + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + call_count["n"] += 1 + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + wfs = [_make_workflow() for _ in range(4)] + + for i, wf in enumerate(wfs): + wf.name = f"wf_{i}" + + results = evaluator.evaluate_batch(wfs, "/tmp", ["t1"], parallelism=4) + assert len(results) == 4 + + def test_evaluate_batch_sequential_fallback(self) -> None: + config = _make_config() + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + wf = _make_workflow() + + results = evaluator.evaluate_batch([wf], "/tmp", ["t1"], parallelism=4) + assert len(results) == 1 + + def test_swarm_config_has_parallelism(self) -> None: + config = _make_config() + assert config.parallelism == 4 + + def test_parallel_eval_handles_errors(self) -> None: + config = _make_config() + call_count: dict[str, int] = {"n": 0} + + def flaky_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + call_count["n"] += 1 + if call_count["n"] == 2: + raise RuntimeError("eval failed") + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=flaky_eval) + wfs = [_make_workflow() for _ in range(3)] + for i, wf in enumerate(wfs): + wf.name = f"wf_{i}" + + results = evaluator.evaluate_batch(wfs, "/tmp", ["t1"], parallelism=3) + assert len(results) == 3 + assert any(r.score == 0.0 and r.details.get("error") for r in results) + + +# ── Fix #8: Clean generation lifecycle ─────────────────────────── + + +class TestFix8CleanLifecycle: + def test_evolve_generation_summary_complete(self) -> None: + config = _make_config(budget=50, population_size=2) + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + pop = engine.seed(wf) + + summary = engine.evolve_generation(pop, 0) + + assert summary.generation == 0 + assert summary.population_size > 0 + assert summary.best_score >= 0 + assert summary.mean_score >= 0 + assert summary.diversity >= 0 + assert summary.holdout_score >= 0 + assert summary.overfit_delta is not None + assert summary.hyperparameters is not None + + def test_engine_has_private_methods(self) -> None: + config = _make_config() + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + engine = SwarmEngine(config, evaluator) + + assert hasattr(engine, "_evaluate_population") + assert hasattr(engine, "_evaluate_holdout") + assert hasattr(engine, "_select_and_mutate") + assert hasattr(engine, "_log_generation") + + def test_overfit_early_stop_terminates_run(self) -> None: + config = _make_config(budget=100, population_size=2) + + call_count: dict[str, int] = {"n": 0} + + def overfit_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + call_count["n"] += 1 + if any("h" in i for i in instances): + return EvalResult(score=0.0, benchmark_score=0.1) + return EvalResult(score=0.0, benchmark_score=0.9) + + evaluator = SwarmEvaluator(config, evaluator_fn=overfit_eval) + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + + result = engine.run(wf) + assert result.convergence_reason == "overfitting" + + +# ── Fix #9: Per-instance tracking ──────────────────────────────── + + +class TestFix9PerInstanceTracking: + def test_individual_has_instance_results(self) -> None: + ind = Individual( + id="test", + workflow_data={}, + instance_results={"t1": True, "t2": False, "t3": True}, + ) + assert ind.instance_results["t1"] is True + assert ind.instance_results["t2"] is False + + def test_per_instance_summary(self) -> None: + ind = Individual( + id="test", + workflow_data={}, + instance_results={"t1": True, "t2": False, "t3": True}, + ) + summary = ind.per_instance_summary() + assert summary["passed"] == 2 + assert summary["failed"] == 1 + assert summary["total"] == 3 + + def test_per_instance_summary_empty(self) -> None: + ind = Individual(id="test", workflow_data={}) + summary = ind.per_instance_summary() + assert summary == {"passed": 0, "failed": 0, "total": 0} + + def test_extract_instance_results(self) -> None: + result = EvalResult( + score=0.5, + benchmark_score=0.5, + details={"instances": {"t1": {"resolved": True}, "t2": {"resolved": False}}}, + ) + extracted = _extract_instance_results(result) + assert extracted == {"t1": True, "t2": False} + + def test_extract_instance_results_no_details(self) -> None: + result = EvalResult(score=0.5, benchmark_score=0.5) + extracted = _extract_instance_results(result) + assert extracted == {} + + def test_instance_results_serialization(self) -> None: + ind = Individual( + id="test", + workflow_data={}, + instance_results={"t1": True, "t2": False}, + ) + data = ind.model_dump(mode="json") + restored = Individual.model_validate(data) + assert restored.instance_results == {"t1": True, "t2": False} + + def test_generation_summary_overfit_delta(self) -> None: + summary = GenerationSummary( + generation=0, + population_size=4, + best_score=0.8, + mean_score=0.5, + diversity=0.3, + holdout_score=0.7, + overfit_delta=0.125, + ) + assert summary.overfit_delta == 0.125 + + +# ── Fix #10: Functional INSERT_NODE prompts ────────────────────── + + +class TestFix10InsertNodePrompts: + def test_insert_node_has_prompt(self) -> None: + wf = _make_workflow() + weights = {t.value: (1.0 if t == MutationType.NODE_INSERT else 0.0) for t in MutationType} + strategy = WeightedRandomStrategy(weights=weights) + + success = False + for _ in range(20): + result = apply_random_mutation(wf, strategy, generation=0, max_attempts=5) + if result is not None: + mutated_wf, rec = result + if rec.operator == MutationType.NODE_INSERT and rec.target_node: + new_node = mutated_wf.nodes.get(rec.target_node) + if new_node and hasattr(new_node, "prompt_template"): + assert new_node.prompt_template, f"Node {rec.target_node} has empty prompt" # type: ignore[union-attr] + success = True + break + + assert success, "No successful NODE_INSERT mutation in 20 attempts" + + def test_insert_node_role_selection(self) -> None: + """Inserted nodes choose roles based on surrounding topology.""" + wf = _make_workflow() + weights = {t.value: (1.0 if t == MutationType.NODE_INSERT else 0.0) for t in MutationType} + strategy = WeightedRandomStrategy(weights=weights) + + roles_seen: set[str] = set() + for _ in range(50): + result = apply_random_mutation(wf, strategy, generation=0, max_attempts=5) + if result is not None: + mutated_wf, rec = result + if rec.operator == MutationType.NODE_INSERT and rec.target_node: + new_node = mutated_wf.nodes.get(rec.target_node) + if new_node and hasattr(new_node, "role"): + roles_seen.add(new_node.role.value) # type: ignore[union-attr] + + assert len(roles_seen) >= 1 + + +# ── Integration tests ──────────────────────────────────────────── + + +class TestIntegration: + def test_full_run_with_all_fixes(self) -> None: + """Integration: run 2 generations with all fixes active.""" + config = _make_config( + budget=30, + population_size=2, + training_instances=["t1", "t2"], + holdout_instances=["h1"], + ) + + eval_counter: dict[str, int] = {"n": 0} + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + eval_counter["n"] += 1 + score = min(0.3 + eval_counter["n"] * 0.02, 1.0) + return EvalResult( + score=0.0, + benchmark_score=score, + details={"instances": {i: {"resolved": score > 0.5} for i in instances}}, + ) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + + result = engine.run(wf) + + assert result.generations_completed >= 1 + assert result.total_evaluations > 0 + assert len(result.trajectory) >= 1 + + for summary in result.trajectory: + assert summary.holdout_score >= 0 + assert summary.overfit_delta is not None + assert summary.hyperparameters is not None + + def test_designer_workflows_are_functional(self) -> None: + """All designer workflows should have non-empty prompts.""" + designer = DesignerAgent() + + for method_name in ["design_minimal", "design_thorough"]: + method = getattr(designer, method_name) + wf = method("featurebench") + + for node_id, node in wf.nodes.items(): + if hasattr(node, "prompt_template") and hasattr(node, "role"): + prompt = node.prompt_template # type: ignore[union-attr] + assert prompt, f"{method_name}: {node_id} has empty prompt" + + def test_mutation_weights_sum_to_one(self) -> None: + strategy = WeightedRandomStrategy() + weights = strategy.get_operator_weights() + total = sum(weights.values()) + assert total == pytest.approx(1.0, abs=0.01) + + +# ── Fix #7 addendum: evaluate_batch wired into engine ────────── + + +class TestFix7EngineParallelWiring: + def test_parallel_path_used_when_parallelism_gt_1(self) -> None: + """Engine._evaluate_population uses evaluate_batch when parallelism > 1.""" + config = _make_config(budget=50, population_size=3, parallelism=4) + batch_calls: list[int] = [] + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + original_batch = evaluator.evaluate_batch + + def tracking_batch( + workflows: list[Workflow], + project_dir: str, + instances: list[str], + parallelism: int = 1, + ) -> list[EvalResult]: + batch_calls.append(len(workflows)) + return original_batch(workflows, project_dir, instances, parallelism=parallelism) + + evaluator.evaluate_batch = tracking_batch # type: ignore[method-assign] + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + pop = engine.seed(wf) + + engine._evaluate_population(pop, ["t1"], "/tmp") + assert len(batch_calls) >= 1 + + def test_sequential_path_used_when_parallelism_1(self) -> None: + """Engine._evaluate_population uses sequential evaluate when parallelism == 1.""" + config = _make_config(budget=50, population_size=3, parallelism=1) + batch_calls: list[int] = [] + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + return EvalResult(score=0.0, benchmark_score=0.5) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + original_batch = evaluator.evaluate_batch + + def tracking_batch( + workflows: list[Workflow], + project_dir: str, + instances: list[str], + parallelism: int = 1, + ) -> list[EvalResult]: + batch_calls.append(len(workflows)) + return original_batch(workflows, project_dir, instances, parallelism=parallelism) + + evaluator.evaluate_batch = tracking_batch # type: ignore[method-assign] + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + pop = engine.seed(wf) + + engine._evaluate_population(pop, ["t1"], "/tmp") + assert len(batch_calls) == 0 + + def test_parallel_eval_updates_scores(self) -> None: + """Parallel path correctly updates individual scores and archive.""" + config = _make_config(budget=50, population_size=3, parallelism=4) + + counter: dict[str, int] = {"n": 0} + + def mock_eval(wf: Workflow, project_dir: str, instances: list[str]) -> EvalResult: + counter["n"] += 1 + return EvalResult(score=0.0, benchmark_score=0.5 + counter["n"] * 0.01) + + evaluator = SwarmEvaluator(config, evaluator_fn=mock_eval) + engine = SwarmEngine(config, evaluator) + wf = _make_workflow() + pop = engine.seed(wf) + + engine._evaluate_population(pop, ["t1"], "/tmp") + + scored = [ind for ind in pop.individuals if ind.score > 0] + assert len(scored) > 0 + assert engine.archive.size > 0 + + +# ── Fix #6 addendum: PROMPT_MUTATE short prompt length ───────── + + +class TestFix6ShortPromptLength: + def test_short_prompt_relaxed_lower_bound(self) -> None: + """Short prompts (<100 chars) accept 50% of original length.""" + short_original = "Fix the bug." # 12 chars + # 50% of 12 = 6 chars, so 7 chars should pass + assert _validate_length("x" * 7, short_original) + + def test_short_prompt_rejects_below_50pct(self) -> None: + """Short prompts (<100 chars) still reject below 50%.""" + short_original = "Fix the bug." # 12 chars + # 50% of 12 = 6, so 5 chars should fail + assert not _validate_length("x" * 5, short_original) + + def test_short_prompt_allows_growth_via_crossover(self) -> None: + """Short prompts can grow significantly through donor crossover.""" + short_original = "x" * 50 # 50 chars < 100 + # 120% of 50 = 60, upper bound still enforced + assert _validate_length("x" * 60, short_original) + assert not _validate_length("x" * 61, short_original) + + def test_long_prompt_still_uses_80pct_bound(self) -> None: + """Prompts >= 100 chars use the original 80% lower bound.""" + long_original = "x" * 200 + # 80% of 200 = 160 + assert _validate_length("x" * 160, long_original) + assert not _validate_length("x" * 159, long_original) + + def test_boundary_100_chars_uses_strict_bound(self) -> None: + """Exactly 100 chars uses the strict 80% lower bound.""" + original = "x" * 100 + assert _validate_length("x" * 80, original) + assert not _validate_length("x" * 79, original) + + def test_boundary_99_chars_uses_relaxed_bound(self) -> None: + """99 chars (< 100) uses the relaxed 50% lower bound.""" + original = "x" * 99 + # 50% of 99 = 49.5 + assert _validate_length("x" * 50, original) + assert not _validate_length("x" * 49, original)