# Master Control Research Integration
**Technical Implementation Guide**
**Version:** 1.0
**Date:** 2026-01-23
---
## Overview
This document specifies how Master Control integrates the Research-First Architecture into its workflow. It describes the exact spawn sequence, prompt templates, and context passing between MC → Scout/Researcher → Planner.
---
## MC Research Phase Flow
### High-Level Sequence
```
User Request
↓
MC receives request
↓
MC checks: Should research run?
↓
├─ NO → Spawn Planner directly (legacy path)
│
└─ YES → Research Phase
↓
┌────┴────┐
↓ ↓
Scout Researcher(s)
│ │
│ (parallel spawn)
│ │
└────┬────┘
↓
Context Assembly
↓
Spawn Planner
(with research context)
```
---
## 1. Research Decision Logic
### MC's Initial Check
```python
def handle_user_request(user_request: str):
"""MC's entry point for all user requests."""
# Load config (if exists)
config = load_config() or get_default_config()
# Determine if research should run
research_decision = should_run_research(user_request, config)
if not research_decision['run']:
# Legacy path - direct to Planner
return spawn_planner(user_request, context={})
# Research-first path
research_context = execute_research_phase(user_request, config)
return spawn_planner(user_request, context=research_context)
```
### Skip Conditions
```python
def should_run_research(user_request: str, config: dict) -> dict:
"""Determine if research phase should execute."""
if not config.get('research', {}).get('enabled', True):
return {'run': False, 'reason': 'Research disabled in config'}
# User explicitly skipped
if 'skip research' in user_request.lower():
return {'run': False, 'reason': 'User requested skip'}
# Quick mode eligible (≤5 files, simple operations)
if quick_mode_eligible(user_request):
return {'run': False, 'reason': 'Quick mode eligible'}
# Debug/fix mode (no new tech involved)
if is_debug_request(user_request):
return {'run': False, 'reason': 'Debug mode'}
# All technologies already cached
techs = extract_technologies(user_request)
if techs and all(is_cached(tech, config) for tech in techs):
return {'run': False, 'reason': 'All tech cached'}
return {'run': True, 'reason': 'Research needed'}
```
---
## 2. Scout Spawning
### When to Spawn Scout
Scout runs when:
- `research.scout.enabled == true` (default)
- Existing codebase detected (`os.listdir(os.getcwd())` shows files)
- Not a greenfield project
### Scout Spawn Template
```python
def spawn_scout(config: dict) -> str:
"""Spawn Scout for codebase reconnaissance."""
scout_config = config.get('research', {}).get('scout', {})
timeout = scout_config.get('timeout_seconds', 120)
max_files = scout_config.get('max_files_analyzed', 1000)
depth_limit = scout_config.get('depth_limit', 4)
skip_dirs = scout_config.get('skip_dirs', ['node_modules', '.git', 'dist', 'build'])
prompt = f"""
First, read ~/.claude/agents/grid-scout.md for your complete role definition.
You are Scout, a reconnaissance program on The Grid. Your mission: rapid codebase analysis in <2 minutes.
## Mission Parameters
project_root: {os.getcwd()}
timeout_seconds: {timeout}
max_files_analyzed: {max_files}
depth_limit: {depth_limit}
skip_directories: {skip_dirs}
## Your Objectives
1. **Structure Scan** (30s max)
- Map directory tree (depth ≤ {depth_limit})
- Count file types
- Identify package managers
2. **Technology Detection** (30s max)
- Parse package.json / pyproject.toml / go.mod / etc.
- Detect frameworks and versions
- Identify databases and ORMs
3. **Pattern Detection** (30s max)
- Recognize architectural patterns (App Router, MVC, etc.)
- Detect coding conventions (naming, imports)
- Identify file organization patterns
4. **Constraint Discovery** (30s max)
- Find locked dependencies
- Locate existing schemas
- Map API routes/contracts
- Note config requirements
## Output Requirements
Write scout report to: `.grid/scout/RECON_{datetime.now().isoformat()}.md`
Report must include:
- Executive summary
- Technology stack table
- Detected patterns with confidence levels
- Constraints that planning MUST respect
- Key files for planning reference
- Recommendations for Planner
## Critical Rules
- SPEED OVER COMPLETENESS - 2 minute hard limit
- Sample first {max_files} files only
- Skip {skip_dirs}
- Parallel operations wherever possible
- Early exit when patterns clear
Begin reconnaissance. End of Line.
"""
# Spawn Scout as Task
scout_result = Task(
prompt=prompt,
subagent_type="general-purpose",
description="Scout: Codebase reconnaissance"
)
# Read the scout report
report_files = glob('.grid/scout/RECON_*.md')
if report_files:
latest_report = sorted(report_files)[-1]
return read(latest_report)
return None
```
---
## 3. Researcher Spawning
### Research Needs Extraction
```python
def extract_research_needs(user_request: str, scout_context: str = None) -> list[dict]:
"""Extract what needs to be researched from user request and Scout findings."""
needs = []
# Extract technologies from user request
techs = extract_technologies(user_request)
for tech in techs:
needs.append({
'type': 'technology',
'topic': tech,
'queries': [
f"{tech} best practices 2024 2025",
f"{tech} production patterns",
f"{tech} common mistakes to avoid"
]
})
# Extract patterns from request
patterns = extract_patterns(user_request)
for pattern in patterns:
needs.append({
'type': 'pattern',
'topic': pattern,
'queries': [
f"{pattern} implementation guide",
f"{pattern} real world examples",
f"when not to use {pattern}"
]
})
# Extract API/integration needs
integrations = extract_integrations(user_request)
for integration in integrations:
needs.append({
'type': 'api_docs',
'topic': integration,
'queries': [
f"{integration} API documentation",
f"{integration} integration examples",
f"{integration} best practices"
]
})
# If Scout detected unfamiliar patterns, research them
if scout_context:
unknown_patterns = extract_unknown_patterns(scout_context)
for pattern in unknown_patterns:
needs.append({
'type': 'pattern',
'topic': pattern,
'queries': [f"{pattern} explanation", f"{pattern} usage"]
})
return needs
```
### Cache Check
```python
def check_research_cache(needs: list[dict], config: dict) -> tuple[list, list]:
"""Check cache for existing research, return (cached, needed)."""
cache_enabled = config.get('research', {}).get('cache', {}).get('enabled', True)
if not cache_enabled:
return [], needs
# Load cache index
cache_index_path = '.grid/research_cache/index.json'
if not os.path.exists(cache_index_path):
return [], needs
with open(cache_index_path) as f:
cache_index = json.load(f)
cached = []
needed = []
for need in needs:
# Check if queries match any cached entry
cache_hit = None
for entry in cache_index.get('entries', []):
# Jaccard similarity > 0.7
if jaccard_similarity(entry['queries'], need['queries']) > 0.7:
# Not expired?
if datetime.now() < datetime.fromisoformat(entry['expires']):
cache_hit = entry
break
if cache_hit:
# Read cached research
cached_path = f".grid/research_cache/{cache_hit['file']}"
with open(cached_path) as f:
cached.append({
'topic': need['topic'],
'content': f.read(),
'source': 'cache'
})
# Update hit count
cache_hit['hit_count'] = cache_hit.get('hit_count', 0) + 1
else:
needed.append(need)
# Save updated index (with new hit counts)
with open(cache_index_path, 'w') as f:
json.dump(cache_index, f, indent=2)
return cached, needed
```
### Researcher Spawn Template
```python
def spawn_researchers(needs: list[dict], config: dict) -> list[str]:
"""Spawn Researcher programs for uncached research needs."""
researcher_config = config.get('research', {}).get('researcher', {})
timeout = researcher_config.get('timeout_seconds', 300)
max_queries = researcher_config.get('max_queries_per_topic', 10)
max_parallel = researcher_config.get('max_researchers_parallel', 3)
tool_preference = researcher_config.get('search_tool_preference', [
'mcp__exa__get_code_context_exa',
'WebSearch',
'WebFetch'
])
# Limit to max parallel
needs = needs[:max_parallel]
research_results = []
for need in needs:
prompt = f"""
First, read ~/.claude/agents/grid-researcher.md for your complete role definition.
You are Researcher, an intelligence gathering program on The Grid. Your mission: gather external context for informed planning.
## Mission Parameters
timeout_seconds: {timeout}
max_queries_per_topic: {max_queries}
search_tools: {tool_preference}
cache_enabled: true
mission_type: {need['type']}
topic: {need['topic']}
suggested_queries: {need['queries']}
## Your Objectives
1. **Query Generation** - Create effective search queries
2. **Parallel Search** - Execute searches using preferred tools
3. **Context Assembly** - Structure findings for Planner consumption
4. **Caching** - Save to `.grid/research_cache/{slugify(need['topic'])}.md`
## Output Requirements
Create structured research context with:
- Executive summary (2-3 sentences)
- Best practices table (practice, source, confidence)
- Recommended patterns (what, when, why, source)
- Anti-patterns to avoid (what, why bad, instead, source)
- Code examples with sources
- API reference (if applicable)
- Confidence assessment (HIGH/MEDIUM/LOW per topic)
- Sources (all URLs cited)
## Cache Output
Write to: `.grid/research_cache/{slugify(need['topic'])}.md`
Update cache index: `.grid/research_cache/index.json`
## Critical Rules
- CITE EVERYTHING - No source = LOW confidence
- RECENT > OLD - Prefer 2024-2025 sources
- OFFICIAL > BLOG - Prefer official docs
- CODE REQUIRED - Include working examples
- TIME-BOX - 5 minutes max, move on if stuck
- PARALLEL SEARCHES - Independent queries at once
- HONEST REPORTING - Say what you couldn't find
Begin research. End of Line.
"""
# Spawn Researcher as Task (they run in parallel)
task = Task(
prompt=prompt,
subagent_type="general-purpose",
description=f"Research: {need['topic']}"
)
research_results.append(task)
# All researchers run in parallel
# MC waits for all to complete, then reads their outputs
# Read research outputs
outputs = []
for need in needs:
cache_file = f".grid/research_cache/{slugify(need['topic'])}.md"
if os.path.exists(cache_file):
with open(cache_file) as f:
outputs.append({
'topic': need['topic'],
'content': f.read(),
'source': 'fresh'
})
return outputs
```
---
## 4. Context Assembly
### Combining Scout + Researcher Outputs
```python
def execute_research_phase(user_request: str, config: dict) -> dict:
"""Execute complete research phase, return assembled context."""
context = {
'research_executed': True,
'scout': None,
'cached_research': [],
'fresh_research': []
}
# Phase 1: Scout (if existing codebase)
if config.get('research', {}).get('scout', {}).get('enabled', True):
if has_existing_code():
context['scout'] = spawn_scout(config)
# Phase 2: Extract research needs
needs = extract_research_needs(user_request, context['scout'])
# Phase 3: Check cache
if config.get('research', {}).get('cache', {}).get('enabled', True):
cached, needed = check_research_cache(needs, config)
context['cached_research'] = cached
else:
needed = needs
# Phase 4: Spawn researchers for uncached needs
if needed and config.get('research', {}).get('researcher', {}).get('enabled', True):
context['fresh_research'] = spawn_researchers(needed, config)
return context
```
---
## 5. Planner Integration
### Planner Spawn with Research Context
```python
def spawn_planner(user_request: str, context: dict):
"""Spawn Planner with research context."""
# Format research context for Planner
codebase_context = context.get('scout', 'No existing codebase detected. Greenfield project.')
research_context = ""
for cached in context.get('cached_research', []):
research_context += f"\n## Cached Research: {cached['topic']}\n{cached['content']}\n"
for fresh in context.get('fresh_research', []):
research_context += f"\n## Fresh Research: {fresh['topic']}\n{fresh['content']}\n"
if not research_context:
research_context = "No external research performed. Use training knowledge."
prompt = f"""
First, read ~/.claude/agents/grid-planner.md for your complete role definition.
You are Planner, a planning program on The Grid. Your mission: create execution plans from User intent, informed by research context.
## Context from Research Phase
{codebase_context}
{research_context}
{user_request}
## Your Objectives
Create an execution plan that:
1. **Respects codebase constraints** (from Scout)
- Preserves existing API contracts
- Follows detected conventions
- Works with locked dependencies
- Extends existing patterns
2. **Leverages research findings** (from Researchers)
- Uses recommended patterns
- Avoids documented anti-patterns
- Follows 2024-2025 best practices
- Incorporates code examples
3. **Decomposes into Blocks and Threads**
- 2-3 Threads per Block (50% context budget)
- Wave numbers for parallel execution
- Dependencies clearly marked
4. **Derives must-haves from goals**
- Goal-backward verification
- Acceptance criteria per Block
## Output Format
Write plan to: `.grid/plans/{cluster}-PLAN-SUMMARY.md`
Then create Block plans: `.grid/plans/{cluster}-block-{N}.md`
## Critical Rules
- CONSTRAINTS ARE SACRED - Scout findings are non-negotiable
- RESEARCH INFORMS - Best practices guide, don't dictate
- CONTEXT BUDGET - 2-3 Threads/Block, never more
- WAVE ASSIGNMENT - Compute during planning, not execution
Begin planning. End of Line.
"""
return Task(
prompt=prompt,
subagent_type="general-purpose",
description="Planner: Create execution plan"
)
```
---
## 6. MC Orchestration Example
### Complete Research-First Flow
```python
def mc_orchestrate_research_first(user_request: str):
"""Master Control's research-first orchestration."""
print("Master Control online.")
print("Analyzing request...")
# Load config
config = load_config() or {
'research': {
'enabled': True,
'scout': {'enabled': True, 'timeout_seconds': 120},
'researcher': {'enabled': True, 'max_researchers_parallel': 3},
'cache': {'enabled': True, 'ttl_hours': 24}
}
}
# Check if research should run
decision = should_run_research(user_request, config)
if not decision['run']:
print(f"Skipping research: {decision['reason']}")
print("Spawning Planner...")
return spawn_planner(user_request, context={})
# Execute research phase
print("Research phase initiated...")
# Scout
if config['research']['scout']['enabled'] and has_existing_code():
print("└─ Spawning Scout for reconnaissance...")
scout_result = spawn_scout(config)
print(" └─ Scout report complete")
else:
print("└─ No existing codebase detected")
scout_result = None
# Researchers
needs = extract_research_needs(user_request, scout_result)
print(f"└─ {len(needs)} research topics identified")
cached, needed = check_research_cache(needs, config)
print(f" ├─ Cache hits: {len(cached)}")
print(f" └─ Fresh research needed: {len(needed)}")
if needed:
print(f"└─ Spawning {len(needed)} Researcher(s) in parallel...")
fresh_research = spawn_researchers(needed, config)
print(" └─ All research complete")
else:
fresh_research = []
# Assemble context
context = {
'scout': scout_result,
'cached_research': cached,
'fresh_research': fresh_research
}
print("Research phase complete.")
print("Spawning Planner with research context...")
# Spawn Planner with context
return spawn_planner(user_request, context=context)
```
---
## 7. Utility Functions
### Technology Extraction
```python
def extract_technologies(text: str) -> list[str]:
"""Extract technology mentions from text."""
# Pattern: common framework/library names
tech_patterns = [
r'\b(Next\.js|React|Vue|Angular|Svelte)\b',
r'\b(Express|FastAPI|Django|Flask|Rails)\b',
r'\b(Prisma|TypeORM|Sequelize|Mongoose|SQLAlchemy)\b',
r'\b(PostgreSQL|MySQL|MongoDB|Redis|SQLite)\b',
r'\b(Tailwind|Bootstrap|Material UI|Chakra UI)\b',
]
techs = set()
for pattern in tech_patterns:
matches = re.findall(pattern, text, re.IGNORECASE)
techs.update(matches)
return list(techs)
```
### Quick Mode Eligibility
```python
def quick_mode_eligible(user_request: str) -> bool:
"""Check if request is simple enough for quick mode."""
# Simple signals
simple_patterns = [
r'\bfix\b.*\bbug\b',
r'\bupdate\b.*\bversion\b',
r'\badd\b.*\btype\b',
r'\brefactor\b',
r'\brename\b',
]
for pattern in simple_patterns:
if re.search(pattern, user_request, re.IGNORECASE):
return True
# File count estimate
file_mentions = re.findall(r'(\d+)\s*(files?|components?)', user_request, re.IGNORECASE)
if file_mentions:
file_count = int(file_mentions[0][0])
if file_count <= 5:
return True
return False
```
### Cache Helpers
```python
def slugify(text: str) -> str:
"""Convert text to filesystem-safe slug."""
text = text.lower()
text = re.sub(r'[^a-z0-9]+', '-', text)
return text.strip('-')
def jaccard_similarity(list1: list, list2: list) -> float:
"""Compute Jaccard similarity between two lists."""
set1 = set(list1)
set2 = set(list2)
intersection = len(set1 & set2)
union = len(set1 | set2)
return intersection / union if union > 0 else 0.0
```
---
## Summary
MC integrates research-first by:
1. **Checking skip conditions** before every planning request
2. **Spawning Scout** (if existing codebase) for constraints
3. **Extracting research needs** from user request + Scout findings
4. **Checking cache** for existing research (24h TTL)
5. **Spawning Researchers** (parallel) for uncached needs
6. **Assembling context** from Scout + cached + fresh research
7. **Spawning Planner** with complete research context
Research context enables Planner to:
- Respect existing codebase patterns
- Use current best practices (2024-2025)
- Avoid known anti-patterns
- Include working code examples
**End of Line.**