feat: open agent roles — accept AgentRole | str for custom roles

CLAUDE.md

@@ -75,7 +75,7 @@ Spawned via `factory ceo /path` or `factory run /path`. The CEO receives `ceo.md
 
 ### Layer 4: Specialist Agents (`factory/agents/`)
 
-Eight specialist Claude Code subprocesses spawned by the CEO via `factory agent <role>`. Agent prompts are resolved via `factory/agents/runner.py` with a two-tier lookup: project-specific override (`.factory/agents/<role>.md`) then factory default (`factory/agents/prompts/<role>.md`). Evolved playbooks from `~/.factory/playbooks/<role>.md` (user-local, ACE-generated) are auto-injected, falling back to factory defaults in `factory/agents/playbooks/<role>.md`.
+Eight specialist Claude Code subprocesses spawned by the CEO via `factory agent <role>`. Roles can be `AgentRole` enum values or custom strings. Agent prompts are resolved via `factory/agents/runner.py` with a three-tier lookup: project-specific override (`.factory/agents/<role>.md`) → user-level (`~/.factory/agents/<role>.md`) → factory default (`factory/agents/prompts/<role>.md`). Evolved playbooks from `~/.factory/playbooks/<role>.md` (user-local, ACE-generated) are auto-injected, falling back to factory defaults in `factory/agents/playbooks/<role>.md`.
 
 **Roles:** Researcher (observe), Strategist (hypothesize and refine ideas), Builder (implement), QA (health check + code review + adversarial QA), Archivist (record), Refiner (scope refinements), Failure Analyst (research mode), CEO (orchestrate).
 

factory/agents/plugin.py

@@ -6,12 +6,15 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+import structlog
 import yaml
 
 from factory.ace.injector import inject_playbook
 from factory.ace.paths import DEFAULTS_DIR as _PLAYBOOKS_DIR
 from factory.agents.runner import _PROMPTS_DIR
 
+_log = structlog.get_logger()
+
 _AGENTS_YML = Path(__file__).parent / "agents.yml"
 _PLUGIN_AGENTS_DIR_CANDIDATE = Path(__file__).resolve().parent.parent.parent / "agents"
 _PLUGIN_AGENTS_DIR: Path | None = _PLUGIN_AGENTS_DIR_CANDIDATE if _PLUGIN_AGENTS_DIR_CANDIDATE.is_dir() else None
@@ -105,9 +108,8 @@ def _sandbox_mode(role: str) -> str:
         return "read-only"
     if role in _WORKSPACE_WRITE_ROLES:
         return "workspace-write"
-    raise ValueError(
-        f"Unknown role {role!r}: not in _READ_ONLY_ROLES or _WORKSPACE_WRITE_ROLES"
-    )
+    _log.warning("custom_role_sandbox_default", role=role, sandbox_mode="workspace-write")
+    return "workspace-write"
 
 
 def _escape_toml_multiline_literal(text: str) -> str:

factory/agents/runner.py

@@ -28,6 +28,9 @@
     "refactory",
 ]
 
+# User-level agent prompts directory
+_USER_AGENTS_DIR = Path.home() / ".factory" / "agents"
+
 # Consecutive failure tracking
 _consecutive_failures: int = 0
 _FAILURE_ABORT_THRESHOLD: int = 2
@@ -72,7 +75,7 @@ def reset_failure_counter() -> None:
 
 
 def resolve_prompt(
-    role: AgentRole,
+    role: str,
     project_path: Path | None = None,
     *,
     use_profile: bool = False,
@@ -82,7 +85,8 @@ def resolve_prompt(
 
     Resolution order:
     1. Project-specific override: <project>/.factory/agents/<role>.md
-    2. Factory default: factory/agents/prompts/<role>.md
+    2. User-level: ~/.factory/agents/<role>.md
+    3. Factory default: factory/agents/prompts/<role>.md
 
     When *use_profile* is True, loads ~/.factory/profile.md and appends it
     after the ACE playbook injection.
@@ -110,14 +114,30 @@ def resolve_prompt(
                 prompt = _maybe_inject_skill(prompt, project_path, workflow_mode)
             return prompt
 
+    # Check for user-level override
+    user_path = _USER_AGENTS_DIR / f"{role}.md"
+    if user_path.exists():
+        logger.info("Using user-level prompt for %s: %s", role, user_path)
+        prompt = user_path.read_text()
+        playbook = load_playbook(role)
+        if playbook:
+            prompt = inject_playbook(prompt, playbook)
+            logger.info("Injected playbook for %s (user override)", role)
+        if use_profile:
+            prompt = _maybe_inject_profile(prompt, role)
+        if role == "ceo" and workflow_mode and project_path is not None:
+            prompt = _maybe_inject_skill(prompt, project_path, workflow_mode)
+        return prompt
+
     # Fall back to factory default
     default_path = _PROMPTS_DIR / f"{role}.md"
     if not default_path.exists():
-        override_hint = (
-            f" or {project_path / '.factory' / 'agents' / f'{role}.md'}" if project_path else ""
-        )
+        searched = [str(default_path)]
+        searched.append(str(user_path))
+        if project_path is not None:
+            searched.append(str(project_path / ".factory" / "agents" / f"{role}.md"))
         raise FileNotFoundError(
-            f"No prompt found for agent role '{role}'. Expected at {default_path}{override_hint}"
+            f"No prompt found for agent role '{role}'. Searched: {', '.join(searched)}"
         )
 
     prompt = default_path.read_text()
@@ -162,7 +182,7 @@ def _maybe_inject_skill(prompt: str, project_path: Path, workflow_mode: str) ->
 
 
 async def invoke_agent(
-    role: AgentRole,
+    role: str,
     task: str,
     project_path: Path,
     *,
@@ -537,7 +557,7 @@ def complete_cycle_session(
 
 
 async def invoke_agents_parallel(
-    tasks: list[tuple[AgentRole, str]],
+    tasks: list[tuple[str, str]],
     project_path: Path,
     *,
     timeout: float = 600.0,

factory/cli/_ceo_helpers.py

@@ -74,7 +74,9 @@ def _tool_exec_protocol(wt_path: Path) -> str:
             f"{overview}\n"
         )
 
-    protocol += (
+    no_lookahead = os.environ.get("FACTORY_TOOL_NO_LOOKAHEAD") == "1"
+
+    commands = (
         "\n## Commands\n"
         "\n"
         f"  factory workflow tool next {p}\n"
@@ -83,6 +85,15 @@ def _tool_exec_protocol(wt_path: Path) -> str:
         "  TOOL_OUTPUT\n"
         f"  factory workflow tool status {p}\n"
         f"  factory workflow tool curr {p}\n"
+    )
+    if not no_lookahead:
+        commands += (
+            f"  factory workflow tool peek {p} --node <NODE_ID>\n"
+            f"  factory workflow tool lookahead {p} --count 5\n"
+        )
+
+    protocol += commands
+    protocol += (
         "\n"
         "## Protocol\n"
         "\n"
@@ -99,6 +110,28 @@ def _tool_exec_protocol(wt_path: Path) -> str:
         '   call "submit" with your verdict (PROCEED, RETRY, or HALT)\n'
         "6. If RETRY: the tool rewinds — run \"next\" to get the retry task\n"
         "7. If DONE: report completion\n"
+    )
+
+    if not no_lookahead:
+        protocol += (
+            "\n"
+            "## Loop Context — Automatic Downstream Visibility\n"
+            "\n"
+            "When you run \"next\" and land on a node that is part of a loop (e.g., "
+            "builder → QA gates → RELOOP back to builder), the tool automatically "
+            "appends LOOP CONTEXT showing every step in that loop: gate criteria, "
+            "agent roles, artifact contracts, and what triggers a RELOOP.\n"
+            "\n"
+            "Use this context to craft better agent tasks — include the success "
+            "criteria from downstream gates, specify which artifacts to produce, "
+            "and warn about checks the agent's output must pass. This is injected "
+            "automatically; no extra command needed.\n"
+            "\n"
+            'You can also use "peek --node <NODE_ID>" to inspect any specific node, '
+            'or "lookahead --count N" to see the next N nodes.\n'
+        )
+
+    protocol += (
         "\n"
         "## Important\n"
         "\n"

factory/cycle_analyzer.py

@@ -11,7 +11,7 @@
 import json
 from dataclasses import dataclass, field, asdict
 from pathlib import Path
-from factory.workflow.primitives import AgentNode, Workflow
+from factory.workflow.primitives import AgentNode, Workflow, _role_str
 
 
 @dataclass
@@ -457,7 +457,7 @@ def _match_node(self, role: str) -> str | None:
         if not self.workflow:
             return None
         for nid, node in self.workflow.nodes.items():
-            if isinstance(node, AgentNode) and node.role.value == role:
+            if isinstance(node, AgentNode) and _role_str(node.role) == role:
                 return nid
         return None
 

factory/workflow/cli.py

@@ -19,6 +19,7 @@
     GateNode,
     JoinNode,
     Study,
+    _role_str,
 )
 
 log = structlog.get_logger()
@@ -132,7 +133,7 @@ def _cmd_show(args: argparse.Namespace) -> int:
         writes = ", ".join(sorted(node.writes)) if node.writes else "-"
 
         if isinstance(node, AgentNode):
-            ntype = f"Agent({node.role.value})"
+            ntype = f"Agent({_role_str(node.role)})"
         elif isinstance(node, GateNode):
             ntype = f"Gate({node.evaluator_type})"
         elif isinstance(node, ForkNode):

factory/workflow/context.py

@@ -17,6 +17,7 @@
     FnNode,
     GateNode,
     Workflow,
+    _role_str,
 )
 
 PROMPTS_DIR = Path(__file__).parent.parent / "agents" / "prompts"
@@ -45,9 +46,9 @@ def _extract_agent_prompts(workflow: Workflow) -> dict[str, str]:
 
     for node in workflow.nodes.values():
         if isinstance(node, AgentNode):
-            roles.add(node.role.value)
+            roles.add(_role_str(node.role))
         elif isinstance(node, GateNode) and node.evaluator_role:
-            roles.add(node.evaluator_role.value)
+            roles.add(_role_str(node.evaluator_role))
 
     prompts: dict[str, str] = {}
     for role in sorted(roles):
@@ -87,14 +88,14 @@ def _extract_node_summary(workflow: Workflow) -> dict[str, dict[str, Any]]:
     for node_id, node in workflow.nodes.items():
         info: dict[str, Any] = {"type": type(node).__name__}
         if isinstance(node, AgentNode):
-            info["role"] = node.role.value
+            info["role"] = _role_str(node.role)
             info["blocking"] = node.blocking
             if node.timeout:
                 info["timeout"] = node.timeout
         elif isinstance(node, GateNode):
             info["evaluator_type"] = node.evaluator_type
             if node.evaluator_role:
-                info["evaluator_role"] = node.evaluator_role.value
+                info["evaluator_role"] = _role_str(node.evaluator_role)
         elif isinstance(node, FnNode):
             info["command"] = node.command[:80]
         if node.reads:

factory/workflow/definitions.py

@@ -1828,6 +1828,9 @@ def create_workflow() -> Workflow:
             "6) Run factory workflow export-skills to generate the SKILL.md "
             "7) Write tests in tests/ "
             "8) Run pytest and ruff check to verify "
+            "9) If the workflow defines custom agent roles not in the built-in AgentRole enum, "
+            "create corresponding prompt files at .factory/agents/<role>.md with clear "
+            "instructions for the role's purpose, output format, and factory conventions "
             "Commit changes and open a draft PR."
         ),
         reads={".factory/strategy/current.md"},

factory/workflow/executor.py

@@ -37,6 +37,7 @@
     Verdict,
     VerdictType,
     Workflow,
+    _role_str,
 )
 
 log = structlog.get_logger()
@@ -811,33 +812,41 @@ async def _run_agent(self, node: AgentNode) -> str:
         """Invoke an agent via factory/agents/runner.py."""
         from factory.agents.runner import invoke_agent
 
+        role = _role_str(node.role)
+
         task = node.prompt_template
         context = self.node_context.get(node.id, "")
         if context:
             task = f"{task}\n\n{context}"
 
         model = node.model
         if not model:
-            pool_entry = self.agent_pool.get(node.role.value)
+            pool_entry = self.agent_pool.get(role)
             if pool_entry:
                 model = pool_entry.model
+            else:
+                model = "sonnet"
+                log.info("custom_role_pool_default", role=role, model="sonnet", timeout=600)
 
         timeout = node.timeout
         if timeout is None:
-            pool_entry = self.agent_pool.get(node.role.value)
+            pool_entry = self.agent_pool.get(role)
             if pool_entry:
                 timeout = pool_entry.timeout
+            else:
+                timeout = 600
+                log.info("custom_role_pool_default", role=role, model=model, timeout=600)
 
         stdout, code = await invoke_agent(
-            node.role.value,  # type: ignore[arg-type]
+            role,
             task,
             self.project_path,
             model=model or None,
             timeout=float(timeout) if timeout is not None else 600.0,
         )
 
         if code != 0:
-            raise RuntimeError(f"agent {node.role.value} exited with code {code}")
+            raise RuntimeError(f"agent {role} exited with code {code}")
 
         return stdout
 

factory/workflow/primitives.py

@@ -5,10 +5,13 @@
 from enum import Enum
 from typing import Any, Callable, Literal
 
+import structlog
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from factory.models import FactoryConfig, ProjectState
 
+log = structlog.get_logger()
+
 
 # ── agent pool ───────────────────────────────────────────────────
 
@@ -27,12 +30,17 @@ class AgentRole(str, Enum):
     SKILL_REVIEWER = "skill_reviewer"
 
 
+def _role_str(role: AgentRole | str) -> str:
+    """Extract the string value from an AgentRole enum or pass through a plain string."""
+    return role.value if isinstance(role, AgentRole) else role
+
+
 class AgentConfig(BaseModel):
     """Configuration for an agent in the pool."""
 
     model_config = ConfigDict(strict=True, extra="forbid")
 
-    role: AgentRole
+    role: AgentRole | str
     model: str
     timeout: int = 600
 
@@ -130,7 +138,7 @@ class AgentNode(Node):
 
     model_config = ConfigDict(strict=True, extra="forbid")
 
-    role: AgentRole
+    role: AgentRole | str
     model: str = ""
     prompt_template: str = ""
     tools: list[str] = Field(default_factory=list)
@@ -155,7 +163,7 @@ class GateNode(Node):
     model_config = ConfigDict(strict=True, extra="forbid")
 
     evaluator_type: Literal["agent", "fn", "user"] = "agent"
-    evaluator_role: AgentRole | None = None
+    evaluator_role: AgentRole | str | None = None
     evaluator_command: str | None = None
     gate_prompt: str = ""
 

factory/workflow/skill_export.py

@@ -30,6 +30,7 @@
     SubgraphForkNode,
     VerdictType,
     Workflow,
+    _role_str,
 )
 from factory.workflow.templates import emit
 
@@ -328,7 +329,7 @@ def _agent_to_instruction(
     is_parallel: bool = False,
 ) -> str:
     """Convert an AgentNode to a CLI invocation instruction with template slots."""
-    role = node.role.value
+    role = _role_str(node.role)
     pool_entry = DEFAULT_AGENT_POOL.get(role)
     default_timeout = node.timeout or (pool_entry.timeout if pool_entry else 600)
     model_flag = " --model haiku" if role == "archivist" else ""
@@ -531,7 +532,7 @@ def _gate_to_checkpoint(
     else:
         gate_prompt_slot = emit(f"gate_prompt_{node.id}", node.gate_prompt)
         ann = [
-            f"<!-- gate: GateNode id={node.id} evaluator_type=agent evaluator_role={node.evaluator_role.value if node.evaluator_role else 'CEO'} -->",
+            f"<!-- gate: GateNode id={node.id} evaluator_type=agent evaluator_role={_role_str(node.evaluator_role) if node.evaluator_role else 'CEO'} -->",
             f"<!-- reads: {reads_ann} -->",
             f"<!-- edges: {edges_str} -->",
         ]
@@ -790,7 +791,7 @@ def workflow_to_skill_md(workflow: Workflow) -> str:
             phase_num += 1
 
         elif isinstance(node, AgentNode):
-            role_title = node.role.value.replace("_", " ").title()
+            role_title = _role_str(node.role).replace("_", " ").title()
             node_title = nid.replace("_", " ").title()
             if role_title.lower() in node_title.lower():
                 section_title = node_title