"""Deep research parallel pipeline — breadth fan-out -> depth fan-out -> compile.
Runs fork_breadth (5 parallel angle scans) -> join_breadth -> breadth_synthesis
(names up to 4 priority areas) -> fork_depth (up to 4 parallel deep dives) ->
join_depth -> compile (opus, writes the final report) -> gate_compile (CEO
acceptance, RELOOP to compile only). No Study node — must run with no project
present (bare-topic research), mirroring research.py's research-standalone.
Terminal mode. Triggered via `factory workflow run deepresearch` or
`factory ceo <topic-or-path> --mode deepresearch`.
Project-local mode (Path B): this file lives under `.factory/workflows/` and is
discovered by `WorkflowRegistry.discover()` as a `source="project"` entry — it
is intentionally NOT registered in `factory/workflow/definitions.py`.
"""
from typing import Any
from factory.models import ProjectState
from factory.workflow.primitives import (
AgentNode,
AgentRole,
ArtifactCheck,
Edge,
ForkNode,
GateNode,
JoinNode,
VerdictType,
Workflow,
)
meta = {
"name": "deepresearch",
"description": (
"Multi-stage parallel research pipeline — up to 5 breadth agents scan a topic "
"from different angles (source-type, ecosystem, or architectural concern), a "
"synthesis step names up to 4 priority areas, up to 4 depth agents dive into "
"those areas, and a single opus-model compiler produces one citation-rich, "
"template-structured markdown report. Works with no existing project. "
"Terminal mode."
),
}
_BASE = ".factory/strategy/deepresearch/{topic_slug}"
_TOPIC_SLUG_PREAMBLE = (
"## Step 0 — derive the topic and slug (do this identically every time; other "
"parallel agents in this run depend on landing on the exact same value)\n"
"1. TOPIC = the research topic as given verbatim in the CEO's task / --focus text.\n"
"2. topic_slug = TOPIC, lowercased, with every run of characters that are not "
"[a-z0-9] replaced by a single underscore, then leading/trailing underscores "
"stripped, then truncated to 50 characters.\n"
" Example: \"vLLM Speculative Decoding!\" -> \"vllm_speculative_decoding\"\n"
"3. Your run directory is `.factory/strategy/deepresearch/<topic_slug>/` — create "
"it if it doesn't exist (mkdir -p).\n"
"4. If any generated verification command below contains the literal string "
"`{topic_slug}` (a placeholder, not a resolved path), replace it with your actual "
"derived slug before running that command — it will never match the literal text.\n"
)
_BREADTH_ANGLE_TABLE = (
"| Slot | library_or_tool | concept_or_technique | codebase_area |\n"
"|---|---|---|---|\n"
"| A{a_mark} | Web: official docs, guides, changelogs | Web: papers, technical "
"blog posts, manufacturer docs | Explore: file structure, entry points, exports |\n"
"| B{b_mark} | Web: GitHub issues, discussions, known bugs | Web: implementation "
"examples, benchmarks, hardware compatibility | Explore: data flow — trace inputs "
"through transformations to outputs |\n"
"| C{c_mark} | Web: tutorials, benchmarks, comparisons | Web: community experience "
"(forums, GitHub discussions, HN) | Explore: dependencies — what it imports, what "
"imports it |\n"
"| D{d_mark} | Codebase: source exploration (entry points, APIs, data structures, "
"config schemas) if installed/cloned locally, else stub | Codebase: existing "
"implementations in current project/installed packages, else stub | Explore: tests "
"— behavior tested, edge cases covered |\n"
"| E{e_mark} | Codebase: existing usage patterns/integrations in the current "
"project if relevant, else stub | stub (not used for this topic type) | stub (not "
"used for this topic type) |\n"
)
_BREADTH_SLOTS = ("a", "b", "c", "d", "e")
def _breadth_prompt(slot: str) -> str:
letter = slot.upper()
marks = {s: (" (YOU)" if s == slot else "") for s in _BREADTH_SLOTS}
table = _BREADTH_ANGLE_TABLE.format(
a_mark=marks["a"], b_mark=marks["b"], c_mark=marks["c"],
d_mark=marks["d"], e_mark=marks["e"],
)
return (
f"You are Breadth Agent {letter} of a 5-agent parallel research fan-out "
f"(slots A-E). Some slots may be unused this run — that is expected, not a "
f"failure.\n\n"
f"{_TOPIC_SLUG_PREAMBLE}"
f"5. Your output file is `{_BASE}/breadth-{slot}.md`.\n\n"
"## Step 1 — classify the topic and pick your angle\n"
"Classify TOPIC into exactly one of: **library_or_tool**, "
"**concept_or_technique**, or **codebase_area** (see skill contract for "
"definitions). Find your slot in the table below. If your slot has no defined "
"angle for this topic type (a \"stub\" cell), STOP: write exactly one line — "
f"`N/A for this topic (<type>) — slot {letter} not needed` — to your output "
"file using the Write tool, and finish immediately.\n\n"
f"{table}\n"
"## Step 2 — do the research (skip if you stubbed)\n"
"Budget: at most 8 WebSearch/WebFetch calls (or Read/Grep calls for "
"codebase_area). Reserve your last call for writing the output file.\n\n"
"## Step 3 — write your output file\n"
"Write, using the Write tool: **Key findings** (max 15 bullets), **Source "
"citations**, **Gaps identified**, **Surprises**. Do NOT rely on your return "
"message — the synthesis step only reads this file.\n"
)
_BREADTH_SYNTHESIS_PROMPT = (
"You are the Breadth Synthesis agent. You read every breadth output file for this "
"run and produce the one artifact both the depth stage and the compile stage "
"depend on.\n\n"
f"{_TOPIC_SLUG_PREAMBLE}"
"## Step 1 — read all breadth files\n"
"Read breadth-a.md through breadth-e.md in your run directory. Some may be "
"one-line stubs — that is expected, not missing data.\n\n"
"## Step 2 — synthesize\n"
"Identify: (1) the 3-4 most important areas needing deeper investigation, each a "
"short imperative title; (2) contradictions between sources; (3) critical gaps.\n\n"
"## Step 3 — write breadth-summary.md\n"
"Write to `<run_dir>/breadth-summary.md` with EXACTLY this structure — the depth "
"stage parses the numbered list positionally, cap it at 4 items:\n\n"
"# Breadth Summary — <topic>\n\n"
"## Areas for deep dive\n"
"1. <title> — <why>\n2. <title> — <why>\n3. <title> — <why>\n4. <optional> — <why>\n\n"
"## Contradictions\n- <contradiction, naming sources>\n\n"
"## Gaps\n- <gap>\n"
)
_DEPTH_SLOTS = {"a": 1, "b": 2, "c": 3, "d": 4}
def _depth_prompt(slot: str) -> str:
letter, idx = slot.upper(), _DEPTH_SLOTS[slot]
return (
f"You are Depth Agent {letter} (area #{idx}) of up to 4 parallel deep-dive "
f"agents.\n\n"
f"{_TOPIC_SLUG_PREAMBLE}"
f"5. Read `{_BASE}/breadth-summary.md`. Find the numbered list under \"Areas "
f"for deep dive\". Your area is **item {idx}** in that list.\n"
f"6. If the list has fewer than {idx} items, STOP: write exactly one line — "
f"`N/A — breadth-summary named fewer than {idx} deep-dive areas` — to your "
"output file and finish immediately.\n"
f"7. Your output file is `{_BASE}/depth-{slot}.md`.\n\n"
"## Step 1 — investigate your area thoroughly\n"
"Read full source files, trace call chains, read complete docs pages — do not "
"skim. Budget: at most 12 WebSearch/WebFetch/Read calls, reserve the last for "
"writing.\n\n"
"## Step 2 — write your output file\n"
"Write: **Detailed findings** with exact citations, **Code examples/API "
"signatures**, **Gotchas and pitfalls**, **Practical recommendations**. Do NOT "
"rely on your return message.\n"
)
_COMPILE_PROMPT = (
"You are the Compile agent — sole author of the final research report.\n\n"
f"{_TOPIC_SLUG_PREAMBLE}"
"5. Default output path: `docs/RESEARCH_<topic_slug>.md`. If the CEO's task / "
"--focus text names an output path explicitly, use that instead.\n\n"
"## Step 1 — read all inputs\n"
"Read breadth-summary.md, breadth-{a..e}.md, depth-{a..d}.md (skip stubs). A "
"partial input set is normal — compile what exists; never refuse for incomplete "
"inputs.\n\n"
"## Step 2 — write the report using EXACTLY this structure (no empty sections):\n\n"
"# [Topic] — Deep Research Report\n\n> Generated: [date] | Sources: [N web + M "
"code]\n\n## TL;DR\n## Overview\n## Key Findings\n### [Finding Area 1]\n"
"### [Finding Area N]\n## Practical Guide\n## Gotchas & Pitfalls\n## Sources\n\n"
"Every factual claim needs a citation. Prefer concrete commands/snippets over "
"abstractions. Note contradictions and which source to trust (source code > "
"official docs > community posts).\n\n"
"## RELOOP handling\n"
"If re-running after gate_compile RELOOPed, read the latest CEO verdict under "
".factory/reviews/ for the gap list. Re-read the same source files — do not "
"re-run research.\n\n"
"State the final report path in your summary.\n"
)
_GATE_COMPILE_PROMPT = (
"Evaluate the compiled research report for acceptance.\n\n"
"Read breadth-summary.md and the final report (default docs/RESEARCH_<slug>.md, "
"or the user-overridden path).\n\n"
"Check: (1) every breadth-summary area got a depth dive or an explicit reason it "
"didn't; (2) every factual claim is cited; (3) the report has all required "
"sections (TL;DR, Overview, Key Findings, Practical Guide, Gotchas & Pitfalls, "
"Sources); (4) no section is empty/placeholder.\n\n"
"PROCEED if all four pass. RELOOP to compile (never to breadth/depth) listing "
"exactly which checks failed."
)
def workflow() -> Workflow:
"""Breadth fan-out -> synthesis -> depth fan-out -> compile -> CEO gate."""
nodes: dict[str, Any] = {}
nodes["fork_breadth"] = ForkNode(
id="fork_breadth",
targets=[f"breadth_{s}" for s in _BREADTH_SLOTS],
)
for s in _BREADTH_SLOTS:
write_path = f"{_BASE}/breadth-{s}.md"
nodes[f"breadth_{s}"] = AgentNode(
id=f"breadth_{s}",
role=AgentRole.RESEARCHER,
prompt_template=_breadth_prompt(s),
writes={write_path},
post_checks=[ArtifactCheck(path=write_path, must_exist=True, min_size=200)],
timeout=900,
)
nodes["join_breadth"] = JoinNode(
id="join_breadth",
sources=[f"breadth_{s}" for s in _BREADTH_SLOTS],
)
nodes["breadth_synthesis"] = AgentNode(
id="breadth_synthesis",
role=AgentRole.RESEARCHER,
prompt_template=_BREADTH_SYNTHESIS_PROMPT,
reads={f"{_BASE}/breadth-{s}.md" for s in _BREADTH_SLOTS},
writes={f"{_BASE}/breadth-summary.md"},
post_checks=[
ArtifactCheck(path=f"{_BASE}/breadth-summary.md", must_exist=True, min_size=200)
],
timeout=600,
)
nodes["fork_depth"] = ForkNode(
id="fork_depth",
targets=[f"depth_{s}" for s in _DEPTH_SLOTS],
)
for s in _DEPTH_SLOTS:
write_path = f"{_BASE}/depth-{s}.md"
nodes[f"depth_{s}"] = AgentNode(
id=f"depth_{s}",
role=AgentRole.RESEARCHER,
prompt_template=_depth_prompt(s),
reads={f"{_BASE}/breadth-summary.md"},
writes={write_path},
post_checks=[ArtifactCheck(path=write_path, must_exist=True, min_size=200)],
timeout=1200,
)
nodes["join_depth"] = JoinNode(
id="join_depth",
sources=[f"depth_{s}" for s in _DEPTH_SLOTS],
)
nodes["compile"] = AgentNode(
id="compile",
role=AgentRole.RESEARCHER,
model="opus",
prompt_template=_COMPILE_PROMPT,
reads=(
{f"{_BASE}/breadth-summary.md"}
| {f"{_BASE}/breadth-{s}.md" for s in _BREADTH_SLOTS}
| {f"{_BASE}/depth-{s}.md" for s in _DEPTH_SLOTS}
),
writes={"docs/RESEARCH_{topic_slug}.md"},
post_checks=[
ArtifactCheck(path="docs/RESEARCH_{topic_slug}.md", must_exist=True, min_size=800)
],
timeout=900,
)
nodes["gate_compile"] = GateNode(
id="gate_compile",
evaluator_type="agent",
evaluator_role=AgentRole.CEO,
gate_prompt=_GATE_COMPILE_PROMPT,
reads={f"{_BASE}/breadth-summary.md", "docs/RESEARCH_{topic_slug}.md"},
)
edges = [
*[Edge(source="fork_breadth", target=f"breadth_{s}") for s in _BREADTH_SLOTS],
*[Edge(source=f"breadth_{s}", target="join_breadth") for s in _BREADTH_SLOTS],
Edge(source="join_breadth", target="breadth_synthesis"),
Edge(source="breadth_synthesis", target="fork_depth"),
*[Edge(source="fork_depth", target=f"depth_{s}") for s in _DEPTH_SLOTS],
*[Edge(source=f"depth_{s}", target="join_depth") for s in _DEPTH_SLOTS],
Edge(source="join_depth", target="compile"),
Edge(source="compile", target="gate_compile"),
Edge(source="gate_compile", target="compile", condition=VerdictType.RELOOP),
]
def trigger(state: ProjectState, ctx: dict[str, Any]) -> bool:
return ctx.get("mode") == "deepresearch"
return Workflow(
name="deepresearch",
nodes=nodes,
edges=edges,
start_node="fork_breadth",
trigger=trigger,
terminal=True,
)2026-08-26 16:33:39 [info ] workflow_registry.discovered count=15
Error: --focus (targeted mode) only works in design, research, create, evolve, study, frontend-design, frontend-design-discover, or design (with --just-plan) mode, got 'deepresearch'. The project must already be built before targeting specific items.
It should use that as input for the mode.
Commit Hash: b3cb950
Reproduction:
.factory/workflows/deepresearch.pyscriptBash CMD:
Output:
Expected Behaviour:
It should use that as input for the mode.