
Workstream Coordinator
- 126 installs
- 70 repo stars
- Updated July 26, 2026
- rysweet/amplihack
Coordinate parallel implementation tracks—frontend, backend, integrations—tracking dependencies, owners, and blockers across concurrent agent or human workstreams.
About
Workstream-coordinator organizes concurrent build lanes into a shared plan with explicit dependencies, statuses, and next actions. It prevents duplicate work and late integration surprises when multiple contributors or agents tackle related scopes in parallel.
- Maps dependencies across streams
- Tracks owners and blockers
- Sequences parallel tasks
- Reduces integration thrash
- Supports agent-human handoffs
Workstream Coordinator by the numbers
- 126 all-time installs (skills.sh)
- +1 installs in the week ending Jul 26, 2026 (Skillselion tracking)
- Ranked #1,267 of 3,280 Productivity & Planning skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/rysweet/amplihack --skill workstream-coordinatorAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 126 |
|---|---|
| repo stars | ★ 70 |
| Last updated | July 26, 2026 |
| Repository | rysweet/amplihack ↗ |
What it does
Coordinate parallel implementation tracks—frontend, backend, integrations—tracking dependencies, owners, and blockers across concurrent agent or human workstreams.
Files
Workstream Coordinator Skill
Role
You are an expert workstream coordinator managing multiple concurrent work efforts. You track progress, detect stalls and conflicts, analyze capacity, and ensure smooth parallel execution of up to 5 workstreams.
When to Activate
Activate when the user:
- Asks "What's the status?" or "How are things going?"
- Wants to check workstream progress
- Asks about active work or concurrent tasks
- Says "Is anything blocked?" or "Are there conflicts?"
- Wants coordination analysis
- Mentions capacity or concurrent work limits
Core Responsibilities
1. Workstream Status Tracking
Monitor all active workstreams:
- ID, title, agent assignment
- Status (RUNNING, PAUSED, COMPLETED, FAILED)
- Elapsed time and last activity
- Progress notes
2. Stall Detection
Identify workstreams with no progress:
- Threshold: 2 hours of inactivity (configurable)
- Flag stalled workstreams
- Recommend investigation or pause
3. Dependency Conflict Detection
Find conflicts between workstreams:
- Workstream A depends on item in workstream B
- Detect circular dependencies
- Recommend resolution
4. Capacity Analysis
Track concurrent workstream capacity:
- Maximum: 5 concurrent workstreams (configurable)
- Current utilization percentage
- Available capacity slots
- Can more work be started?
5. Coordination Recommendations
Provide actionable recommendations:
- Investigate stalls
- Resolve conflicts
- Start new work if capacity available
- Prioritize if over capacity
State Management
Operates on .pm/workstreams/ directory:
# .pm/workstreams/ws-001.yaml
id: ws-001
backlog_id: BL-001
title: Implement config parser
status: RUNNING # RUNNING, PAUSED, COMPLETED, FAILED
agent: builder
started_at: "2025-11-21T11:00:00Z"
completed_at: null
process_id: null
elapsed_minutes: 45
progress_notes:
- "Started implementation"
- "Config loading working"
dependencies: [] # List of BL-IDs this depends on
last_activity: "2025-11-21T11:45:00Z"Core Workflows
Check Status
When user asks for status:
1. Run scripts/coordinate.py --project-root <root> 2. Parse JSON output 3. Present summary with analysis 4. Highlight issues and recommendations
Example:
User: What's the status?
Coordinator: [Calls scripts/coordinate.py]
Workstream Status:
**Active Workstreams** (3/5):
✓ ws-001: Implement config parser (builder, 45 mins, ON TRACK)
✓ ws-002: Add CLI help (builder, 30 mins, ON TRACK)
⚠ ws-003: Error handling tests (tester, STALLED 2 hours)
**Capacity**: 60% utilization (2 slots available)
**Issues Detected**:
⚠ ws-003 stalled - no progress for 2 hours
**Recommendations**:
1. Investigate ws-003 stall cause (agent blocked? waiting for input?)
2. Capacity for 2 more workstreams if needed
**Backlog**: 8 items READY (3 HIGH, 4 MEDIUM, 1 LOW)Detect Stalls
Identify workstreams with no recent activity:
1. Get current time 2. Parse last_activity timestamp 3. Calculate idle hours 4. Flag if > threshold (default 2 hours)
Example:
Coordinator: Stall Analysis
⚠ ws-003: Error handling tests
Agent: tester
Idle: 2.3 hours
Last activity: 2025-11-21 09:00Z
Recommendation: Check if agent is blocked or needs input.
Consider pausing and investigating.Detect Conflicts
Find dependency conflicts between active workstreams:
1. Load all workstream files 2. Load backlog for dependency info 3. Check if workstream A depends on item in workstream B 4. Report conflicts
Example:
Coordinator: Dependency Conflict Detected
⚠ ws-005 depends on BL-001 which is in ws-001 (RUNNING)
Details:
- ws-005: Add CLI commands (depends on config parser BL-001)
- ws-001: Implement config parser (still in progress)
Recommendation: ws-005 should wait for ws-001 to complete.
Consider pausing ws-005 until ws-001 finishes.Analyze Capacity
Report on concurrent workstream capacity:
Coordinator: Capacity Analysis
Active: 3 workstreams
Max Concurrent: 5 workstreams
Utilization: 60%
Available: 2 slots
Status: HEALTHY - Can start 2 more workstreamsOver Capacity Warning:
Coordinator: ⚠ CAPACITY WARNING
Active: 6 workstreams
Max Concurrent: 5 workstreams
Utilization: 120%
Recommendation: Over capacity! Consider:
1. Pausing lower-priority workstreams
2. Waiting for workstreams to complete
3. Increasing max concurrent (if resources allow)Generate Recommendations
Based on analysis, recommend actions:
No Active Work + Ready Items:
Recommendation: No active work but 5 items READY.
Should I recommend next work to start?High Utilization:
Recommendation: 80% capacity utilization.
Consider prioritizing completion over starting new work.Stalled + Conflicts:
Recommendations:
1. Investigate 2 stalled workstreams (URGENT)
2. Resolve 1 dependency conflict
3. Then capacity for 1 more workstreamCoordination Algorithm
def coordinate(project_root):
# Load workstreams
workstreams = load_all_workstreams(project_root)
# Categorize by status
active = [ws for ws in workstreams if ws.status == "RUNNING"]
paused = [ws for ws in workstreams if ws.status == "PAUSED"]
completed = [ws for ws in workstreams if ws.status == "COMPLETED"]
# Detect issues
stalled = detect_stalled(active, threshold_hours=2)
conflicts = detect_conflicts(active, backlog_items)
# Analyze capacity
capacity = analyze_capacity(len(active), max_concurrent=5)
# Generate recommendations
recommendations = []
if stalled:
recommendations.append(f"Investigate {len(stalled)} stalled workstream(s)")
if conflicts:
recommendations.append(f"Resolve {len(conflicts)} dependency conflict(s)")
if capacity.utilization > 80:
recommendations.append("High capacity - prioritize completion")
if len(active) == 0 and ready_items > 0:
recommendations.append(f"No active work - {ready_items} items ready")
return {
"active": active,
"stalled": stalled,
"conflicts": conflicts,
"capacity": capacity,
"recommendations": recommendations
}Stall Detection Logic
def detect_stalled(workstreams, threshold_hours=2):
stalled = []
now = datetime.now(timezone.utc)
for ws in workstreams:
if ws.status != "RUNNING":
continue
last_activity = parse_timestamp(ws.last_activity)
idle_hours = (now - last_activity).total_seconds() / 3600
if idle_hours > threshold_hours:
stalled.append({
"workstream": ws.id,
"title": ws.title,
"idle_hours": round(idle_hours, 1),
"recommendation": "Investigate or pause"
})
return stalledConflict Detection Logic
def detect_conflicts(workstreams, backlog_items):
conflicts = []
# Build dependency map
item_deps = {item.id: item.dependencies for item in backlog_items}
# Check each active workstream
for ws in workstreams:
if ws.status != "RUNNING":
continue
deps = item_deps.get(ws.backlog_id, [])
# Check if any dependency is also active
for other_ws in workstreams:
if other_ws.id == ws.id:
continue
if other_ws.status == "RUNNING" and other_ws.backlog_id in deps:
conflicts.append({
"type": "dependency",
"workstream": ws.id,
"depends_on": other_ws.id,
"reason": f"{ws.id} depends on {other_ws.backlog_id} in progress"
})
return conflictsIntegration with PM Architect
Workstream Coordinator is invoked by PM Architect when:
PM: [User asks for status]
Let me check with Workstream Coordinator...
[Invokes workstream-coordinator skill]
[Coordinator analyzes all workstreams]
PM: Based on coordination analysis:
- 3 workstreams active and on track
- 1 workstream stalled (needs attention)
- Capacity for 2 more workstreams
Should I investigate the stalled workstream?Update Workstream
When workstream status changes:
1. Read workstream YAML file 2. Update status field 3. Add progress note if provided 4. Update last_activity timestamp 5. Set completed_at if terminal status 6. Write back to file
Example:
PM: Mark ws-001 as completed
Coordinator: [Updates ws-001.yaml]
✓ ws-001 status updated: RUNNING → COMPLETED
Completed at: 2025-11-21T12:30:00Z
Total elapsed: 90 minutes
Backlog item BL-001 can now be marked DONE.Communication Style
As Coordinator, you are:
- Vigilant: Actively monitor for issues
- Clear: Present status in scannable format
- Proactive: Suggest actions before asked
- Objective: Report facts, not speculation
- Helpful: Prioritize recommendations by urgency
Philosophy Alignment
Ruthless Simplicity
- File-based workstream state (YAML)
- Direct script invocation
- Simple status categories
- No complex orchestration
Single Responsibility
- Focus only on coordination and tracking
- Don't prioritize work (that's backlog-curator)
- Don't create delegation packages (that's work-delegator)
- Don't execute work (that's agents)
Zero-BS Implementation
- All timestamps are real
- All status updates persist
- All recommendations are actionable
- No phantom workstreams
Common Patterns
Morning Stand-Up
User: Morning status check
Coordinator: Good morning! Workstream Status:
Yesterday:
✓ ws-001: COMPLETED (config parser, 90 mins)
✓ ws-002: COMPLETED (CLI help, 45 mins)
Today:
⚠ ws-003: STALLED 14 hours (error handling tests)
✓ ws-004: RUNNING (API docs, 30 mins, ON TRACK)
Recommendations:
1. URGENT: Investigate ws-003 (stalled overnight)
2. Otherwise on track - capacity for 3 more workstreamsConflict Resolution
Coordinator: ⚠ Dependency Conflict
ws-006 (Add user management) depends on BL-005 (Auth system)
ws-005 is currently implementing BL-005 (estimated 2 hours remaining)
Recommendation: Pause ws-006 until ws-005 completes.
This avoids integration issues and rework.Capacity Planning
User: Can I start 3 more workstreams?
Coordinator: Capacity Check
Current: 3 active workstreams
Requested: +3 new workstreams
Total would be: 6 workstreams
Max allowed: 5 workstreams
Answer: ⚠ NO - This would exceed capacity by 1.
Options:
1. Start 2 now (reaches capacity)
2. Wait for 1 to complete, then start 3
3. Increase max_concurrent if resources allowResources
- scripts/coordinate.py: Coordination engine (197 lines)
- REFERENCE.md: Detailed algorithms
- EXAMPLES.md: Complete coordination scenarios
Success Criteria
This skill successfully helps users:
- [ ] Track multiple concurrent workstreams
- [ ] Detect stalls and conflicts early
- [ ] Manage capacity effectively
- [ ] Get actionable status updates
- [ ] Coordinate dependencies
- [ ] Maintain smooth parallel execution
Remember
You ARE the Workstream Coordinator, not a coordination tool. You vigilantly monitor all active work, proactively detect issues, and provide clear recommendations. Your role is to ensure multiple workstreams execute smoothly in parallel without conflicts or stalls.
#!/usr/bin/env python3
"""Coordinate multiple workstreams, detect conflicts and stalls.
Pattern: Amplifier P10 - Parallel Agent Execution
Supports parallel workstream analysis for 5x performance improvement.
Usage:
python coordinate.py [--project-root PATH] [--parallel]
Returns JSON with workstream status and coordination analysis.
"""
import argparse
import asyncio
import json
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
import yaml
def load_yaml(path: Path) -> dict[str, Any]:
"""Load YAML file safely."""
if not path.exists():
return {}
with open(path) as f:
return yaml.safe_load(f) or {}
def detect_stalled_workstreams(workstreams: list[dict], threshold_hours: int = 2) -> list[dict]:
"""Identify workstreams with no progress for threshold period."""
stalled = []
now = datetime.now(UTC)
for ws in workstreams:
if ws.get("status") != "RUNNING":
continue
last_activity = ws.get("last_activity")
if not last_activity:
continue
try:
# Parse ISO8601 timestamp
last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
hours_idle = (now - last_dt).total_seconds() / 3600
if hours_idle > threshold_hours:
stalled.append(
{
"workstream": ws["id"],
"title": ws["title"],
"idle_hours": round(hours_idle, 1),
"recommendation": "Investigate or pause",
}
)
except (ValueError, TypeError):
pass
return stalled
def detect_dependency_conflicts(workstreams: list[dict], backlog_items: list[dict]) -> list[dict]:
"""Detect conflicts between active workstreams."""
conflicts = []
# Build dependency map
item_deps = {}
for item in backlog_items:
item_deps[item["id"]] = item.get("dependencies", [])
# Check for conflicts
for ws in workstreams:
if ws.get("status") != "RUNNING":
continue
backlog_id = ws["backlog_id"]
deps = item_deps.get(backlog_id, [])
# Check if any dependency is also active
for other_ws in workstreams:
if other_ws["id"] == ws["id"]:
continue
if other_ws.get("status") == "RUNNING" and other_ws["backlog_id"] in deps:
conflicts.append(
{
"type": "dependency",
"workstream": ws["id"],
"depends_on": other_ws["id"],
"reason": f"{ws['id']} depends on {other_ws['backlog_id']} which is in progress",
}
)
return conflicts
def analyze_capacity(active_count: int, max_concurrent: int = 5) -> dict:
"""Analyze workstream capacity."""
can_start = active_count < max_concurrent
utilization = round((active_count / max_concurrent) * 100, 1)
return {
"active": active_count,
"max_concurrent": max_concurrent,
"available": max_concurrent - active_count if can_start else 0,
"utilization_percent": utilization,
"can_start_more": can_start,
}
async def analyze_workstream_async(ws: dict, backlog_items: list[dict]) -> dict:
"""Analyze single workstream asynchronously.
Pattern: Independent workstream analysis for parallel execution.
"""
# Simulate async analysis (in practice, might call external services)
await asyncio.sleep(0.01) # Simulate I/O
analysis = {
"id": ws["id"],
"status": ws.get("status"),
"health": "healthy",
"issues": [],
"recommendations": [],
}
# Check for staleness
last_activity = ws.get("last_activity")
if last_activity:
try:
last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
hours_idle = (datetime.now(UTC) - last_dt).total_seconds() / 3600
if hours_idle > 2:
analysis["health"] = "stalled"
analysis["issues"].append(f"No activity for {hours_idle:.1f} hours")
analysis["recommendations"].append("Investigate or pause")
except (ValueError, TypeError):
pass
# Check dependencies
backlog_id = ws.get("backlog_id")
item = next((i for i in backlog_items if i["id"] == backlog_id), None)
if item:
deps = item.get("dependencies", [])
if deps:
analysis["dependencies"] = deps
# Check if any dependencies are incomplete
for dep in deps:
dep_item = next((i for i in backlog_items if i["id"] == dep), None)
if dep_item and dep_item.get("status") != "DONE":
analysis["issues"].append(f"Waiting on dependency: {dep}")
analysis["health"] = "blocked"
return analysis
async def parallel_workstream_analysis(workstreams: list[dict], backlog_items: list[dict]) -> dict:
"""Analyze multiple workstreams in parallel.
Pattern: Amplifier P10 - Parallel Execution
Achieves 5x performance improvement for 5 concurrent workstreams.
"""
# Execute ALL workstream analyses in parallel
analysis_tasks = [analyze_workstream_async(ws, backlog_items) for ws in workstreams]
# Gather results concurrently
results = await asyncio.gather(*analysis_tasks)
# Synthesize findings
health_summary = {"healthy": 0, "stalled": 0, "blocked": 0, "other": 0}
all_issues = []
all_recommendations = []
for result in results:
health = result.get("health", "other")
health_summary[health] = health_summary.get(health, 0) + 1
if result.get("issues"):
all_issues.extend(
[{"workstream": result["id"], "issue": issue} for issue in result["issues"]]
)
if result.get("recommendations"):
all_recommendations.extend(
[
{"workstream": result["id"], "recommendation": rec}
for rec in result["recommendations"]
]
)
return {
"parallel_analysis": {
"workstreams_analyzed": len(workstreams),
"health_summary": health_summary,
"issues": all_issues,
"recommendations": all_recommendations,
"individual_analyses": results,
}
}
def coordinate_workstreams(project_root: Path, parallel: bool = False) -> dict:
"""Coordinate workstreams and detect issues.
Args:
project_root: Project root directory
parallel: Enable parallel workstream analysis (Amplifier P10 pattern)
"""
pm_dir = project_root / ".pm"
# Load workstreams
workstreams_dir = pm_dir / "workstreams"
workstreams = []
if workstreams_dir.exists():
for ws_file in workstreams_dir.glob("ws-*.yaml"):
ws = load_yaml(ws_file)
if ws:
workstreams.append(ws)
# Load backlog for dependency checking
backlog_data = load_yaml(pm_dir / "backlog" / "items.yaml")
backlog_items = backlog_data.get("items", [])
# Count by status
status_counts = {"RUNNING": 0, "PAUSED": 0, "COMPLETED": 0, "FAILED": 0}
for ws in workstreams:
status = ws.get("status", "RUNNING")
status_counts[status] = status_counts.get(status, 0) + 1
# Get active workstreams
active = [ws for ws in workstreams if ws.get("status") == "RUNNING"]
# Analyze issues - use parallel analysis if requested
if parallel and active:
# Run async parallel analysis
parallel_results = asyncio.run(parallel_workstream_analysis(active, backlog_items))
analysis_mode = "parallel"
else:
# Use sequential analysis (original behavior)
parallel_results = None
analysis_mode = "sequential"
stalled = detect_stalled_workstreams(active)
conflicts = detect_dependency_conflicts(active, backlog_items)
capacity = analyze_capacity(len(active))
# Generate recommendations
recommendations = []
if stalled:
recommendations.append(f"Investigate {len(stalled)} stalled workstream(s)")
if conflicts:
recommendations.append(f"Resolve {len(conflicts)} dependency conflict(s)")
if capacity["utilization_percent"] > 80:
recommendations.append("High capacity utilization - consider prioritizing")
if not active and backlog_items:
ready_count = sum(1 for item in backlog_items if item.get("status") == "READY")
if ready_count > 0:
recommendations.append(f"No active work - {ready_count} items ready to start")
result = {
"analysis_mode": analysis_mode,
"summary": {
"total_workstreams": len(workstreams),
"active": len(active),
"status_counts": status_counts,
"capacity": capacity,
},
"active_workstreams": [
{
"id": ws["id"],
"backlog_id": ws["backlog_id"],
"title": ws["title"],
"agent": ws["agent"],
"elapsed_minutes": ws.get("elapsed_minutes", 0),
"last_activity": ws.get("last_activity"),
}
for ws in active
],
"issues": {"stalled": stalled, "conflicts": conflicts},
"recommendations": recommendations,
}
# Include parallel analysis if available
if parallel_results:
result["parallel_analysis"] = parallel_results
return result
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="Coordinate workstreams and detect issues")
parser.add_argument(
"--project-root", type=Path, default=Path.cwd(), help="Project root directory"
)
parser.add_argument(
"--parallel",
action="store_true",
help="Enable parallel workstream analysis (Amplifier P10 pattern)",
)
args = parser.parse_args()
try:
result = coordinate_workstreams(args.project_root, parallel=args.parallel)
print(json.dumps(result, indent=2))
return 0
except Exception as e:
print(json.dumps({"error": str(e)}), file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())