diff --git a/.factory/reviews/builder-latest.md b/.factory/reviews/builder-latest.md new file mode 100644 index 000000000..332b21500 --- /dev/null +++ b/.factory/reviews/builder-latest.md @@ -0,0 +1,26 @@ +# Builder Agent Output + +- **timestamp:** 2026-09-15 +- **exit_code:** 0 +- **branch:** factory/run-e3ddbac6 +- **pr:** #1494 (existing — pushed fixes to branch) + +## Changes + +### Fix A — `tests/test_compose.py` (path resolution) +- Added module-level constant `_CHESS_EVOLVE_TOML` using `Path(__file__).resolve().parent.parent / ...` to resolve the chess-evolve.toml path absolutely (stable under pytest-xdist `-n auto` where CWD differs from repo root) +- Updated both `test_chess_evolve_toml_no_builder_required` and `test_chess_evolve_toml_passes_any_workflow` to use the constant instead of the relative path string + +### Fix B — `factory/outer_loop/similarity.py` (NoveltyFilter GED=0 logic) +- Modified the GED loop in `NoveltyFilter.is_novel()` to `continue` when `ged == 0` (identical topology) +- Rationale: When GED=0, topology is identical to an archived workflow. If the structural hash check above already passed (hash is novel), the difference must be content-only (prompts, params). Content-only mutations are intentionally novel — exact duplicates are caught by the hash dedup. The GED loop should only reject when topology distance is non-zero but below threshold. + +## Verification + +- 3 previously-failing tests now pass: + - `test_chess_evolve_toml_no_builder_required` ✅ + - `test_chess_evolve_toml_passes_any_workflow` ✅ + - `test_prompt_only_mutation_passes_is_novel` ✅ +- Full test suites pass with no regressions: + - `tests/test_compose.py`: 51/51 passed + - `tests/test_outer_loop/test_similarity.py`: 18/18 passed diff --git a/factory/inner_loop.py b/factory/inner_loop.py index c1af74b84..9d30249bd 100644 --- a/factory/inner_loop.py +++ b/factory/inner_loop.py @@ -26,9 +26,14 @@ from pathlib import Path from typing import Any, Protocol, runtime_checkable +import structlog + +from factory.compose import IncompatibleCompositionError from factory.cycle_analyzer import CycleAnalyzer, CycleRecord from factory.workflow.primitives import DataNode, Workflow +log = structlog.get_logger() + @dataclass class EvalResult: @@ -272,6 +277,7 @@ def _step_with_task(self, directives: dict[str, Any] | None = None) -> CycleReco import asyncio import statistics + from factory.compose import validate_composition from factory.models import AggregateMethod, InnerLoopConfig from factory.workflow.executor import WorkflowExecutor @@ -281,6 +287,31 @@ def _step_with_task(self, directives: dict[str, Any] | None = None) -> CycleReco if self._workflow_has_data_node(): return self._step_with_data_node(directives) + # Belt-and-suspenders: catch post-mutation composition failures + # (e.g. NODE_REMOVE stripping the Builder after initial composition) + try: + validate_composition(self.workflow, self.task) + except IncompatibleCompositionError as exc: + log.warning( + "composition_incompatible", + workflow=getattr(self.workflow, "name", "unknown"), + task=getattr(self.task, "name", "unknown"), + error=str(exc), + ) + self._step_count += 1 + record = CycleRecord( + cycle_number=self._step_count, + mode=self.mode, + started_at=None, + ended_at=None, + duration_s=0.0, + score_start=None, + score_end=0.0, + score_delta=None, + ) + self._history.append(record) + return record + event_offset = self._count_lines(self.factory_dir / "events.jsonl") t0 = time.monotonic() diff --git a/factory/outer_loop/designer.py b/factory/outer_loop/designer.py index c31af5794..2727fdb43 100644 --- a/factory/outer_loop/designer.py +++ b/factory/outer_loop/designer.py @@ -15,9 +15,11 @@ from factory.workflow.primitives import ( AgentNode, AgentRole, + DataNode, Edge, FnNode, GateNode, + NodeType, Workflow, ) @@ -31,12 +33,17 @@ class DesignerAgent: Mutation mode proposes targeted mutations from failure telemetry. """ - def design_minimal(self, benchmark_spec: str) -> Workflow: + def design_minimal( + self, + benchmark_spec: str, + seed_workflow: Workflow | None = None, + frozen_node_ids: set[str] | None = None, + ) -> Workflow: """Create a 3-4 node workflow optimized for speed. Structure: researcher → builder → gate """ - nodes: dict[str, AgentNode | FnNode | GateNode] = { + nodes: dict[str, NodeType] = { "researcher": AgentNode( id="researcher", role=AgentRole.RESEARCHER, @@ -57,20 +64,36 @@ def design_minimal(self, benchmark_spec: str) -> Workflow: reads={".factory/reviews/builder-latest.md"}, ), } + + _inject_frozen_nodes(nodes, seed_workflow, frozen_node_ids) + edges = [ Edge(source="researcher", target="builder"), Edge(source="builder", target="gate_qa"), ] + + start_node = "researcher" + new_start = _rewire_data_nodes( + nodes, edges, start_node, seed_workflow, frozen_node_ids + ) + if new_start is not None: + start_node = new_start + wf = Workflow( name=f"minimal_{_slug(benchmark_spec)}", nodes=nodes, # type: ignore[arg-type] edges=edges, - start_node="researcher", + start_node=start_node, ) log.info("designed_minimal", nodes=len(wf.nodes), benchmark=benchmark_spec[:40]) return wf - def design_thorough(self, benchmark_spec: str) -> Workflow: + def design_thorough( + self, + benchmark_spec: str, + seed_workflow: Workflow | None = None, + frozen_node_ids: set[str] | None = None, + ) -> Workflow: """Create an 8-10 node workflow optimized for thoroughness. Structure: study → researcher → strategist → fork(builder_a, builder_b) @@ -78,7 +101,7 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: """ from factory.workflow.primitives import ForkNode, JoinNode - nodes: dict[str, AgentNode | FnNode | GateNode | ForkNode | JoinNode] = { + nodes: dict[str, NodeType] = { "study": FnNode( id="study", command="factory study {project_path}", @@ -142,6 +165,9 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: reads={".factory/reviews/adversarial-qa.md"}, ), } + + _inject_frozen_nodes(nodes, seed_workflow, frozen_node_ids) + edges = [ Edge(source="study", target="researcher"), Edge(source="researcher", target="strategist"), @@ -154,16 +180,30 @@ def design_thorough(self, benchmark_spec: str) -> Workflow: Edge(source="code_reviewer", target="adversarial_tester"), Edge(source="adversarial_tester", target="gate_qa"), ] + + start_node = "study" + new_start = _rewire_data_nodes( + nodes, edges, start_node, seed_workflow, frozen_node_ids + ) + if new_start is not None: + start_node = new_start + wf = Workflow( name=f"thorough_{_slug(benchmark_spec)}", nodes=nodes, # type: ignore[arg-type] edges=edges, - start_node="study", + start_node=start_node, ) log.info("designed_thorough", nodes=len(wf.nodes), benchmark=benchmark_spec[:40]) return wf - def design_custom(self, benchmark_spec: str, constraints: dict[str, object]) -> Workflow: + def design_custom( + self, + benchmark_spec: str, + constraints: dict[str, object], + seed_workflow: Workflow | None = None, + frozen_node_ids: set[str] | None = None, + ) -> Workflow: """Create a custom from-scratch workflow with optional constraints. Constraints can specify: @@ -176,7 +216,7 @@ def design_custom(self, benchmark_spec: str, constraints: dict[str, object]) -> raw_roles = constraints.get("require_roles", []) require_roles: list[object] = list(raw_roles) if isinstance(raw_roles, list) else [] - nodes: dict[str, AgentNode | FnNode | GateNode] = {} + nodes: dict[str, NodeType] = {} edges: list[Edge] = [] prev_id: str | None = None @@ -216,7 +256,15 @@ def design_custom(self, benchmark_spec: str, constraints: dict[str, object]) -> ) edges.append(Edge(source=prev_id, target=gate_id)) + _inject_frozen_nodes(nodes, seed_workflow, frozen_node_ids) + start = core_roles[0][0] if core_roles else "gate_qa" + new_start = _rewire_data_nodes( + nodes, edges, start, seed_workflow, frozen_node_ids + ) + if new_start is not None: + start = new_start + wf = Workflow( name=f"custom_{_slug(benchmark_spec)}", nodes=nodes, # type: ignore[arg-type] @@ -305,6 +353,99 @@ def propose( return proposals[:3] +def _inject_frozen_nodes( + nodes: dict[str, NodeType], + seed_workflow: Workflow | None, + frozen_node_ids: set[str] | None, +) -> None: + """Inject frozen nodes from a seed workflow into a template nodes dict. + + Frozen nodes take precedence over template nodes on ID collision. + """ + if not seed_workflow or not frozen_node_ids: + return + for frozen_id in frozen_node_ids: + if frozen_id in seed_workflow.nodes: + if frozen_id in nodes: + log.warning( + "frozen_node_collision", + node_id=frozen_id, + action="preferring_frozen_over_template", + ) + nodes[frozen_id] = seed_workflow.nodes[frozen_id] + else: + log.warning("frozen_node_missing_in_seed", node_id=frozen_id) + + +def _rewire_data_nodes( + nodes: dict[str, NodeType], + edges: list[Edge], + original_start: str, + seed_workflow: Workflow | None, + frozen_node_ids: set[str] | None, +) -> str | None: + """Rewire injected frozen DataNodes so they integrate into the template. + + For each frozen DataNode: + 1. Update subgraph_entry → template's original start_node + 2. Update subgraph_exit → template's terminal node (no outgoing edges) + + No explicit edge is added from the DataNode to subgraph_entry — the + executor reads subgraph_entry directly from the DataNode object. + Adding an explicit edge would fail validation (_validate_datanode_edges + rejects edges from a DataNode to its own subgraph nodes). + + Returns the DataNode ID (new start_node) or None if no DataNode was injected. + """ + if not seed_workflow or not frozen_node_ids: + return None + + # Find terminal node: the node with no outgoing edges (among template edges) + sources = {e.source for e in edges} + all_node_ids = set(nodes.keys()) + terminal_candidates = all_node_ids - sources + # Exclude the frozen DataNodes themselves from terminal candidates + frozen_data_ids: set[str] = set() + + for fid in frozen_node_ids: + node = nodes.get(fid) + if isinstance(node, DataNode): + frozen_data_ids.add(fid) + + if not frozen_data_ids: + return None + + terminal_candidates -= frozen_data_ids + terminal_node = next(iter(terminal_candidates)) if terminal_candidates else original_start + + new_start: str | None = None + for data_id in frozen_data_ids: + data_node = nodes[data_id] + assert isinstance(data_node, DataNode) + + # Determine subgraph entry: if the DataNode ID collides with the + # template's original_start, follow edges to find the actual first + # template node (otherwise subgraph_entry would point to itself). + # Also remove the now-stale edges from original_start — they would + # become invalid edges from the DataNode to its own subgraph. + entry = original_start + if data_id == original_start: + for edge in edges: + if edge.source == original_start: + entry = edge.target + break + edges[:] = [e for e in edges if e.source != original_start] + + # Replace with updated subgraph_entry/exit pointing to template nodes + updated = data_node.model_copy( + update={"subgraph_entry": entry, "subgraph_exit": terminal_node} + ) + nodes[data_id] = updated + new_start = data_id + + return new_start + + def extract_telemetry(eval_result: EvalResult) -> dict[str, object]: """Extract structured diagnostics from an EvalResult. diff --git a/factory/outer_loop/engine.py b/factory/outer_loop/engine.py index e7a96eceb..b34c22700 100644 --- a/factory/outer_loop/engine.py +++ b/factory/outer_loop/engine.py @@ -40,10 +40,27 @@ def _auto_frozen_nodes(workflow: Workflow) -> set[str]: - """Return node IDs that should always be frozen during mutation.""" - from factory.workflow.primitives import DataNode + """Return node IDs that should always be frozen during mutation. - return {nid for nid, node in workflow.nodes.items() if isinstance(node, DataNode)} + Includes DataNode IDs and all nodes in their subgraphs (entry→exit). + """ + from factory.workflow.primitives import DataNode + from factory.workflow.executor import _collect_subgraph_nodes + + frozen: set[str] = set() + for nid, node in workflow.nodes.items(): + if isinstance(node, DataNode): + frozen.add(nid) + subgraph_ids = _collect_subgraph_nodes( + workflow, node.subgraph_entry, node.subgraph_exit, + ) + frozen.update(subgraph_ids) + log.debug( + "data_node_subgraph_frozen", + data_node=nid, + subgraph_ids=list(subgraph_ids), + ) + return frozen class BudgetTracker: @@ -195,7 +212,7 @@ def seed( self._mode_registry.register(ind.id, 0, mutated_wf) if designer_count > 0: - self._add_designer_variants(pop, cfg, designer_count) + self._add_designer_variants(pop, cfg, designer_count, base_workflow) log.info( "population_seeded", @@ -210,21 +227,34 @@ def _add_designer_variants( pop: Population, cfg: SwarmConfig, designer_count: int, + seed_workflow: Workflow | None = None, ) -> None: """Add from-scratch designed workflows to the population.""" benchmark_spec = cfg.benchmark designs: list[Workflow] = [] + frozen_ids = set(cfg.frozen_node_ids) if cfg.frozen_node_ids else set() + if seed_workflow is not None: + frozen_ids |= _auto_frozen_nodes(seed_workflow) + frozen = frozen_ids if frozen_ids else None if designer_count >= 1: try: - minimal = self._designer.design_minimal(benchmark_spec) + minimal = self._designer.design_minimal( + benchmark_spec, + seed_workflow=seed_workflow, + frozen_node_ids=frozen, + ) designs.append(minimal) except Exception: log.warning("designer_minimal_failed", exc_info=True) if designer_count >= 2: try: - thorough = self._designer.design_thorough(benchmark_spec) + thorough = self._designer.design_thorough( + benchmark_spec, + seed_workflow=seed_workflow, + frozen_node_ids=frozen, + ) designs.append(thorough) except Exception: log.warning("designer_thorough_failed", exc_info=True) @@ -234,6 +264,8 @@ def _add_designer_variants( custom = self._designer.design_custom( benchmark_spec, {"max_nodes": 4 + i, "parallel": i % 2 == 0}, + seed_workflow=seed_workflow, + frozen_node_ids=frozen, ) designs.append(custom) except Exception: diff --git a/factory/outer_loop/similarity.py b/factory/outer_loop/similarity.py index 4a0b9a3ef..2713726d8 100644 --- a/factory/outer_loop/similarity.py +++ b/factory/outer_loop/similarity.py @@ -23,6 +23,10 @@ def structural_hash(workflow: Workflow) -> str: node = workflow.nodes[nid] d = node.model_dump(mode="json") d["_type"] = type(node).__name__ + if hasattr(node, "prompt_template"): + d["_prompt_hash"] = hashlib.sha256( + (getattr(node, "prompt_template", "") or "").encode() + ).hexdigest() nodes_canonical.append(d) edges_canonical = sorted( @@ -169,7 +173,14 @@ def is_novel(self, workflow: Workflow, threshold: int | None = None) -> bool: t = threshold if threshold is not None else self.min_edit_distance for archived in self._archived_workflows: - if archived.knob_values == workflow.knob_values and graph_edit_distance(workflow, archived) < t: + if archived.knob_values != workflow.knob_values: + continue + ged = graph_edit_distance(workflow, archived) + if ged == 0: + # Identical topology — if hash is novel (checked above), + # the difference is content-only (prompts, params) → novel + continue + if ged < t: return False return True diff --git a/factory/workflow/executor.py b/factory/workflow/executor.py index f27fc32e5..6acbf0aa7 100644 --- a/factory/workflow/executor.py +++ b/factory/workflow/executor.py @@ -7,6 +7,7 @@ import shlex import time import uuid +from collections.abc import Callable from pathlib import Path from typing import Any @@ -87,12 +88,19 @@ def __init__( dry_run: bool = False, auto_approve: bool = False, initial_context: str | None = None, + agent_fn: Callable[..., Any] | None = None, ) -> None: self.workflow = workflow self.project_path = project_path self.agent_pool = agent_pool or {} self.dry_run = dry_run self.auto_approve = auto_approve + if agent_fn is not None: + self._agent_fn = agent_fn + else: + from factory.agents.runner import invoke_agent + + self._agent_fn = invoke_agent self.run_id = uuid.uuid4().hex[:12] self.completed_files: set[str] = set() self.node_context: dict[str, str] = {} @@ -429,6 +437,13 @@ async def _execute_gate(self, node: GateNode) -> None: if target_id is None: target_id = self._next_unconditional(node_id) + if target_id is None: + log.warning( + "gate_proceed_edge_missing", + gate_id=node_id, + workflow=self.workflow.name, + ) + if target_id: await self._execute_from(target_id) @@ -581,6 +596,7 @@ async def run_branch(idx: int) -> dict[str, Any]: wt_path if not self.dry_run else self.project_path, agent_pool=self.agent_pool, dry_run=self.dry_run, + agent_fn=self._agent_fn, ) branch_result = await branch_executor.execute() @@ -841,6 +857,42 @@ async def run_item( prompt=resolved_task.prompt(inst), ) + # Diagnostic: list files created by setup() + _setup_files = [ + str(f.relative_to(item_project_path)) + for f in item_project_path.rglob('*') if f.is_file() + ] + log.info( + 'data_item_setup_complete', + item_id=item.id, + workspace=str(item_project_path), + files_found=len(_setup_files), + files=_setup_files[:20], + ) + + # Re-scan subgraph reads for files created by setup() + setup_reads: set[str] = set() + for sg_node in sub_workflow.nodes.values(): + for r in sg_node.reads: + if (item_project_path / r).exists(): + setup_reads.add(r) + else: + # Check if file exists under a different relative path + basename = Path(r).name + matches = [ + str(f.relative_to(item_project_path)) + for f in item_project_path.rglob(basename) + if f.is_file() + ] + if matches: + log.warning( + 'setup_read_path_mismatch', + node_id=sg_node.id, + declared_read=r, + found_at=matches, + hint='node.reads path does not match setup() output location', + ) + # Write current_item.json for subgraph visibility item_json_path = item_project_path / ".factory" / "current_item.json" item_json_path.parent.mkdir(parents=True, exist_ok=True) @@ -852,9 +904,10 @@ async def run_item( item_project_path, agent_pool=self.agent_pool, dry_run=self.dry_run, + agent_fn=self._agent_fn, initial_context=item.prompt or None, ) - item_executor.completed_files = self.completed_files | disk_reads + item_executor.completed_files = self.completed_files | disk_reads | setup_reads item_result = await item_executor.execute() finally: item_json_path.unlink(missing_ok=True) @@ -1126,8 +1179,6 @@ async def _run_fn(self, node: FnNode) -> str: async def _run_agent(self, node: AgentNode) -> str: """Invoke an agent via factory/agents/runner.py.""" - from factory.agents.runner import invoke_agent - task = node.prompt_template.replace( "{project_path}", str(self.project_path), ) @@ -1147,7 +1198,7 @@ async def _run_agent(self, node: AgentNode) -> str: if pool_entry: timeout = pool_entry.timeout - stdout, code = await invoke_agent( + stdout, code = await self._agent_fn( node.role.value, # type: ignore[arg-type] task, self.project_path, @@ -1163,6 +1214,13 @@ async def _run_agent(self, node: AgentNode) -> str: output_len=len(stdout), ) + # Persist output to node.writes paths (mirrors _run_llm pattern) + if node.writes: + for wpath in node.writes: + fpath = self.project_path / wpath + fpath.parent.mkdir(parents=True, exist_ok=True) + fpath.write_text(stdout) + return stdout async def _run_llm(self, node: LLMNode) -> str: @@ -1427,6 +1485,15 @@ async def _wait_for_reads(self, node: NodeType) -> None: if not missing: return if waited >= max_wait: + # Diagnostic: show what files exist vs what's expected + existing = sorted(self.completed_files) + log.warning( + 'wait_for_reads_timeout_diagnostic', + node_id=node.id, + missing_reads=sorted(missing), + completed_files_count=len(self.completed_files), + sample_completed=existing[:10], + ) self.result.halted = True self.result.halt_reason = ( f"node '{node.id}' timed out waiting for reads: {sorted(missing)}" diff --git a/factory/workflow/validation.py b/factory/workflow/validation.py index 42c736b20..1454a3363 100644 --- a/factory/workflow/validation.py +++ b/factory/workflow/validation.py @@ -150,6 +150,38 @@ def _validate_datanode_edges(workflow: Workflow, issues: list[str]) -> None: ) +def _validate_datanode_exit(workflow: Workflow, issues: list[str]) -> None: + """Warn when a DataNode's subgraph_exit points to a Loop GateNode. + + When subgraph_exit is a GateNode that participates in a Loop (has + outgoing RELOOP edges), _collect_subgraph_nodes stops BFS at the gate, + excluding the PROCEED edge target (the real exit node). + Workflow.subgraph() then drops the PROCEED edge, causing execution + to silently halt after one loop iteration. + + Terminal GateNodes (no RELOOP edges) are fine as subgraph_exit — they + don't have a PROCEED edge that would be dropped. + """ + from factory.workflow.primitives import VerdictType + + for nid, node in workflow.nodes.items(): + if type(node).__name__ != "DataNode": + continue + exit_id = node.subgraph_exit # type: ignore[union-attr] + exit_node = workflow.nodes.get(exit_id) + if exit_node is not None and type(exit_node).__name__ == "GateNode": + has_reloop = any( + e.source == exit_id and e.condition == VerdictType.RELOOP + for e in workflow.edges + ) + if not has_reloop: + continue + issues.append( + f"DataNode '{nid}' has subgraph_exit pointing to GateNode '{exit_id}'. " + f"This drops the PROCEED edge. Use the Loop's exit_node instead." + ) + + def _collect_subgraph_nodes( workflow: Workflow, entry: str, @@ -210,6 +242,7 @@ def validate_workflow(workflow: Workflow) -> list[str]: _validate_data_dependencies(g, workflow, issues) _validate_fork_join_nodes(workflow, issues) _validate_datanode_edges(workflow, issues) + _validate_datanode_exit(workflow, issues) for nid, node in nodes.items(): if type(node).__name__ == "SubgraphForkNode": diff --git a/tests/test_compose.py b/tests/test_compose.py index a4d84af26..f61b2fd62 100644 --- a/tests/test_compose.py +++ b/tests/test_compose.py @@ -25,6 +25,9 @@ ) +_CHESS_EVOLVE_TOML = Path(__file__).resolve().parent.parent / "benchmarks" / "configs" / "chess-evolve.toml" + + # ── Capability StrEnum tests ──────────────────────────────────── @@ -271,7 +274,7 @@ def test_chess_evolve_toml_no_builder_required(self): """chess-evolve.toml with required_capabilities=[] should need no capabilities.""" from factory.task import TaskDefinition - defn = TaskDefinition.from_toml("benchmarks/configs/chess-evolve.toml") + defn = TaskDefinition.from_toml(_CHESS_EVOLVE_TOML) assert defn.constraints.required_capabilities == [] task = Task(definition=defn) caps = TaskCapabilities.from_task(task) @@ -281,7 +284,7 @@ def test_chess_evolve_toml_passes_any_workflow(self): """chess-evolve.toml should pass composition with a research-only workflow.""" from factory.task import TaskDefinition - defn = TaskDefinition.from_toml("benchmarks/configs/chess-evolve.toml") + defn = TaskDefinition.from_toml(_CHESS_EVOLVE_TOML) task = Task(definition=defn) wf = _make_workflow(researcher=True, name="research-only") validate_composition(wf, task) @@ -543,3 +546,21 @@ def test_to_dict_from_dict_round_trip(self): d = compiled.to_dict() restored = Workflow.from_dict(d) assert restored.declared_capabilities == frozenset(["code-generation", "health-check"]) + + +class TestExplicitEmptyCapsValidation: + """Gap 0: Tasks with required_capabilities=[] pass any workflow.""" + + def test_explicit_empty_caps_passes_research_workflow(self): + """A task with TaskConstraints(required_capabilities=[]) and exit_code scoring + passes validate_composition() with a minimal workflow that has no builder node.""" + wf = _make_workflow(researcher=True, name="no-builder") + task = Task( + definition=TaskDefinition( + name="eval-only-task", + scoring=ScoringContract(method="exit_code"), + constraints=TaskConstraints(required_capabilities=[]), + ) + ) + # Should NOT raise IncompatibleCompositionError + validate_composition(wf, task) diff --git a/tests/test_data_node.py b/tests/test_data_node.py index 4d64e4da4..634966aba 100644 --- a/tests/test_data_node.py +++ b/tests/test_data_node.py @@ -1494,3 +1494,458 @@ def test_executor_failure_defaults_score(self, tmp_path: Path) -> None: record = loop._step_with_data_node() assert record.score_end == 0.0 + + +# ── disk_reads re-scan after setup() ──────────────────────────── + + +class _SetupWritingTask: + """Task whose setup() creates a file that a subgraph node reads.""" + + def __init__(self, setup_file: str) -> None: + self._setup_file = setup_file + + def instances(self): + from factory.task import TaskInstance + return [TaskInstance(id="inst1")] + + def setup(self, instance: Any, workspace: Path) -> None: + target = workspace / self._setup_file + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("setup content") + + def prompt(self, instance: Any) -> str: + return "go" + + def verify(self, instance: Any, workspace: Path): + from factory.task import VerifyResult + return VerifyResult(passed=True, score=1.0) + + +class TestDataNodeLoopSubgraph: + """DataNode + Loop/Gate subgraph integration tests.""" + + @pytest.mark.asyncio + async def test_data_node_with_loop_subgraph(self, tmp_path: Path) -> None: + """DataNode whose subgraph is a Loop should execute body 3 times via fn gate.""" + from factory.workflow.executor import WorkflowExecutor + from factory.workflow.package import Loop, Package + from factory.workflow.primitives import GateNode + + project_path = tmp_path + (project_path / ".factory").mkdir(parents=True, exist_ok=True) + counter_file = project_path / "counter.txt" + + pp = str(project_path) + + body_node = FnNode( + id="loop_body", + command=f"python3 -c \"open('{pp}/counter.txt','a').write('x\\n')\"", + reads=set(), + writes={"counter.txt"}, + ) + body_pkg = Package( + name="body", + graph=Workflow( + name="body_graph", + nodes={"loop_body": body_node}, + edges=[], + start_node="loop_body", + ), + entry_node="loop_body", + exit_node="loop_body", + ) + + gate = GateNode( + id="loop_gate", + evaluator_type="fn", + evaluator_command=( + f"python3 -c \"" + f"import pathlib; " + f"p=pathlib.Path('{pp}/counter.txt'); " + f"c=len(p.read_text().splitlines()) if p.exists() else 0; " + f"print('PROCEED' if c >= 3 else 'RELOOP: try again')" + f"\"" + ), + reads=set(), + ) + + loop_pkg = Loop(body_pkg, gate, max_iterations=10, name="test_loop") + + # Build DataNode workflow with correct subgraph_exit = loop exit_node + data_node = DataNode( + id="data_driver", + inline_items=[DataItem(id="game1", prompt="play")], + subgraph_entry=loop_pkg.entry_node, + subgraph_exit=loop_pkg.exit_node, + ) + + all_nodes: dict[str, Any] = {"data_driver": data_node} + for nid, node in loop_pkg.graph.nodes.items(): + all_nodes[nid] = node + + wf = Workflow( + name="loop_data_test", + nodes=all_nodes, + edges=list(loop_pkg.graph.edges), + start_node="data_driver", + ) + + # Validate graph — should have no issues + issues = wf.validate_graph() + assert not issues, f"Unexpected validation issues: {issues}" + + executor = WorkflowExecutor(wf, project_path, dry_run=False) + result = await executor.execute() + + assert result.success, f"Execution failed: {result.halt_reason}" + assert counter_file.exists(), "counter.txt should exist" + lines = counter_file.read_text().splitlines() + assert len(lines) == 3, f"Expected 3 lines, got {len(lines)}" + # Check inner executor nodes: 3 body + 3 gate + 1 exit = 7 + parsed = json.loads(result.node_outputs["data_driver"]) + assert len(parsed) == 1 + inner_nodes = parsed[0]["nodes_executed"] + assert inner_nodes >= 7, f"Expected >= 7 inner nodes, got {inner_nodes}" + + def test_data_node_loop_wrong_exit_warns(self) -> None: + """DataNode with subgraph_exit pointing to GateNode should produce a validation warning.""" + from factory.workflow.package import Loop, Package + from factory.workflow.primitives import GateNode + + body_node = FnNode( + id="loop_body", + command="echo body", + reads=set(), + writes={"counter.txt"}, + ) + body_pkg = Package( + name="body", + graph=Workflow( + name="body_graph", + nodes={"loop_body": body_node}, + edges=[], + start_node="loop_body", + ), + entry_node="loop_body", + exit_node="loop_body", + ) + + gate = GateNode( + id="loop_gate", + evaluator_type="fn", + evaluator_command="echo PROCEED", + reads=set(), + ) + + loop_pkg = Loop(body_pkg, gate, max_iterations=5, name="test_loop") + + # INCORRECT: subgraph_exit points to gate instead of exit_node + data_node = DataNode( + id="data_driver", + inline_items=[DataItem(id="game1", prompt="play")], + subgraph_entry=loop_pkg.entry_node, + subgraph_exit=gate.id, # WRONG — should be loop_pkg.exit_node + ) + + all_nodes: dict[str, Any] = {"data_driver": data_node} + for nid, node in loop_pkg.graph.nodes.items(): + all_nodes[nid] = node + + wf = Workflow( + name="wrong_exit_test", + nodes=all_nodes, + edges=list(loop_pkg.graph.edges), + start_node="data_driver", + ) + + issues = wf.validate_graph() + gate_warnings = [ + i for i in issues + if "GateNode" in i and "subgraph_exit" in i + ] + assert len(gate_warnings) >= 1, f"Expected GateNode warning, got: {issues}" + + def test_loop_package_compiled_preserves_edges(self) -> None: + """Loop Package compiled into a Workflow preserves all 3 loop edges in subgraph.""" + from factory.workflow.executor import _collect_subgraph_nodes + from factory.workflow.package import Loop, Package + from factory.workflow.primitives import GateNode, VerdictType + + body_node = FnNode( + id="loop_body", + command="echo body", + reads=set(), + ) + body_pkg = Package( + name="body", + graph=Workflow( + name="body_graph", + nodes={"loop_body": body_node}, + edges=[], + start_node="loop_body", + ), + entry_node="loop_body", + exit_node="loop_body", + ) + + gate = GateNode( + id="loop_gate", + evaluator_type="fn", + evaluator_command="echo PROCEED", + reads=set(), + ) + + loop_pkg = Loop(body_pkg, gate, max_iterations=5, name="test_loop") + + # Build DataNode with CORRECT exit_node + data_node = DataNode( + id="data_driver", + inline_items=[DataItem(id="game1", prompt="play")], + subgraph_entry=loop_pkg.entry_node, + subgraph_exit=loop_pkg.exit_node, + ) + + all_nodes: dict[str, Any] = {"data_driver": data_node} + for nid, node in loop_pkg.graph.nodes.items(): + all_nodes[nid] = node + + wf = Workflow( + name="edge_preservation_test", + nodes=all_nodes, + edges=list(loop_pkg.graph.edges), + start_node="data_driver", + ) + + # Collect subgraph nodes + subgraph_ids = _collect_subgraph_nodes( + wf, loop_pkg.entry_node, loop_pkg.exit_node, + ) + + # All 3 loop nodes + exit must be in subgraph + assert "loop_body" in subgraph_ids + assert "loop_gate" in subgraph_ids + assert loop_pkg.exit_node in subgraph_ids + + # Extract subgraph and check edges + sub_wf = wf.subgraph(subgraph_ids, name="sub", start_node=loop_pkg.entry_node) + + # Check all 3 loop edges are preserved + edge_tuples = [(e.source, e.target, e.condition) for e in sub_wf.edges] + + # body → gate (unconditional) + assert ("loop_body", "loop_gate", None) in edge_tuples, ( + f"Missing body→gate edge. Edges: {edge_tuples}" + ) + # gate → body (RELOOP) + assert ("loop_gate", "loop_body", VerdictType.RELOOP) in edge_tuples, ( + f"Missing gate→body RELOOP edge. Edges: {edge_tuples}" + ) + # gate → exit (PROCEED) + assert ("loop_gate", loop_pkg.exit_node, VerdictType.PROCEED) in edge_tuples, ( + f"Missing gate→exit PROCEED edge. Edges: {edge_tuples}" + ) + + +class TestDiskReadsRescanAfterSetup: + """setup()-created files must appear in sub-executor completed_files.""" + + def test_setup_created_file_in_completed_files(self, tmp_path: Path) -> None: + """When task.setup() writes a file declared in a subgraph node's reads, + the sub-executor's completed_files must include it so _wait_for_reads() + doesn't block for 60 s.""" + from factory.workflow.executor import WorkflowExecutor + + setup_file = "data/input.txt" + + wf = Workflow( + name="rescan_test", + nodes={ + "data": DataNode( + id="data", + task_ref="fake.module:SetupWritingTask", + subgraph_entry="reader", + subgraph_exit="reader", + ), + "reader": FnNode( + id="reader", + command="echo ok", + reads={setup_file}, + ), + }, + edges=[], + start_node="data", + ) + + fake_task = _SetupWritingTask(setup_file) + + # Capture the completed_files set on the sub-executor + captured_completed: list[set[str]] = [] + original_execute = WorkflowExecutor.execute + + async def spy_execute(self_inner): + if self_inner.workflow.name.endswith("__data_item"): + captured_completed.append(set(self_inner.completed_files)) + return await original_execute(self_inner) + + with patch("factory.task.TaskRef.resolve", return_value=fake_task), \ + patch.object(WorkflowExecutor, "execute", spy_execute): + executor = WorkflowExecutor(wf, tmp_path, dry_run=True) + result = asyncio.run(executor.execute()) + + assert result.success, f"halted: {result.halt_reason}" + # The sub-executor's completed_files must contain the setup-written file + assert len(captured_completed) == 1 + assert setup_file in captured_completed[0] + + @pytest.mark.asyncio + async def test_data_node_loop_with_agent_body(self, tmp_path: Path) -> None: + """DataNode + Loop where body is an AgentNode — loop should iterate via RELOOP.""" + from factory.workflow.executor import WorkflowExecutor + from factory.workflow.package import Loop, Package + from factory.workflow.primitives import AgentNode, AgentRole, GateNode + + project_path = tmp_path + (project_path / '.factory').mkdir(parents=True, exist_ok=True) + counter_file = project_path / 'counter.txt' + pp = str(project_path) + + # Agent body: mock agent_fn that appends to counter file + call_count = 0 + + async def mock_agent_fn(role, task, proj_path, **kwargs): + nonlocal call_count + call_count += 1 + cf = Path(proj_path) / 'counter.txt' + cf.parent.mkdir(parents=True, exist_ok=True) + with open(cf, 'a') as f: + f.write(f'move {call_count}\n') + return (f'Generated move {call_count}', 0) + + generator = AgentNode( + id='generator', + role=AgentRole.BUILDER, + prompt_template='Generate the next move', + reads=set(), + writes=set(), # mock_agent_fn handles file I/O directly + timeout=30, + ) + body_pkg = Package( + name='gen_body', + graph=Workflow( + name='gen_graph', + nodes={'generator': generator}, + edges=[], + start_node='generator', + ), + entry_node='generator', + exit_node='generator', + ) + + gate = GateNode( + id='game_gate', + evaluator_type='fn', + evaluator_command=( + f"python3 -c \"" + f"import pathlib; " + f"p=pathlib.Path('{pp}/counter.txt'); " + f"c=len(p.read_text().splitlines()) if p.exists() else 0; " + f"print('PROCEED' if c >= 3 else 'RELOOP: keep playing')" + f"\"" + ), + reads=set(), + ) + + loop_pkg = Loop(body_pkg, gate, max_iterations=10, name='game_loop') + + data_node = DataNode( + id='game_data', + inline_items=[DataItem(id='game1', prompt='Play chess')], + subgraph_entry=loop_pkg.entry_node, + subgraph_exit=loop_pkg.exit_node, + ) + + all_nodes: dict[str, Any] = {'game_data': data_node} + for nid, node in loop_pkg.graph.nodes.items(): + all_nodes[nid] = node + + wf = Workflow( + name='agent_loop_test', + nodes=all_nodes, + edges=list(loop_pkg.graph.edges), + start_node='game_data', + ) + + issues = wf.validate_graph() + assert not issues, f'Validation issues: {issues}' + + executor = WorkflowExecutor(wf, project_path, agent_fn=mock_agent_fn) + result = await executor.execute() + + assert result.success, f'Execution failed: {result.halt_reason}' + assert counter_file.exists(), 'counter.txt should exist' + lines = counter_file.read_text().strip().splitlines() + assert len(lines) == 3, f'Expected 3 lines (3 iterations), got {len(lines)}: {lines}' + assert call_count == 3, f'Expected agent called 3 times, got {call_count}' + + @pytest.mark.asyncio + async def test_setup_read_path_mismatch_logs_warning(self, tmp_path: Path) -> None: + """When setup() creates a file at a different path than node.reads expects, + the reader should timeout waiting for the mismatched read path.""" + from factory.workflow.executor import WorkflowExecutor + + project_path = tmp_path + (project_path / '.factory').mkdir(parents=True, exist_ok=True) + + # Create a file at .factory/memory.md but declare reads as 'memory.md' + (project_path / '.factory' / 'memory.md').write_text('game state') + + wf = Workflow( + name='mismatch_test', + nodes={ + 'data': DataNode( + id='data', + inline_items=[DataItem(id='item1', prompt='test')], + subgraph_entry='reader', + subgraph_exit='reader', + ), + 'reader': FnNode( + id='reader', + command='echo ok', + reads={'memory.md'}, # WRONG — file is at .factory/memory.md + ), + }, + edges=[], + start_node='data', + ) + + # Patch max_wait to avoid 60s timeout in CI + async def fast_wait(self_inner, node): + # Reduced max_wait for test speed + poll_interval = 0.1 + waited = 0.0 + while True: + missing = node.reads - self_inner.completed_files + if not missing: + return + if waited >= 0.3: + self_inner.result.halted = True + self_inner.result.halt_reason = ( + f"node '{node.id}' timed out waiting for reads: {sorted(missing)}" + ) + return + await asyncio.sleep(poll_interval) + waited += poll_interval + + with patch.object(WorkflowExecutor, '_wait_for_reads', fast_wait): + executor = WorkflowExecutor(wf, project_path, dry_run=False) + result = await executor.execute() + + # DataNode fault-isolates: outer succeeds but inner item fails due to read timeout + assert result.success + parsed = json.loads(result.node_outputs['data']) + assert len(parsed) == 1 + item_result = parsed[0] + # Inner executor halted waiting for 'memory.md' that doesn't exist at that path + assert not item_result['success'] + assert item_result['nodes_executed'] == 0 # reader never ran diff --git a/tests/test_outer_loop/test_designer.py b/tests/test_outer_loop/test_designer.py index ef0e8a97a..e8ff9d152 100644 --- a/tests/test_outer_loop/test_designer.py +++ b/tests/test_outer_loop/test_designer.py @@ -4,6 +4,15 @@ from factory.outer_loop.designer import DesignerAgent from factory.outer_loop.models import MutationType +from factory.workflow.primitives import ( + AgentNode, + AgentRole, + DataItem, + DataNode, + Edge, + FnNode, + Workflow, +) class TestDesignMinimal: @@ -221,3 +230,319 @@ def test_max_3_proposals(self, simple_workflow) -> None: # type: ignore[no-unty benchmark_spec="test", ) assert len(proposals) <= 3 + + +class TestFrozenNodePreservation: + """Tests for frozen node injection in designer methods.""" + + @staticmethod + def _seed_with_positions() -> Workflow: + """Create a seed workflow containing a FnNode with id='positions'.""" + return Workflow( + name="seed", + nodes={ + "positions": FnNode( + id="positions", + command="load_positions", + writes={".factory/positions.json"}, + ), + "researcher": AgentNode( + id="researcher", + role=AgentRole.RESEARCHER, + ), + }, + edges=[Edge(source="positions", target="researcher")], + start_node="positions", + ) + + def test_design_minimal_preserves_frozen_nodes(self) -> None: + designer = DesignerAgent() + seed = self._seed_with_positions() + result = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert "positions" in result.nodes + assert result.nodes["positions"].command == "load_positions" # type: ignore[union-attr] + + def test_design_thorough_preserves_frozen_nodes(self) -> None: + designer = DesignerAgent() + seed = self._seed_with_positions() + result = designer.design_thorough( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert "positions" in result.nodes + assert result.nodes["positions"].command == "load_positions" # type: ignore[union-attr] + + def test_design_custom_preserves_frozen_nodes(self) -> None: + designer = DesignerAgent() + seed = self._seed_with_positions() + result = designer.design_custom( + "bench", + {"max_nodes": 6}, + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert "positions" in result.nodes + assert result.nodes["positions"].command == "load_positions" # type: ignore[union-attr] + + def test_frozen_node_overwrites_template_on_collision(self) -> None: + """When a frozen node ID collides with a template node, frozen wins.""" + seed = Workflow( + name="seed", + nodes={ + "researcher": AgentNode( + id="researcher", + role=AgentRole.RESEARCHER, + timeout=999, + ), + }, + edges=[], + start_node="researcher", + ) + designer = DesignerAgent() + result = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"researcher"}, + ) + assert result.nodes["researcher"].timeout == 999 # type: ignore[union-attr] + + def test_design_without_frozen_nodes_unchanged(self) -> None: + """Calling design_minimal() without seed/frozen params works as before.""" + designer = DesignerAgent() + wf = designer.design_minimal("test benchmark") + assert 3 <= len(wf.nodes) <= 4 + issues = wf.validate_graph() + assert issues == [], f"Validation issues: {issues}" + + def test_design_minimal_preserves_frozen_data_node(self) -> None: + """DataNode auto-frozen via _auto_frozen_nodes should be preserved.""" + from factory.workflow.primitives import DataItem, DataNode + + designer = DesignerAgent() + seed = Workflow( + name="seed", + nodes={ + "positions": DataNode( + id="positions", + inline_items=[DataItem(id="pos1", prompt="test")], + subgraph_entry="solver", + subgraph_exit="solver", + ), + "solver": AgentNode( + id="solver", + role=AgentRole.BUILDER, + ), + }, + edges=[Edge(source="positions", target="solver")], + start_node="positions", + ) + result = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert "positions" in result.nodes + assert type(result.nodes["positions"]).__name__ == "DataNode" + + def test_engine_designer_includes_auto_frozen_data_nodes(self) -> None: + """_add_designer_variants should include auto-frozen DataNodes.""" + from factory.outer_loop.engine import _auto_frozen_nodes + from factory.workflow.primitives import DataItem, DataNode + + seed = Workflow( + name="seed", + nodes={ + "positions": DataNode( + id="positions", + inline_items=[DataItem(id="pos1", prompt="test")], + subgraph_entry="solver", + subgraph_exit="solver", + ), + "solver": AgentNode( + id="solver", + role=AgentRole.BUILDER, + ), + }, + edges=[Edge(source="positions", target="solver")], + start_node="positions", + ) + + # Verify _auto_frozen_nodes detects the DataNode + auto_frozen = _auto_frozen_nodes(seed) + assert "positions" in auto_frozen + + +class TestDataNodeRewiring: + """Tests that frozen DataNodes are properly wired into designer templates.""" + + @staticmethod + def _seed_with_data_node() -> Workflow: + """Seed workflow containing a DataNode with subgraph refs to 'solver'.""" + return Workflow( + name="seed", + nodes={ + "positions": DataNode( + id="positions", + inline_items=[DataItem(id="pos1", prompt="test")], + subgraph_entry="solver", + subgraph_exit="solver", + ), + "solver": AgentNode( + id="solver", + role=AgentRole.BUILDER, + ), + }, + edges=[Edge(source="positions", target="solver")], + start_node="positions", + ) + + def test_minimal_start_node_is_data_node(self) -> None: + """Designer variant with DataNode has start_node == DataNode ID.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert wf.start_node == "positions" + + def test_minimal_subgraph_entry_points_to_template_start(self) -> None: + """DataNode.subgraph_entry points to the template's original start.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + data_node = wf.nodes["positions"] + assert isinstance(data_node, DataNode) + assert data_node.subgraph_entry == "researcher" + + def test_minimal_subgraph_exit_points_to_terminal(self) -> None: + """DataNode.subgraph_exit points to template's terminal node.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + data_node = wf.nodes["positions"] + assert isinstance(data_node, DataNode) + assert data_node.subgraph_exit == "gate_qa" + + def test_minimal_no_explicit_edge_from_data_node_to_entry(self) -> None: + """No explicit edge from DataNode to subgraph_entry (executor uses subgraph_entry directly).""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + edge_pairs = [(e.source, e.target) for e in wf.edges] + assert ("positions", "researcher") not in edge_pairs + + def test_minimal_without_data_node_unchanged(self) -> None: + """Designer without frozen DataNode retains original start_node.""" + designer = DesignerAgent() + wf = designer.design_minimal("bench") + assert wf.start_node == "researcher" + edge_sources = {e.source for e in wf.edges} + assert "positions" not in edge_sources + + def test_thorough_start_node_is_data_node(self) -> None: + """design_thorough variant with DataNode has start_node == DataNode ID.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_thorough( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert wf.start_node == "positions" + data_node = wf.nodes["positions"] + assert isinstance(data_node, DataNode) + assert data_node.subgraph_entry == "study" + assert data_node.subgraph_exit == "gate_qa" + # No explicit edge from DataNode to subgraph_entry + edge_pairs = [(e.source, e.target) for e in wf.edges] + assert ("positions", "study") not in edge_pairs + + def test_custom_start_node_is_data_node(self) -> None: + """design_custom variant with DataNode has start_node == DataNode ID.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_custom( + "bench", + {"max_nodes": 6}, + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + assert wf.start_node == "positions" + data_node = wf.nodes["positions"] + assert isinstance(data_node, DataNode) + assert data_node.subgraph_entry == "researcher" + assert data_node.subgraph_exit == "gate_qa" + # No explicit edge from DataNode to subgraph_entry + edge_pairs = [(e.source, e.target) for e in wf.edges] + assert ("positions", "researcher") not in edge_pairs + + def test_rewired_workflow_validates_graph(self) -> None: + """Rewired workflow with DataNode passes validate_graph() without issues.""" + designer = DesignerAgent() + seed = self._seed_with_data_node() + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"positions"}, + ) + issues = wf.validate_graph() + assert issues == [], f"Validation issues: {issues}" + + def test_data_node_id_collision_with_start(self) -> None: + """DataNode ID == template start_node must not create self-referential subgraph_entry.""" + designer = DesignerAgent() + # Create a seed where the DataNode ID is 'researcher' — same as + # the minimal template's start_node. + seed = Workflow( + name="seed", + nodes={ + "researcher": DataNode( + id="researcher", + inline_items=[DataItem(id="pos1", prompt="test")], + subgraph_entry="solver", + subgraph_exit="solver", + ), + "solver": AgentNode( + id="solver", + role=AgentRole.BUILDER, + ), + }, + edges=[Edge(source="researcher", target="solver")], + start_node="researcher", + ) + wf = designer.design_minimal( + "bench", + seed_workflow=seed, + frozen_node_ids={"researcher"}, + ) + data_node = wf.nodes["researcher"] + assert isinstance(data_node, DataNode) + # subgraph_entry must NOT be 'researcher' (self-reference) + assert data_node.subgraph_entry != "researcher" + # It should point to the first node reachable from original start via edges + assert data_node.subgraph_entry == "builder" + # No structural issues (cycle, double-execution edge, unreachable). + # Data dependency warnings are expected since the DataNode replaced + # the researcher that would normally write the file. + issues = wf.validate_graph() + structural = [i for i in issues if "no predecessor writes" not in i] + assert structural == [], f"Structural issues: {structural}" diff --git a/tests/test_outer_loop/test_mutations.py b/tests/test_outer_loop/test_mutations.py index bdc4c70c8..757a67970 100644 --- a/tests/test_outer_loop/test_mutations.py +++ b/tests/test_outer_loop/test_mutations.py @@ -960,7 +960,7 @@ def test_auto_frozen_nodes_returns_data_node_ids(self) -> None: ] wf = Workflow(name="with_data", nodes=nodes, edges=edges, start_node="study") frozen = _auto_frozen_nodes(wf) - assert frozen == {"data_loader"} + assert frozen == {"data_loader", "builder"} def test_auto_frozen_nodes_empty_when_no_data_nodes(self) -> None: from factory.outer_loop.engine import _auto_frozen_nodes @@ -1012,3 +1012,78 @@ def test_data_node_protected_from_direct_removal(self) -> None: assert mutate_params( wf, "data_loader", {"timeout": 999}, frozen_nodes=frozen, ) is None + + def test_auto_frozen_nodes_includes_multi_node_subgraph(self) -> None: + """Subgraph spanning multiple nodes is fully frozen.""" + from factory.outer_loop.engine import _auto_frozen_nodes + from factory.workflow.primitives import DataNode, DataItem + + nodes: dict[str, AgentNode | FnNode | DataNode] = { + "data_loader": DataNode( + id="data_loader", + inline_items=[DataItem(id="item1", prompt="test")], + subgraph_entry="sub_entry", + subgraph_exit="sub_exit", + ), + "sub_entry": AgentNode(id="sub_entry", role=AgentRole.BUILDER), + "sub_mid": FnNode(id="sub_mid", command="echo mid"), + "sub_exit": AgentNode(id="sub_exit", role=AgentRole.CODE_REVIEWER), + "external": FnNode(id="external", command="echo external"), + } + edges = [ + Edge(source="external", target="data_loader"), + Edge(source="data_loader", target="sub_entry"), + Edge(source="sub_entry", target="sub_mid"), + Edge(source="sub_mid", target="sub_exit"), + ] + wf = Workflow( + name="multi_subgraph", nodes=nodes, edges=edges, start_node="external", + ) + frozen = _auto_frozen_nodes(wf) + # DataNode + all subgraph nodes frozen + assert frozen == {"data_loader", "sub_entry", "sub_mid", "sub_exit"} + # External node is NOT frozen + assert "external" not in frozen + + def test_subgraph_node_protected_from_removal(self) -> None: + """Subgraph nodes auto-frozen via DataNode cannot be removed.""" + from factory.outer_loop.engine import _auto_frozen_nodes + from factory.workflow.primitives import DataNode, DataItem + + nodes: dict[str, AgentNode | FnNode | DataNode] = { + "start": FnNode(id="start", command="echo start"), + "data_loader": DataNode( + id="data_loader", + inline_items=[DataItem(id="item1", prompt="test")], + subgraph_entry="sub_builder", + subgraph_exit="sub_builder", + ), + "sub_builder": AgentNode(id="sub_builder", role=AgentRole.BUILDER), + } + edges = [ + Edge(source="start", target="data_loader"), + Edge(source="data_loader", target="sub_builder"), + ] + wf = Workflow( + name="protected_subgraph", nodes=nodes, edges=edges, start_node="start", + ) + + # With DataNode present, sub_builder is auto-frozen + frozen = _auto_frozen_nodes(wf) + assert "sub_builder" in frozen + assert remove_node(wf, "sub_builder", frozen_nodes=frozen) is None + + # Without DataNode, sub_builder is NOT frozen and can be removed + nodes_no_data: dict[str, AgentNode | FnNode] = { + "start": FnNode(id="start", command="echo start"), + "sub_builder": AgentNode(id="sub_builder", role=AgentRole.BUILDER), + } + edges_no_data = [Edge(source="start", target="sub_builder")] + wf_no_data = Workflow( + name="no_data", nodes=nodes_no_data, edges=edges_no_data, + start_node="start", + ) + frozen_no_data = _auto_frozen_nodes(wf_no_data) + assert "sub_builder" not in frozen_no_data + result = remove_node(wf_no_data, "sub_builder", frozen_nodes=frozen_no_data) + assert result is not None diff --git a/tests/test_outer_loop/test_similarity.py b/tests/test_outer_loop/test_similarity.py index d01b54e68..214f4f2ea 100644 --- a/tests/test_outer_loop/test_similarity.py +++ b/tests/test_outer_loop/test_similarity.py @@ -35,6 +35,52 @@ def test_different_workflows_different_hash(self, simple_workflow: Workflow) -> ) assert structural_hash(simple_workflow) != structural_hash(other) + def test_prompt_change_affects_structural_hash(self) -> None: + """Two workflows identical except prompt_template must hash differently.""" + wf1 = Workflow( + name="w", + nodes={ + "a": AgentNode( + id="a", role=AgentRole.RESEARCHER, prompt_template="analyze code" + ), + }, + edges=[], + start_node="a", + ) + wf2 = Workflow( + name="w", + nodes={ + "a": AgentNode( + id="a", role=AgentRole.RESEARCHER, prompt_template="review code" + ), + }, + edges=[], + start_node="a", + ) + assert structural_hash(wf1) != structural_hash(wf2) + + def test_empty_prompt_stable_hash(self) -> None: + """Workflow with prompt_template='' and default (also '') hash the same.""" + wf1 = Workflow( + name="w", + nodes={ + "a": AgentNode( + id="a", role=AgentRole.RESEARCHER, prompt_template="" + ), + }, + edges=[], + start_node="a", + ) + wf2 = Workflow( + name="w", + nodes={ + "a": AgentNode(id="a", role=AgentRole.RESEARCHER), + }, + edges=[], + start_node="a", + ) + assert structural_hash(wf1) == structural_hash(wf2) + def test_same_structure_same_hash(self) -> None: nodes1 = { "a": FnNode(id="a", command="echo a"), @@ -198,6 +244,36 @@ def test_very_different_workflow_is_novel(self, simple_workflow: Workflow) -> No ) assert nf.is_novel(other) is True + def test_prompt_only_mutation_passes_is_novel(self) -> None: + """Prompt-only mutation should be considered novel.""" + wf1 = Workflow( + name="w", + nodes={ + "a": AgentNode( + id="a", + role=AgentRole.RESEARCHER, + prompt_template="analyze code", + ), + }, + edges=[], + start_node="a", + ) + wf2 = Workflow( + name="w", + nodes={ + "a": AgentNode( + id="a", + role=AgentRole.RESEARCHER, + prompt_template="review bugs", + ), + }, + edges=[], + start_node="a", + ) + nf = NoveltyFilter(min_edit_distance=5) + nf.add(wf1) + assert nf.is_novel(wf2) is True + def test_custom_threshold(self, simple_workflow: Workflow) -> None: nf = NoveltyFilter(min_edit_distance=100) nf.add(simple_workflow) diff --git a/tests/test_workflow_executor.py b/tests/test_workflow_executor.py index 7744ecfc9..16cfb1220 100644 --- a/tests/test_workflow_executor.py +++ b/tests/test_workflow_executor.py @@ -729,3 +729,144 @@ def test_no_initial_context_leaves_node_context_empty(self, tmp_project: Path) - executor = WorkflowExecutor(wf, tmp_project, dry_run=True) assert executor.node_context == {} + + +class TestRunAgentPersistsToNodeWrites: + """Gap 1: _run_agent() persists stdout to node.writes paths.""" + + async def test_run_agent_persists_to_node_writes(self, tmp_project: Path) -> None: + """AgentNode with writes={'output.md'} persists agent stdout to that file.""" + from unittest.mock import AsyncMock + + wf = Workflow( + name="persist_test", + nodes={ + "agent": AgentNode( + id="agent", + role=AgentRole.BUILDER, + prompt_template="build", + writes={"output.md"}, + ), + }, + edges=[], + start_node="agent", + ) + + mock_agent_fn = AsyncMock(return_value=("agent output", 0)) + executor = WorkflowExecutor(wf, tmp_project, agent_fn=mock_agent_fn) + await executor.execute() + + output_file = tmp_project / "output.md" + assert output_file.exists() + assert output_file.read_text() == "agent output" + + async def test_run_agent_no_writes_skips_file_creation(self, tmp_project: Path) -> None: + """AgentNode without writes doesn't create extra files.""" + from unittest.mock import AsyncMock + + wf = Workflow( + name="no_writes_test", + nodes={ + "agent": AgentNode( + id="agent", + role=AgentRole.BUILDER, + prompt_template="build", + ), + }, + edges=[], + start_node="agent", + ) + + files_before = set(tmp_project.rglob("*")) + mock_agent_fn = AsyncMock(return_value=("agent output", 0)) + executor = WorkflowExecutor(wf, tmp_project, agent_fn=mock_agent_fn) + await executor.execute() + + files_after = set(tmp_project.rglob("*")) + new_files = files_after - files_before + # Only event log files should be created, not agent output files + for f in new_files: + assert "output.md" not in f.name + + +class TestAgentFnInjection: + """Gap 2: WorkflowExecutor supports agent_fn injection.""" + + async def test_custom_agent_fn_used(self, tmp_project: Path) -> None: + """Custom agent_fn is called instead of default invoke_agent.""" + from unittest.mock import AsyncMock + + mock_fn = AsyncMock(return_value=("custom output", 0)) + + wf = Workflow( + name="custom_fn_test", + nodes={ + "agent": AgentNode( + id="agent", + role=AgentRole.BUILDER, + prompt_template="build", + ), + }, + edges=[], + start_node="agent", + ) + + executor = WorkflowExecutor(wf, tmp_project, agent_fn=mock_fn) + await executor.execute() + + mock_fn.assert_called_once() + + async def test_agent_fn_propagates_to_data_node_sub_executor( + self, tmp_project: Path, + ) -> None: + """agent_fn propagates to DataNode per-item sub-executors.""" + from unittest.mock import AsyncMock + + from factory.workflow.primitives import DataItem, DataNode + + mock_fn = AsyncMock(return_value=("sub output", 0)) + + wf = Workflow( + name="data_propagation_test", + nodes={ + "data": DataNode( + id="data", + inline_items=[DataItem(id="item1", prompt="do it")], + subgraph_entry="sub_agent", + subgraph_exit="sub_agent", + ), + "sub_agent": AgentNode( + id="sub_agent", + role=AgentRole.BUILDER, + prompt_template="build", + ), + }, + edges=[], + start_node="data", + ) + + executor = WorkflowExecutor(wf, tmp_project, agent_fn=mock_fn) + result = await executor.execute() + + assert result.success + mock_fn.assert_called_once() + + def test_agent_fn_defaults_to_invoke_agent(self, tmp_project: Path) -> None: + """When agent_fn is not provided, defaults to invoke_agent.""" + from factory.agents.runner import invoke_agent + + wf = Workflow( + name="default_fn_test", + nodes={ + "agent": AgentNode( + id="agent", + role=AgentRole.BUILDER, + prompt_template="build", + ), + }, + edges=[], + start_node="agent", + ) + + executor = WorkflowExecutor(wf, tmp_project) + assert executor._agent_fn is invoke_agent