feat: replace OptimizationLoop with optimize workflow graph

factory/cli/__init__.py

@@ -48,6 +48,12 @@
 from factory.cli.optimize import (
     cmd_optimize as cmd_optimize,
 )
+from factory.cli.optimize_step import (
+    cmd_optimize_step_apply_patch as cmd_optimize_step_apply_patch,
+    cmd_optimize_step_check_gate as cmd_optimize_step_check_gate,
+    cmd_optimize_step_run_dev as cmd_optimize_step_run_dev,
+    cmd_optimize_step_run_test as cmd_optimize_step_run_test,
+)
 from factory.cli.skillopt import (
     cmd_skillopt as cmd_skillopt,
 )

factory/cli/_main.py

@@ -105,7 +105,7 @@
             "backfill-archive",
         ],
     ),
-    ("Self-Evolution", ["ace", "ace-stats", "digest", "workflow", "graph", "skillopt", "optimize"]),
+    ("Self-Evolution", ["ace", "ace-stats", "digest", "workflow", "graph", "skillopt", "optimize", "optimize-step"]),
     (
         "Configuration",
         [
@@ -301,6 +301,17 @@ def main(argv: list[str] | None = None) -> int:
         "agent": _cli.cmd_agent,
         "skillopt": _cli.cmd_skillopt,
         "optimize": _cli.cmd_optimize,
+        "optimize-step": lambda a: {
+            "run-dev": _cli.cmd_optimize_step_run_dev,
+            "run-test": _cli.cmd_optimize_step_run_test,
+            "apply-patch": _cli.cmd_optimize_step_apply_patch,
+            "check-gate": _cli.cmd_optimize_step_check_gate,
+        }.get(
+            str(getattr(a, "optimize_step_command", "")),
+            lambda args: (
+                print("Usage: factory optimize-step {run-dev,run-test,apply-patch,check-gate}") or 1
+            ),
+        )(a),
         "ceo": _cli.cmd_ceo,
         "run": _cli.cmd_run,
         "tmux": _cli.cmd_tmux,

factory/cli/_parser_groups.py

@@ -253,6 +253,19 @@ def add_self_evolution_parsers(sub: argparse._SubParsersAction) -> None:  # type
     p.add_argument("--epochs", type=int, default=1, help="Number of training epochs")
     p.add_argument("--steps-per-epoch", type=int, default=1, help="Steps per epoch")
 
+    opt_step_parser = sub.add_parser("optimize-step", help="Workflow node helpers for optimize graph")
+    opt_step_sub = opt_step_parser.add_subparsers(dest="optimize_step_command")
+    p_rd = opt_step_sub.add_parser("run-dev", help="Run benchmark dev split")
+    p_rd.add_argument("--project", required=True, help="Path to the project")
+    p_rt = opt_step_sub.add_parser("run-test", help="Run benchmark test split")
+    p_rt.add_argument("--project", required=True, help="Path to the project")
+    p_ap = opt_step_sub.add_parser("apply-patch", help="Apply mutation rules to current skill")
+    p_ap.add_argument("--project", required=True, help="Path to the project")
+    p_cg = opt_step_sub.add_parser("check-gate", help="Check optimization gate verdict")
+    p_cg.add_argument("--project", required=True, help="Path to the project")
+    p_cg.add_argument("--baseline", action="store_true", default=False,
+                       help="Check baseline score instead of improvement gate")
+
     p = sub.add_parser("optimize", help="Run inner-outer optimization loop with HarborBenchmark")
     p.add_argument("path", help="Path to the project")
     p.add_argument("--benchmark", default="searchqa", choices=["searchqa", "featurebench", "auto"],
@@ -268,6 +281,8 @@ def add_self_evolution_parsers(sub: argparse._SubParsersAction) -> None:  # type
     p.add_argument("--model", default="sonnet", help="Model for AgenticMutator (default: sonnet)")
     p.add_argument("--split-seed", type=int, default=42, help="Seed for reproducible split generation (default: 42)")
     p.add_argument("--splits-dir", default=None, help="Path to pre-generated JSONL split files")
+    p.add_argument("--legacy", action="store_true", default=False,
+                    help="Use legacy OptimizationLoop instead of workflow graph")
 
 
 def add_configuration_parsers(sub: argparse._SubParsersAction) -> None:  # type: ignore[type-arg]

factory/cli/optimize.py

@@ -3,18 +3,20 @@
 from __future__ import annotations
 
 import argparse
+import json
 import os
+import subprocess
 import sys
 from pathlib import Path
+from typing import Any
 
+import structlog
+
+log = structlog.get_logger()
 
-def cmd_optimize(args: argparse.Namespace) -> int:
-    """Run the optimization loop with HarborBenchmark executor."""
-    project = Path(args.path).resolve()
-    if not project.exists():
-        print(f"Error: project path does not exist: {project}", file=sys.stderr)
-        return 1
 
+def _run_legacy(args: argparse.Namespace, project: Path) -> int:
+    """Legacy path: use OptimizationLoop directly."""
     from factory.optimization import AgenticMutator, LoopConfig, OptimizationLoop, Surface
     from factory.optimization.benchmarks.harbor import HarborBenchmark
     from factory.optimization.protocols import Evaluator, Executor
@@ -170,3 +172,100 @@ def cmd_optimize(args: argparse.Namespace) -> int:
         print(f"Total improvement: {total_delta:+.4f}")
 
     return 0
+
+
+def cmd_optimize(args: argparse.Namespace) -> int:
+    """Run the optimization loop with HarborBenchmark executor."""
+    project = Path(args.path).resolve()
+    if not project.exists():
+        print(f"Error: project path does not exist: {project}", file=sys.stderr)
+        return 1
+
+    legacy = getattr(args, "legacy", False)
+    if legacy:
+        return _run_legacy(args, project)
+
+    benchmark = getattr(args, "benchmark", None) or "searchqa"
+    git_ref = getattr(args, "git_ref", None) or os.environ.get("FACTORY_GIT_REF", "main")
+    docker_host = getattr(args, "docker_host", None) or os.environ.get("DOCKER_HOST", "")
+    concurrency = getattr(args, "concurrency", 5)
+    steps = getattr(args, "steps", 3)
+    model = getattr(args, "model", None) or "sonnet"
+    skill_path = getattr(args, "skill_path", None)
+
+    # Setup: write initial state files
+    opt_dir = project / ".factory" / "optimization"
+    opt_dir.mkdir(parents=True, exist_ok=True)
+
+    # Initial skill
+    skill_file = opt_dir / "current_skill.md"
+    if skill_path:
+        sp = Path(skill_path)
+        if sp.exists():
+            skill_file.write_text(sp.read_text())
+    elif not skill_file.exists():
+        default_skill = (
+            "# Question Answering Skill\n\n"
+            "(No learned rules yet.)\n\n"
+            "## Instructions\n\n"
+            "Read the question and search results from /tmp/task-instruction.md.\n"
+            "Answer the question and write ONLY your final answer to /workspace/answer.txt.\n"
+            "Also include your answer in <answer> tags in your response.\n"
+        )
+        skill_file.write_text(default_skill)
+
+    # Initial state.json
+    state_file = opt_dir / "state.json"
+    initial_state = {
+        "step": 0,
+        "current_score": 0.0,
+        "best_score": 0.0,
+        "best_step": 0,
+        "history": [],
+    }
+    state_file.write_text(json.dumps(initial_state, indent=2) + "\n")
+
+    # Pass config to optimize-step via env vars
+    env = os.environ.copy()
+    env["FACTORY_OPT_BENCHMARK"] = benchmark
+    env["FACTORY_OPT_CONCURRENCY"] = str(concurrency)
+    env["FACTORY_GIT_REF"] = git_ref
+    env["FACTORY_OPT_MODEL"] = model
+    env["FACTORY_OPT_MAX_ITERATIONS"] = str(steps)
+    if docker_host:
+        env["DOCKER_HOST"] = docker_host
+
+    print(f"Starting optimization (workflow): benchmark={benchmark}, max_iterations={steps}, concurrency={concurrency}")
+
+    # Execute workflow
+    cmd = ["factory", "workflow", "run", "optimize", str(project)]
+    result = subprocess.run(cmd, env=env)
+
+    # Read results
+    test_result_path = opt_dir / "test_result.json"
+    state: dict[str, Any] = json.loads(state_file.read_text()) if state_file.exists() else dict(initial_state)
+
+    print(f"\n{'='*50}")
+    history: list[dict[str, Any]] = state.get("history", [])
+    print(f"Training complete: {len(history)} steps")
+
+    if history:
+        baseline_score: float = history[0].get("score_start", 0.0)
+        print(f"Baseline score: {baseline_score:.4f}")
+        for h in history:
+            delta: float = h.get("score_delta", 0.0)
+            print(f"  Step {h['step']}: {h['score_start']:.4f} -> {h['score_end']:.4f} "
+                  f"({delta:+.4f}) verdict={h.get('verdict', 'n/a')}")
+
+    print(f"Best score: {state['best_score']:.4f} (step {state['best_step']})")
+    print(f"Final score: {state['current_score']:.4f}")
+
+    if test_result_path.exists():
+        test_data = json.loads(test_result_path.read_text())
+        print(f"Test score:  {test_data['score']:.4f}")
+
+    if history:
+        total_delta_f: float = state["current_score"] - history[0].get("score_end", 0.0)
+        print(f"Total improvement: {total_delta_f:+.4f}")
+
+    return result.returncode