From c6b2066674f28ac50554c1fcfeab1b29635e6840 Mon Sep 17 00:00:00 2001 From: colehurwitz Date: Wed, 12 Aug 2026 13:31:52 -0400 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20open=20agent=20roles=20=E2=80=94=20?= =?UTF-8?q?accept=20AgentRole=20|=20str=20for=20custom=20roles?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Widen AgentNode.role, AgentConfig.role, and GateNode.evaluator_role to accept plain strings alongside the AgentRole enum, enabling custom agent roles without modifying factory source code. Key changes: - Add _role_str() helper in primitives.py; use at all .value call sites - Three-tier prompt resolution: project → user (~/.factory/agents/) → builtin - Default sandbox mode "workspace-write" for unknown roles (with warning) - Pool fallback (model=sonnet, timeout=600) for unknown roles in executor - JSON roundtrip safety in tool.py via try/except on AgentRole() - Create mode checklist updated to include custom role prompt creation - Comprehensive test suite (tests/test_open_agent_roles.py) Co-Authored-By: Claude Opus 4.6 --- factory/agents/plugin.py | 8 +- factory/agents/runner.py | 36 ++++- factory/cycle_analyzer.py | 4 +- factory/workflow/cli.py | 3 +- factory/workflow/context.py | 9 +- factory/workflow/definitions.py | 3 + factory/workflow/executor.py | 17 ++- factory/workflow/primitives.py | 14 +- factory/workflow/skill_export.py | 7 +- factory/workflow/tool.py | 24 ++- factory/workflow/verification.py | 4 +- tests/test_open_agent_roles.py | 243 +++++++++++++++++++++++++++++++ tests/test_plugin_agents.py | 5 +- 13 files changed, 336 insertions(+), 41 deletions(-) create mode 100644 tests/test_open_agent_roles.py diff --git a/factory/agents/plugin.py b/factory/agents/plugin.py index 09c145afe..4b7faf169 100644 --- a/factory/agents/plugin.py +++ b/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: diff --git a/factory/agents/runner.py b/factory/agents/runner.py index 45e54cef7..ba1482078 100644 --- a/factory/agents/runner.py +++ b/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: /.factory/agents/.md - 2. Factory default: factory/agents/prompts/.md + 2. User-level: ~/.factory/agents/.md + 3. Factory default: factory/agents/prompts/.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, diff --git a/factory/cycle_analyzer.py b/factory/cycle_analyzer.py index 2901161de..de096dc8e 100644 --- a/factory/cycle_analyzer.py +++ b/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 diff --git a/factory/workflow/cli.py b/factory/workflow/cli.py index 7d42a9f50..438153066 100644 --- a/factory/workflow/cli.py +++ b/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): diff --git a/factory/workflow/context.py b/factory/workflow/context.py index 7d4f48b7f..c7b5ae7b1 100644 --- a/factory/workflow/context.py +++ b/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: diff --git a/factory/workflow/definitions.py b/factory/workflow/definitions.py index eceddeabf..e2569874d 100644 --- a/factory/workflow/definitions.py +++ b/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/.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"}, diff --git a/factory/workflow/executor.py b/factory/workflow/executor.py index 4b24b3e89..b028478ec 100644 --- a/factory/workflow/executor.py +++ b/factory/workflow/executor.py @@ -37,6 +37,7 @@ Verdict, VerdictType, Workflow, + _role_str, ) log = structlog.get_logger() @@ -811,6 +812,8 @@ 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: @@ -818,18 +821,24 @@ async def _run_agent(self, node: AgentNode) -> str: 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, @@ -837,7 +846,7 @@ async def _run_agent(self, node: AgentNode) -> str: ) 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 diff --git a/factory/workflow/primitives.py b/factory/workflow/primitives.py index 8100eef13..add2eb6d1 100644 --- a/factory/workflow/primitives.py +++ b/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 = "" diff --git a/factory/workflow/skill_export.py b/factory/workflow/skill_export.py index ca299cfaf..19dd1671a 100644 --- a/factory/workflow/skill_export.py +++ b/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"", + f"", f"", f"", ] @@ -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 diff --git a/factory/workflow/tool.py b/factory/workflow/tool.py index 4c15e2f5b..2cf0f4e60 100644 --- a/factory/workflow/tool.py +++ b/factory/workflow/tool.py @@ -21,6 +21,7 @@ Study, VerdictType, Workflow, + _role_str, ) from factory.workflow.registry import WorkflowRegistry from factory.workflow.skill_export import _topological_sort @@ -91,6 +92,13 @@ def _rebuild_workflow(cache_data: dict) -> Workflow: from factory.workflow.primitives import AgentRole, Edge, VerdictType from factory.workflow.primitives import NodeType + + def _parse_role(raw: str) -> AgentRole | str: + try: + return AgentRole(raw) + except ValueError: + return raw + nodes: dict[str, NodeType] = {} for nid, info in cache_data["nodes"].items(): ntype = info["type"] @@ -104,7 +112,7 @@ def _rebuild_workflow(cache_data: dict) -> Workflow: if ntype == "AgentNode": nodes[nid] = AgentNode( **common, # type: ignore[arg-type] - role=AgentRole(info["role"]), + role=_parse_role(info["role"]), model=info.get("model", ""), prompt_template=info.get("prompt_template", ""), timeout=info.get("timeout"), @@ -116,7 +124,7 @@ def _rebuild_workflow(cache_data: dict) -> Workflow: evaluator_type=info.get("evaluator_type", "agent"), evaluator_command=info.get("evaluator_command"), gate_prompt=info.get("gate_prompt", ""), - evaluator_role=AgentRole(info["evaluator_role"]) if info.get("evaluator_role") else None, + evaluator_role=_parse_role(info["evaluator_role"]) if info.get("evaluator_role") else None, ) elif ntype == "Study": nodes[nid] = Study( @@ -239,7 +247,7 @@ def tool_init(workflow_name: str, project_path: Path) -> str: "writes": sorted(node.writes), } if isinstance(node, AgentNode): - node_info["role"] = node.role.value + node_info["role"] = _role_str(node.role) node_info["model"] = node.model node_info["prompt_template"] = node.prompt_template node_info["timeout"] = node.timeout @@ -249,7 +257,7 @@ def tool_init(workflow_name: str, project_path: Path) -> str: node_info["evaluator_command"] = node.evaluator_command node_info["gate_prompt"] = node.gate_prompt if node.evaluator_role: - node_info["evaluator_role"] = node.evaluator_role.value + node_info["evaluator_role"] = _role_str(node.evaluator_role) elif isinstance(node, Study): node_info["command"] = node.command node_info["focus"] = node.focus @@ -542,7 +550,7 @@ def _phase_label(nid: str, node: object) -> str: name = nid.replace("_", " ").title() if isinstance(node, AgentNode): - role = node.role.value.replace("_", " ").title() + role = _role_str(node.role).replace("_", " ").title() return role if role.lower() in name.lower() else f"{role} — {name}" elif isinstance(node, GateNode): gate_name = nid.replace("gate_", "").replace("_", " ").title() @@ -674,7 +682,7 @@ def _find_loop_context( continue parts = [f"- **{loop_nid}**"] if isinstance(loop_node, AgentNode): - parts.append(f"(agent: {loop_node.role.value})") + parts.append(f"(agent: {_role_str(loop_node.role)})") if loop_node.reads: parts.append(f"reads: {', '.join(sorted(loop_node.reads))}") if loop_node.writes: @@ -703,7 +711,7 @@ def _format_node_task( lines = [f"Node: {nid}"] if isinstance(node, AgentNode): - role = node.role.value + role = _role_str(node.role) pool_cfg: AgentConfig | None = DEFAULT_AGENT_POOL.get(role) model = node.model or (pool_cfg.model if pool_cfg else "opus") timeout = node.timeout or (pool_cfg.timeout if pool_cfg else 600) @@ -798,7 +806,7 @@ def _fresh(f: Path) -> bool: return session_start <= 0 or f.stat().st_mtime >= session_start if isinstance(node, AgentNode): - role = node.role.value + role = _role_str(node.role) tag = nid.replace(f"{role}_", "").replace(role, "") if tag and tag != nid: tagged_file = reviews_dir / f"{role}-{tag}-latest.md" diff --git a/factory/workflow/verification.py b/factory/workflow/verification.py index fc0003857..44ddb9a70 100644 --- a/factory/workflow/verification.py +++ b/factory/workflow/verification.py @@ -12,7 +12,7 @@ from pathlib import Path from typing import Any -from factory.workflow.primitives import AgentNode, ArtifactCheck, Workflow +from factory.workflow.primitives import AgentNode, ArtifactCheck, Workflow, _role_str def checks_to_bash(checks: list[ArtifactCheck], node_id: str) -> str: @@ -128,7 +128,7 @@ def generate_hook_script(workflow: Workflow) -> str: verify = compile_agent_verification(node) if not verify: continue - role = node.role.value + role = _role_str(node.role) agent_checks.append((role, verify)) if not agent_checks: diff --git a/tests/test_open_agent_roles.py b/tests/test_open_agent_roles.py new file mode 100644 index 000000000..f851cfe86 --- /dev/null +++ b/tests/test_open_agent_roles.py @@ -0,0 +1,243 @@ +"""Tests for open agent roles — AgentRole | str union type, three-tier resolution, +sandbox default, JSON roundtrip, pool fallback, and existing workflow regression.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +import pytest + +from factory.workflow.primitives import ( + AgentConfig, + AgentNode, + AgentRole, + DEFAULT_AGENT_POOL, + GateNode, + _role_str, +) + + +class TestRoleStrHelper: + def test_enum_returns_value(self) -> None: + assert _role_str(AgentRole.RESEARCHER) == "researcher" + + def test_string_passes_through(self) -> None: + assert _role_str("security_auditor") == "security_auditor" + + +class TestAgentRoleUnionType: + def test_agent_node_accepts_enum(self) -> None: + node = AgentNode(id="r", role=AgentRole.RESEARCHER) + assert node.role == AgentRole.RESEARCHER + + def test_agent_node_accepts_string(self) -> None: + node = AgentNode(id="s", role="security_auditor") + assert node.role == "security_auditor" + + def test_agent_config_accepts_enum(self) -> None: + cfg = AgentConfig(role=AgentRole.BUILDER, model="opus") + assert cfg.role == AgentRole.BUILDER + + def test_agent_config_accepts_string(self) -> None: + cfg = AgentConfig(role="security_auditor", model="sonnet") + assert cfg.role == "security_auditor" + + def test_gate_node_evaluator_role_accepts_string(self) -> None: + gate = GateNode(id="g", evaluator_type="agent", evaluator_role="custom_reviewer") + assert gate.evaluator_role == "custom_reviewer" + + def test_gate_node_evaluator_role_accepts_enum(self) -> None: + gate = GateNode(id="g", evaluator_type="agent", evaluator_role=AgentRole.CEO) + assert gate.evaluator_role == AgentRole.CEO + + def test_gate_node_evaluator_role_accepts_none(self) -> None: + gate = GateNode(id="g", evaluator_type="fn") + assert gate.evaluator_role is None + + def test_pydantic_validation_survives(self) -> None: + node = AgentNode(id="x", role="my_custom_agent", prompt_template="do stuff") + data = node.model_dump() + restored = AgentNode.model_validate(data) + assert restored.role == "my_custom_agent" + + +class TestThreeTierPromptResolution: + def test_project_override_found_first(self, tmp_path: Path) -> None: + from factory.agents.runner import resolve_prompt + + project = tmp_path / "proj" + project.mkdir() + agents_dir = project / ".factory" / "agents" + agents_dir.mkdir(parents=True) + (agents_dir / "custom_role.md").write_text("# Project prompt") + + user_dir = tmp_path / "user_agents" + user_dir.mkdir() + (user_dir / "custom_role.md").write_text("# User prompt") + + with patch("factory.agents.runner._USER_AGENTS_DIR", user_dir): + prompt = resolve_prompt("custom_role", project) + assert "Project prompt" in prompt + + def test_user_override_found_second(self, tmp_path: Path) -> None: + from factory.agents.runner import resolve_prompt + + project = tmp_path / "proj" + project.mkdir() + + user_dir = tmp_path / "user_agents" + user_dir.mkdir() + (user_dir / "custom_role.md").write_text("# User prompt") + + with patch("factory.agents.runner._USER_AGENTS_DIR", user_dir): + prompt = resolve_prompt("custom_role", project) + assert "User prompt" in prompt + + def test_builtin_found_third(self) -> None: + from factory.agents.runner import resolve_prompt + + prompt = resolve_prompt("researcher") + assert len(prompt) > 0 + + def test_all_miss_raises_with_paths(self, tmp_path: Path) -> None: + from factory.agents.runner import resolve_prompt + + project = tmp_path / "proj" + project.mkdir() + user_dir = tmp_path / "nonexistent_agents" + + with patch("factory.agents.runner._USER_AGENTS_DIR", user_dir): + with pytest.raises(FileNotFoundError, match="Searched:"): + resolve_prompt("totally_unknown_role_xyz", project) + + +class TestSandboxModeDefault: + def test_custom_role_returns_workspace_write(self) -> None: + from factory.agents.plugin import _sandbox_mode + + assert _sandbox_mode("custom_agent") == "workspace-write" + + def test_builtin_roles_unchanged(self) -> None: + from factory.agents.plugin import _READ_ONLY_ROLES, _WORKSPACE_WRITE_ROLES, _sandbox_mode + + for role in _READ_ONLY_ROLES: + assert _sandbox_mode(role) == "read-only" + for role in _WORKSPACE_WRITE_ROLES: + assert _sandbox_mode(role) == "workspace-write" + + +class TestToolModeJsonRoundtrip: + def test_agent_node_with_custom_role_survives_json(self) -> None: + from factory.workflow.tool import _rebuild_workflow + + cache_data = { + "name": "test", + "start_node": "custom", + "nodes": { + "custom": { + "type": "AgentNode", + "id": "custom", + "role": "security_auditor", + "model": "sonnet", + "prompt_template": "audit security", + "timeout": 300, + "max_iterations": 1, + "blocking": True, + "reads": [], + "writes": [], + }, + }, + "edges": [], + } + wf = _rebuild_workflow(cache_data) + rebuilt_node = wf.nodes["custom"] + assert isinstance(rebuilt_node, AgentNode) + assert rebuilt_node.role == "security_auditor" + + def test_gate_node_with_custom_evaluator_role(self) -> None: + from factory.workflow.tool import _rebuild_workflow + + cache_data = { + "name": "test", + "start_node": "gate", + "nodes": { + "gate": { + "type": "GateNode", + "id": "gate", + "evaluator_type": "agent", + "evaluator_role": "custom_reviewer", + "evaluator_command": None, + "gate_prompt": "review it", + "blocking": True, + "reads": [], + "writes": [], + }, + }, + "edges": [], + } + wf = _rebuild_workflow(cache_data) + gate = wf.nodes["gate"] + assert isinstance(gate, GateNode) + assert gate.evaluator_role == "custom_reviewer" + + def test_builtin_role_roundtrips_as_enum(self) -> None: + from factory.workflow.tool import _rebuild_workflow + + cache_data = { + "name": "test", + "start_node": "r", + "nodes": { + "r": { + "type": "AgentNode", + "id": "r", + "role": "researcher", + "model": "sonnet", + "prompt_template": "", + "timeout": 600, + "max_iterations": 1, + "blocking": True, + "reads": [], + "writes": [], + }, + }, + "edges": [], + } + wf = _rebuild_workflow(cache_data) + assert wf.nodes["r"].role == AgentRole.RESEARCHER + + +class TestDefaultPoolFallback: + def test_unknown_role_not_in_pool(self) -> None: + assert DEFAULT_AGENT_POOL.get("security_auditor") is None + + def test_known_role_in_pool(self) -> None: + assert DEFAULT_AGENT_POOL.get("researcher") is not None + assert DEFAULT_AGENT_POOL["researcher"].model == "sonnet" + + def test_agent_config_with_custom_role(self) -> None: + cfg = AgentConfig(role="custom_agent", model="sonnet", timeout=600) + assert cfg.model == "sonnet" + assert cfg.timeout == 600 + + +class TestExistingWorkflowsUnchanged: + def test_all_registered_workflows_validate(self) -> None: + from factory.workflow.definitions import register_all + + all_wf = register_all() + for name, wf in all_wf.items(): + issues = wf.validate_graph() + assert issues == [], f"workflow '{name}' has validation issues: {issues}" + + def test_all_builtin_nodes_use_agent_role_enum(self) -> None: + from factory.workflow.definitions import register_all + + all_wf = register_all() + for name, wf in all_wf.items(): + for nid, node in wf.nodes.items(): + if isinstance(node, AgentNode): + assert isinstance(node.role, AgentRole), ( + f"workflow '{name}' node '{nid}' uses string role '{node.role}' " + f"instead of AgentRole enum" + ) diff --git a/tests/test_plugin_agents.py b/tests/test_plugin_agents.py index 1d18df551..2eac1c953 100644 --- a/tests/test_plugin_agents.py +++ b/tests/test_plugin_agents.py @@ -204,9 +204,8 @@ def test_all_known_roles_covered(self): f"{role} is in both _READ_ONLY_ROLES and _WORKSPACE_WRITE_ROLES" ) - def test_unknown_role_raises(self): - with pytest.raises(ValueError, match="Unknown role"): - _sandbox_mode("nonexistent_role") + def test_unknown_role_defaults_to_workspace_write(self): + assert _sandbox_mode("nonexistent_role") == "workspace-write" def test_researcher_is_read_only(self): assert _sandbox_mode("researcher") == "read-only" From 90906f24c0694e59e6ff1ce11352ad176b877ba8 Mon Sep 17 00:00:00 2001 From: colehurwitz Date: Wed, 12 Aug 2026 14:04:03 -0400 Subject: [PATCH 2/3] fix: remove duplicate _find_loop_context, fix call signature and role handling - Remove old 4-param _find_loop_context (shadowed by new 5-param version) - Update _format_node_task call to pass topo order argument - Use _role_str(node.role) instead of node.role.value for custom role support - Remove extraneous f-string prefix on non-interpolated string Co-Authored-By: Claude Opus 4.6 --- factory/workflow/tool.py | 275 +++++++++++++++++++++++++-------------- 1 file changed, 178 insertions(+), 97 deletions(-) diff --git a/factory/workflow/tool.py b/factory/workflow/tool.py index 2cf0f4e60..ddb6bb348 100644 --- a/factory/workflow/tool.py +++ b/factory/workflow/tool.py @@ -377,7 +377,13 @@ def tool_next(project_path: Path, dry_run: bool = False) -> str: if isinstance(node, GateNode) and node.evaluator_type == "user": return f"APPROVAL_NEEDED\n{node.gate_prompt}" - return _format_node_task(nid, node, wf, state, project_path) + result = _format_node_task(nid, node, wf, state, project_path) + + loop_ctx = _find_loop_context(nid, wf, order, state, project_path) + if loop_ctx: + result += f"\n\n{loop_ctx}" + + return result def tool_submit(project_path: Path, node_id: str, output: str) -> str: @@ -609,101 +615,6 @@ def _format_progress( return "\n".join(lines) -def _find_loop_context( - nid: str, wf: Workflow, state: dict, project_path: Path, -) -> str: - """Build a LOOP CONTEXT section for a node that is a RELOOP target. - - Returns an empty string when nid is not a reloop target. Otherwise, - always returns a markdown section showing the loop topology, gate - criteria, and iteration count — even on the first invocation (iteration - 0). The feedback history subsection is only included when feedback - entries exist (i.e. after at least one RELOOP). - """ - reloop_edges = [ - e for e in wf.edges - if e.target == nid and e.condition == VerdictType.RELOOP - ] - if not reloop_edges: - return "" - - topo = state.get("topo_order", []) - iteration_counts = state.get("iteration_counts", {}) - feedback_log = state.get("feedback_log", {}) - - from factory.workflow.primitives import Edge as _Edge - latest_entry: dict | None = None - latest_gate_edge: _Edge | None = None - entries_for_node = feedback_log.get(nid, []) - if entries_for_node: - latest_entry = max(entries_for_node, key=lambda e: e.get("timestamp", 0)) - for e in reloop_edges: - if e.source == latest_entry.get("gate"): - latest_gate_edge = e - break - - active_edge = latest_gate_edge or reloop_edges[0] - gate_id = active_edge.source - iter_key = f"{gate_id}->{nid}" - count = iteration_counts.get(iter_key, 0) - max_iter = 3 - - lines: list[str] = [ - "", - "## LOOP CONTEXT", - f"Iteration: {count}/{max_iter}", - ] - if count >= max_iter: - lines.append("⚠ FINAL ATTEMPT — this is the last iteration before HALT") - - gate_node = wf.nodes.get(gate_id) - lines.append(f"Triggered by: {gate_id}") - if isinstance(gate_node, GateNode): - if gate_node.gate_prompt: - prompt_text = gate_node.gate_prompt.replace("{project_path}", str(project_path)) - lines.append(f"Gate criteria: {prompt_text}") - if gate_node.evaluator_command: - cmd_text = gate_node.evaluator_command.replace("{project_path}", str(project_path)) - lines.append(f"Gate command: {cmd_text}") - - try: - nid_idx = topo.index(nid) - gate_idx = topo.index(gate_id) - except ValueError: - nid_idx = gate_idx = -1 - - if 0 <= nid_idx < gate_idx: - loop_path = topo[nid_idx:gate_idx + 1] - lines.append("") - lines.append("### Loop topology") - for loop_nid in loop_path: - loop_node = wf.nodes.get(loop_nid) - if loop_node is None: - continue - parts = [f"- **{loop_nid}**"] - if isinstance(loop_node, AgentNode): - parts.append(f"(agent: {_role_str(loop_node.role)})") - if loop_node.reads: - parts.append(f"reads: {', '.join(sorted(loop_node.reads))}") - if loop_node.writes: - parts.append(f"writes: {', '.join(sorted(loop_node.writes))}") - elif isinstance(loop_node, GateNode): - parts.append(f"(gate: {loop_node.evaluator_type})") - elif isinstance(loop_node, FnNode): - parts.append("(fn)") - lines.append(" ".join(parts)) - - if entries_for_node: - lines.append("") - lines.append("### Feedback history") - recent = sorted(entries_for_node, key=lambda e: e.get("timestamp", 0))[-2:] - for entry in recent: - fb_text = entry.get("feedback", "")[:500] - lines.append(f"- [{entry.get('gate', '?')} iter {entry.get('iteration', '?')}] {fb_text}") - - return "\n".join(lines) - - def _format_node_task( nid: str, node: object, wf: Workflow, state: dict, project_path: Path, ) -> str: @@ -756,7 +667,7 @@ def _format_node_task( lines.append(f"Targets: {', '.join(node.targets)}") lines.append("Execute all targets (listed as subsequent nodes).") - loop_ctx = _find_loop_context(nid, wf, state, project_path) + loop_ctx = _find_loop_context(nid, wf, state.get("topo_order", []), state, project_path) if loop_ctx: lines.append(loop_ctx) @@ -922,6 +833,176 @@ def _auto_evaluate_fn_gate( return None +def tool_peek(project_path: Path, node_id: str) -> str: + """Return full details for any node without advancing the cursor. + + Shows gate criteria, reads/writes, reloop targets, max iterations — + everything the CEO needs to understand what a downstream node does. + """ + state = _load_state(project_path) + wf = _get_workflow_cached(state["workflow_name"], project_path) + + node = wf.nodes.get(node_id) + if node is None: + return f"Unknown node: {node_id}" + + lines = _format_node_task(node_id, node, wf, state, project_path).split("\n") + + if isinstance(node, GateNode): + reloop_targets: list[str] = [] + halt_targets: list[str] = [] + for edge in wf.edges: + if edge.source == node_id: + if edge.condition == VerdictType.RELOOP: + reloop_targets.append(edge.target) + elif edge.condition == VerdictType.HALT: + halt_targets.append(edge.target) + if reloop_targets: + lines.append(f"Reloop to: {', '.join(reloop_targets)}") + if halt_targets: + lines.append(f"Halt skips to: {', '.join(halt_targets)}") + + iter_key_prefix = f"{node_id}->" + for k, v in state.get("iteration_counts", {}).items(): + if k.startswith(iter_key_prefix): + lines.append(f"Iterations used: {v}/3") + + if isinstance(node, AgentNode) and node.max_iterations > 1: + lines.append(f"Max iterations: {node.max_iterations}") + + status = "completed" if node_id in state["completed"] else ( + "current" if state["topo_order"][state["pointer_idx"]] == node_id + and state["pointer_idx"] < len(state["topo_order"]) else "pending" + ) + lines.append(f"Status: {status}") + + return "\n".join(lines) + + +def tool_lookahead(project_path: Path, count: int = 5) -> str: + """Return the next N nodes with full details, up to and including the next gate. + + Gives the CEO downstream visibility without advancing the cursor. + """ + state = _load_state(project_path) + wf = _get_workflow_cached(state["workflow_name"], project_path) + order = state["topo_order"] + idx = state["pointer_idx"] + + if idx >= len(order): + return "DONE — no nodes ahead." + + sections: list[str] = [] + shown = 0 + for i in range(idx + 1, len(order)): + if shown >= count: + remaining = len(order) - i + sections.append(f"... {remaining} more node(s)") + break + + nid = order[i] + node = wf.nodes.get(nid) + if node is None: + continue + + detail = tool_peek(project_path, nid) + sections.append(detail) + shown += 1 + + return "\n\n---\n\n".join(sections) if sections else "No nodes ahead." + + +def _find_loop_context( + nid: str, wf: Workflow, order: list[str], state: dict, project_path: Path, +) -> str | None: + """If nid is a RELOOP target, return the full loop context. + + Traces from nid forward through the topo order until it reaches the gate + that RELOOP's back to nid. Returns a formatted description of every node + in the loop — gates with their criteria, agents with their roles and + artifact contracts, etc. + """ + reloop_gates: list[str] = [] + for edge in wf.edges: + if edge.condition == VerdictType.RELOOP and edge.target == nid: + reloop_gates.append(edge.source) + + if not reloop_gates: + return None + + try: + nid_idx = order.index(nid) + except ValueError: + return None + + gate_indices = [] + for g in reloop_gates: + try: + gate_indices.append((order.index(g), g)) + except ValueError: + pass + if not gate_indices: + return None + + farthest_gate_idx, farthest_gate = max(gate_indices, key=lambda x: x[0]) + + loop_nodes: list[str] = [] + for i in range(nid_idx + 1, farthest_gate_idx + 1): + loop_nodes.append(order[i]) + + if not loop_nodes: + return None + + lines = [ + f"LOOP CONTEXT — after {nid}, your output goes through these steps " + f"(failures RELOOP back to {nid}):", + "", + ] + + for loop_nid in loop_nodes: + node = wf.nodes.get(loop_nid) + if node is None: + continue + + if isinstance(node, GateNode): + gate_type = node.evaluator_type + lines.append(f" [{loop_nid}] Gate ({gate_type})") + if node.gate_prompt: + prompt = node.gate_prompt.replace("{project_path}", str(project_path)) + lines.append(f" Criteria: {prompt}") + if node.evaluator_command: + cmd = node.evaluator_command.replace("{project_path}", str(project_path)) + lines.append(f" Command: {cmd}") + if node.reads: + lines.append(f" Reads: {', '.join(sorted(node.reads))}") + for edge in wf.edges: + if edge.source == loop_nid and edge.condition == VerdictType.RELOOP: + lines.append(f" On failure: RELOOP to {edge.target}") + elif edge.source == loop_nid and edge.condition == VerdictType.HALT: + lines.append(" On HALT: skip remaining loop steps") + + elif isinstance(node, AgentNode): + role = _role_str(node.role) + lines.append(f" [{loop_nid}] Agent: {role}") + if node.prompt_template: + task = node.prompt_template.replace("{project_path}", str(project_path)) + lines.append(f" Will check: {task[:200]}") + if node.reads: + lines.append(f" Reads: {', '.join(sorted(node.reads))}") + if node.writes: + lines.append(f" Writes: {', '.join(sorted(node.writes))}") + + elif isinstance(node, FnNode): + lines.append(f" [{loop_nid}] Function") + if node.command: + cmd = node.command.replace("{project_path}", str(project_path)) + lines.append(f" Command: {cmd}") + + lines.append("") + + return "\n".join(lines) + + def _find_reloop_target(wf: Workflow, gate_id: str) -> str | None: """Find the RELOOP target for a gate node.""" for edge in wf.edges: From db4205ab1c2c0e35ed8af26d01359bf43f385a32 Mon Sep 17 00:00:00 2001 From: colehurwitz Date: Wed, 12 Aug 2026 15:00:28 -0400 Subject: [PATCH 3/3] docs: update CLAUDE.md prompt lookup to three-tier, fix F541 lint warnings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CLAUDE.md: update agent prompt resolution from 'two-tier lookup' to 'three-tier lookup' (project → user-level → factory default), note that roles accept AgentRole enum values or custom strings - factory/cli/_ceo_helpers.py: remove extraneous f-string prefixes on lines 130-131 (ruff F541) Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 2 +- factory/cli/_ceo_helpers.py | 35 ++++++++++++++++++++++++++++++++++- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 854cdb034..e9b5e943c 100644 --- a/CLAUDE.md +++ b/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 `. Agent prompts are resolved via `factory/agents/runner.py` with a two-tier lookup: project-specific override (`.factory/agents/.md`) then factory default (`factory/agents/prompts/.md`). Evolved playbooks from `~/.factory/playbooks/.md` (user-local, ACE-generated) are auto-injected, falling back to factory defaults in `factory/agents/playbooks/.md`. +Eight specialist Claude Code subprocesses spawned by the CEO via `factory agent `. 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/.md`) → user-level (`~/.factory/agents/.md`) → factory default (`factory/agents/prompts/.md`). Evolved playbooks from `~/.factory/playbooks/.md` (user-local, ACE-generated) are auto-injected, falling back to factory defaults in `factory/agents/playbooks/.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). diff --git a/factory/cli/_ceo_helpers.py b/factory/cli/_ceo_helpers.py index 731d0c645..e5ab63a98 100644 --- a/factory/cli/_ceo_helpers.py +++ b/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 \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 " to inspect any specific node, ' + 'or "lookahead --count N" to see the next N nodes.\n' + ) + + protocol += ( "\n" "## Important\n" "\n"