feat: end-to-end task→workflow→outer-loop creation pipeline (#1478)

docs/outer-loop.md

@@ -337,34 +337,58 @@ factory outer-loop calibrate <project> \
 
 ## Task Discovery
 
-Projects with custom `Task` subclasses (e.g. `chess-evolve`, `harbor`) can pass their task class to the outer loop via the `--task-module` flag. This uses the same `module:ClassName` import-string pattern as `EvaluatorRef`.
+Tasks can be passed to the outer loop via `--task` (recommended) or `--task-module` (escape hatch).
 
-### Format
+### `--task` (recommended)
 
+The `--task` flag accepts three formats, resolved in order by `resolve_task()`:
+
+1. **TOML file** — path ending in `.toml` or resolving to an existing `.toml` file. Parsed via `TaskDefinition.from_toml()`.
+2. **Python file** — path ending in `.py` or resolving to an existing `.py` file. The module is loaded and introspected for a single `Task` subclass.
+3. **Module:Class string** — `module.path:ClassName` format, resolved via `TaskRef.resolve()`. Same as `--task-module`.
+
+```bash
+# TOML task
+factory outer-loop calibrate /path/to/factory \
+  --task .factory/tasks/chess-evolve.toml \
+  --seed-workflow chess_pkg:build_pipeline
+
+# Python task file
+factory outer-loop calibrate /path/to/factory \
+  --task examples/chess_evolve_task.py \
+  --seed-workflow chess_pkg:build_pipeline
+
+# Module:Class (same as --task-module)
+factory outer-loop calibrate /path/to/factory \
+  --task chess_evolve.task:ChessEvolveTask \
+  --seed-workflow chess_pkg:build_pipeline
+
+# Override on evaluate
+factory outer-loop evaluate /path/to/factory \
+  --generation 0 \
+  --task .factory/tasks/chess-evolve.toml
 ```
-module.path:ClassName
-```
 
-The module is imported via `importlib.import_module()` and the class is resolved via `getattr()`. The class must be a subclass of `factory.task.Task`.
+Relative paths are resolved against the project directory.
+
+### `--task-module` (escape hatch)
 
-### Precondition
+The `--task-module` flag accepts only `module.path:ClassName` import strings. It is retained for backward compatibility. The module is imported via `importlib.import_module()` and the class is resolved via `getattr()`. The class must be a subclass of `factory.task.Task`.
 
 The task's package must be importable — install it first:
 
 ```bash
 pip install -e /path/to/my-project
 ```
 
-### Usage
-
 ```bash
-# Calibrate with a custom task
+# Calibrate with --task-module
 factory outer-loop calibrate /path/to/factory \
   --benchmark chess-evolve \
   --task-module chess_evolve.task:ChessEvolveTask \
   --project-dir /path/to/chess-evolve
 
-# Evaluate with a custom task (overrides persisted config)
+# Evaluate with --task-module (overrides persisted config)
 factory outer-loop evaluate /path/to/factory \
   --generation 0 \
   --task-module chess_evolve.task:ChessEvolveTask
@@ -374,8 +398,8 @@ factory outer-loop evaluate /path/to/factory \
 
 `SwarmConfig.get_task()` resolves tasks with 3-tier precedence:
 
-1. **`set_task()`** — explicit runtime attachment (highest, used by tests and in-process callers)
-2. **`task_module`** — `module:ClassName` string from CLI flag or persisted config
+1. **`set_task()`** — explicit runtime attachment (highest, used by `--task` CLI flag, tests, and in-process callers)
+2. **`task_module`** — `module:ClassName` string from `--task-module` CLI flag or persisted config
 3. **`Task.from_legacy()`** — constructed from flat fields (`test_command`, `test_format`, etc.)
 
-The `task_module` field is serialized with `SwarmConfig`, so it persists across `save_config`/`load_config` — no need to re-specify it on every `evaluate` invocation after `calibrate`.
+The `task_module` field is serialized with `SwarmConfig`, so it persists across `save_config`/`load_config` — no need to re-specify it on every `evaluate` invocation after `calibrate`. Tasks resolved via `--task` are attached at runtime via `set_task()` and take highest precedence.

factory/agents/prompts/ceo.md

@@ -323,6 +323,7 @@ Each mode's full instructions live in a workflow skill under `skills/workflow-<n
 **Mode overrides (from task directives):**
 - `--mode design` or `## Plan Loop (Interactive)` → read `skills/workflow-design/SKILL.md`
 - `--mode create` or `## Create Mode` → read `skills/workflow-create/SKILL.md`
+- `--mode task-setup` → read `skills/workflow-task-setup/SKILL.md`
 
 **Invocation:** Read the selected SKILL.md file, then follow its instructions as your mode-specific playbook. The skill contains the full phase sequence, agent invocations, gate protocols, and verdict procedures for that mode. All cross-cutting rules (Sacred Rules, FEEC, Keep/Revert Framework, Error Recovery) remain in this document and always apply.
 

factory/cli/_ceo_helpers.py

@@ -535,6 +535,7 @@ def _execute_ceo(
     update_existing_mode: str | None,
     plugin_mode: bool = False,
     plugin_folder: str | None = None,
+    task_ref: str | None = None,
     deferred_spec: str | None,
     needs_materialize: bool,
     refine_request: str | None,
@@ -715,6 +716,7 @@ def _execute_ceo(
         update_existing_mode=update_existing_mode,
         plugin_mode=plugin_mode,
         plugin_folder=plugin_folder,
+        task_ref=task_ref,
         from_plan=resolved_plan.plan if resolved_plan else None,
         from_plan_feedback=resolved_plan.feedback if resolved_plan else None,
         just_plan=just_plan,

factory/cli/_helpers.py

@@ -43,6 +43,7 @@ def _resolve_inactivity_timeout() -> float:
     "review",
     "deep-qa",
     "create",
+    "task-setup",
     "study",
     "swebench",
     "frontend-design",

factory/cli/_parser_groups.py

@@ -454,6 +454,10 @@ def add_entry_point_parsers(sub: argparse._SubParsersAction) -> None:  # type: i
     p.add_argument("--folder", default=None, metavar="PATH",
                     help="Output directory for plugin package (default: ./<mode-name>-plugin). "
                          "Only used with --plugin.")
+    p.add_argument("--task", default=None, metavar="PATH",
+                    help="Task reference for task-aware create mode: .toml file, .py file, "
+                         "or module:ClassName. Injects the resolved TaskDefinition's scoring "
+                         "contract into create mode's workflow generation.")
     p.add_argument("--engine", choices=["skill", "tool", "deterministic"], default="skill",
                     help="Execution engine: skill (CEO follows SKILL.md, default), "
                          "tool (CEO drives via factory workflow tool commands), "

factory/cli/_task_builder.py

@@ -69,6 +69,14 @@ def _mode_suffix(mode: str, discover_only: bool) -> str:
             "with structural graph context included. "
             "Terminal mode — does not chain to other modes."
         ),
+        "task-setup": (
+            "\n\nRun Task Setup mode: scaffold a Task file for the target project. "
+            "Study the repository to understand its domain and verification methods. "
+            "Classify whether the task needs TOML (executable verification) or Python "
+            "(judgmental verification). Produce a validated .factory/tasks/<name>.toml "
+            "or .py file. Terminal mode — does not chain to other modes. "
+            "The full step-by-step playbook is in your system prompt above."
+        ),
     }
     if mode == "discover":
         if discover_only:
@@ -162,6 +170,89 @@ def _append_deep_research_topic(task: str, focus: str) -> str:
     )
 
 
+def _build_task_aware_directive(task_ref: str, project_path: Path) -> str:
+    """Build the task-aware create mode directive section.
+
+    Same directive-injection pattern as Plugin Package and Update Existing Mode
+    extensions — injects into the CEO task text, same file, same audit surface.
+    """
+    from factory.task import resolve_task
+
+    try:
+        resolved = resolve_task(task_ref, project_path)
+        defn = resolved.definition
+        scoring_info = f"method={defn.scoring.method}, metric_path={defn.scoring.metric_path}"
+        constraints_info = (
+            f"timeout={defn.constraints.timeout}s, "
+            f"max_retries={defn.constraints.max_retries}"
+        )
+        if defn.constraints.required_capabilities:
+            caps = ", ".join(str(c) for c in defn.constraints.required_capabilities)
+            constraints_info += f", required_capabilities=[{caps}]"
+        verify_info = defn.verify_config.command or "(no verify command)"
+    except Exception as exc:
+        return (
+            f"\n\n## Create Mode (Task-Aware) — RESOLUTION FAILED\n\n"
+            f"Could not resolve --task {task_ref!r}: {exc}\n"
+            f"Proceed without task awareness.\n"
+        )
+
+    return (
+        f"\n\n## Create Mode (Task-Aware)\n\n"
+        f"A TaskDefinition has been provided via `--task {task_ref}`.\n\n"
+        f"**Resolved Task:** {defn.name}\n"
+        f"**Description:** {defn.description}\n"
+        f"**Scoring Contract:** {scoring_info}\n"
+        f"**Verify Command:** {verify_info}\n"
+        f"**Constraints:** {constraints_info}\n\n"
+        f"The generated workflow MUST be compatible with this task's scoring contract.\n"
+        f"The Strategist and Builder should tailor the workflow to this task's needs:\n"
+        f"- Agent prompts should reference the task's verification method\n"
+        f"- Gate conditions should align with the scoring method ({defn.scoring.method})\n"
+        f"- Timeout values should respect the task's constraint ({defn.constraints.timeout}s)\n\n"
+        f"### Workflow-Level OptKnob Auto-Generation\n\n"
+        f"The generated workflow MUST include OptKnobs using this mechanical derivation:\n\n"
+        f"| Node Field | OptKnob Kind | Bounds | Expandable |\n"
+        f"|---|---|---|---|\n"
+        f"| AgentNode.role (each agent) | model | ['haiku', 'sonnet', 'opus'] | False |\n"
+        f"| AgentNode.timeout | threshold | [default/2, default, default*2] | False |\n"
+        f"| AgentNode.prompt_template (when non-empty) | prompt | [current_value] | True |\n\n"
+        f"### DataNode OptKnob Auto-Generation\n\n"
+        f"When the generated workflow contains DataNode instances, "
+        f"derive additional threshold OptKnobs from their fields:\n\n"
+        f"| Node Field | OptKnob Kind | Bounds | Expandable | Description |\n"
+        f"|---|---|---|---|---|\n"
+        f"| DataNode.parallelism | threshold | "
+        f"[1, current_value, current_value*2, 8] | False | "
+        f"Number of concurrent data items processed in parallel |\n"
+        f"| DataNode.limit (when set, not None) | threshold | "
+        f"[limit//2, limit, limit*2] | True | "
+        f"Maximum number of data items to process per run |\n"
+        f"| DataNode.max_items | threshold | "
+        f"[100, 250, 500, 1000] | True | "
+        f"Safety ceiling for total data items loaded |\n\n"
+        f"Do NOT add topology knobs for DataNode.split or DataNode.shuffle.\n\n"
+        f"Never auto-generate kind='topology' knobs.\n"
+        f"Use `compose.py validate_composition()` as a post-build gate.\n\n"
+        f"### Domain-Level OptKnob Suggestions\n\n"
+        f"When a TaskDefinition is provided, the Strategist should also propose "
+        f"domain-level knobs as LLM-authored suggestions (not deterministic derivations). "
+        f"These are CEO-gated — the Strategist proposes them, the CEO reviews.\n\n"
+        f"**Rules:**\n"
+        f"- Restrict to prompt/model/threshold kinds only (never topology)\n"
+        f"- Default expandable=False for auto-generated domain knobs\n"
+        f"- When scoring.method is 'json' with a numeric metric_path, propose a "
+        f"threshold-kind knob using ScoringContract.threshold\n"
+        + (
+            f"- Current threshold: {defn.scoring.threshold}\n"
+            if defn.scoring.threshold is not None
+            else "- No threshold configured — Strategist may propose one\n"
+        )
+        + "- Domain knobs should be grounded in the task's actual constraints, "
+        "not hallucinated from descriptions\n"
+    )
+
+
 def _build_ceo_task(
     project_path: Path,
     mode: str,
@@ -188,6 +279,7 @@ def _build_ceo_task(
     update_existing_mode: str | None = None,
     plugin_mode: bool = False,
     plugin_folder: str | None = None,
+    task_ref: str | None = None,
     from_plan: str | None = None,
     from_plan_feedback: list[str] | None = None,
     just_plan: bool = False,
@@ -448,6 +540,9 @@ def _build_ceo_task(
             f"factory/workflow/skill_export.py, factory/cli.py, tests/.\n"
         )
 
+    if task_ref and create_description:
+        task += _build_task_aware_directive(task_ref, project_path)
+
     if prompt_file:
         task += (
             f"\n\n## Directive\n\n"

factory/cli/ceo.py

@@ -75,6 +75,7 @@ def cmd_ceo(args: argparse.Namespace) -> int:
 
     plugin_mode = getattr(args, "plugin", False)
     plugin_folder = getattr(args, "folder", None)
+    task_ref = getattr(args, "task", None)
 
     if plugin_mode and mode != "create":
         print(
@@ -161,6 +162,7 @@ def cmd_ceo(args: argparse.Namespace) -> int:
         update_existing_mode=update_existing_mode,
         plugin_mode=plugin_mode,
         plugin_folder=plugin_folder,
+        task_ref=task_ref,
         deferred_spec=deferred_spec,
         needs_materialize=needs_materialize,
         refine_request=refine_request,

factory/cli/outer_loop.py

@@ -196,6 +196,7 @@ def _cmd_calibrate(args: argparse.Namespace) -> int:
         resolved_prep_command = bench_config.prep_command if bench_config else ""
 
         task_module = getattr(args, "task_module", "")
+        cli_task_ref = getattr(args, "task", "")
         seed_workflow_module = getattr(args, "seed_workflow", "")
         config = SwarmConfig(
             benchmark=benchmark,
@@ -216,6 +217,12 @@ def _cmd_calibrate(args: argparse.Namespace) -> int:
         if task_module:
             _log.info("task_module_resolved", ref=task_module)
 
+        if cli_task_ref:
+            from factory.task import resolve_task
+            resolved = resolve_task(cli_task_ref, project_path)
+            config.set_task(resolved)
+            _log.info("task_resolved_from_cli", ref=cli_task_ref, name=resolved.name)
+
     root = init_filesystem(project_path, config)
 
     benchmark = config.benchmark
@@ -461,6 +468,13 @@ def _cmd_evaluate(args: argparse.Namespace) -> int:
         config = config.model_copy(update={"task_module": cli_task_module})
         _log.info("task_module_override", ref=cli_task_module)
 
+    cli_task_ref = getattr(args, "task", "")
+    if cli_task_ref:
+        from factory.task import resolve_task
+        resolved = resolve_task(cli_task_ref, project_path)
+        config.set_task(resolved)
+        _log.info("task_resolved_from_cli", ref=cli_task_ref, name=resolved.name)
+
     eval_project_dir = getattr(args, "project_dir", None)
     if eval_project_dir is not None:
         eval_project_dir = str(Path(eval_project_dir).resolve())
@@ -928,6 +942,11 @@ def add_outer_loop_parser(subparsers: argparse._SubParsersAction) -> None:  # ty
         default="",
         help="Task class ref as 'module.path:ClassName' (e.g. chess_evolve.task:ChessEvolveTask)",
     )
+    cal.add_argument(
+        "--task",
+        default="",
+        help="Task reference: .toml file, .py file, or module:ClassName (e.g. .factory/tasks/chess.toml)",
+    )
     cal.add_argument(
         "--seed-workflow",
         required=True,
@@ -947,6 +966,11 @@ def add_outer_loop_parser(subparsers: argparse._SubParsersAction) -> None:  # ty
         default="",
         help="Task class ref as 'module.path:ClassName' (overrides config value)",
     )
+    ev.add_argument(
+        "--task",
+        default="",
+        help="Task reference: .toml file, .py file, or module:ClassName (overrides config value)",
+    )
 
     ref = outer_sub.add_parser("reflect", help="Run reflection on generation")
     ref.add_argument("project_path", nargs="?", default=".")