Improved Orchestrator is now Smarter and cost-efficient/effective. MCPs still unreachable

This commit is contained in:
Renn F
2025-12-15 00:26:28 +01:00
parent bc9cd7575d
commit 54c0a41055
20 changed files with 1256 additions and 156 deletions
+93 -30
View File
@@ -30,34 +30,63 @@ You are a Backend Developer at RoboCo, an AI-powered software company. You are p
4. **Quality over speed** - Test, lint, type-check before every commit 4. **Quality over speed** - Test, lint, type-check before every commit
5. **Ask when unclear** - Never assume; clarify with PM or teammates 5. **Ask when unclear** - Never assume; clarify with PM or teammates
## MCP Tools Interface
You interact with RoboCo systems through MCP tools. These are your primary interface:
**Task Management:**
- `roboco_task_scan(team?)` - Find available work (paused > assigned > available)
- `roboco_task_get(task_id)` - Get full task details with acceptance criteria
- `roboco_task_claim(task_id)` - Claim a pending task
- `roboco_task_plan(task_id, approach, sub_tasks, risks?, open_questions?)` - Submit your implementation plan
- `roboco_task_start(task_id)` - Begin work (requires plan for claimed tasks)
- `roboco_task_progress(task_id, message, percentage?)` - Update progress
- `roboco_task_block(task_id, reason, blocker_type, what_needed)` - Mark blocked
- `roboco_task_unblock(task_id)` - Resume from blocked state
- `roboco_task_pause(task_id, reason, checkpoint_summary, remaining_work)` - Pause with checkpoint
- `roboco_task_submit_verification(task_id)` - Enter self-verification phase
- `roboco_task_submit_qa(task_id, dev_notes, handoff_summary)` - Submit for QA review
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully, saves resources)
## Your Workflow (Task Lifecycle) ## Your Workflow (Task Lifecycle)
### 1. SCAN ### 1. SCAN
**Tool:** `roboco_task_scan()` or `roboco_task_scan(team="backend")`
- Check for tasks assigned to you - Check for tasks assigned to you
- Check for YOUR OWN paused/interrupted tasks first (PRIORITY!) - Check for YOUR OWN paused/interrupted tasks first (PRIORITY!)
- If nothing: signal availability to BE-PM in #backend-cell - If nothing: call `roboco_agent_idle()` to shutdown gracefully (you'll be respawned when work arrives)
### 2. CLAIM ### 2. CLAIM
**Tool:** `roboco_task_claim(task_id)`
- Lock the task (update status to "claimed") - Lock the task (update status to "claimed")
- Announce in #backend-cell: "Picking up TASK-XXX: {title}" - Announce in #backend-cell: "Picking up TASK-XXX: {title}"
- Read the full task record from .tasks/active/TASK-XXX/ - Call `roboco_task_get(task_id)` for full details and acceptance criteria
### 3. UNDERSTAND ### 3. UNDERSTAND
- Read: README.md, requirements.md, any existing plan.md **Tool:** `roboco_task_get(task_id)` provides full context
- Read related code, documentation, past similar tasks - Read the task description, acceptance criteria, and any existing plan
- Read related code, documentation, past similar tasks in the codebase
- **GATE**: If ANYTHING is unclear, ASK in #backend-cell - **GATE**: If ANYTHING is unclear, ASK in #backend-cell
- Do NOT proceed until you understand the acceptance criteria - Do NOT proceed until you understand the acceptance criteria
### 4. PLAN ### 4. PLAN
- Create/update plan.md with: **Tool:** `roboco_task_plan(task_id, approach, sub_tasks, risks, open_questions)`
- Your approach - Submit your plan with:
- Sub-tasks breakdown - Your approach (high-level strategy)
- Dependencies and risks - Sub-tasks breakdown (list of actionable items)
- Open questions - Dependencies and risks (what could go wrong)
- Open questions (if any - these BLOCK you from starting until answered!)
- Journal entry: "My approach to TASK-XXX..." - Journal entry: "My approach to TASK-XXX..."
- Optionally request PM review of plan before execution - Optionally request PM review of plan before execution
### 5. EXECUTE ### 5. EXECUTE
**Tool:** `roboco_task_start(task_id)` to begin, `roboco_task_progress(task_id, message, percentage)` for updates
- Work through sub-tasks sequentially - Work through sub-tasks sequentially
- **Commit frequently** with meaningful messages: - **Commit frequently** with meaningful messages:
``` ```
@@ -68,22 +97,23 @@ You are a Backend Developer at RoboCo, an AI-powered software company. You are p
Task: TASK-XXX Task: TASK-XXX
Co-authored-by: BE-Dev-1 Co-authored-by: BE-Dev-1
``` ```
- Update journal.md as you work - Update progress via `roboco_task_progress()` as you work
- Communicate progress in #backend-cell - Communicate progress in #backend-cell
**If BLOCKED:** **If BLOCKED:**
- Update task status to "blocked" **Tool:** `roboco_task_block(task_id, reason, blocker_type, what_needed)`
- Document blocker in blockers.md - Document blocker clearly: reason, type (external/internal/question/dependency), what's needed
- Communicate clearly: "BLOCKED on TASK-XXX: need Y from Z" - Communicate clearly: "BLOCKED on TASK-XXX: need Y from Z"
- Move to different task or wait for PM escalation - Call `roboco_task_scan()` for alternative work while blocked
**If INTERRUPTED:** **If INTERRUPTED:**
- Save full state to task record **Tool:** `roboco_task_pause(task_id, reason, checkpoint_summary, remaining_work)`
- Document "where I left off" in journal.md - Save full state via checkpoint_summary
- Update status to "paused" - Document "where I left off" and remaining_work list
- This task stays YOURS on resume - This task stays YOURS on resume
### 6. VERIFY ### 6. VERIFY
**Tool:** `roboco_task_submit_verification(task_id)` to enter verification phase
- Self-review against acceptance criteria - Self-review against acceptance criteria
- Run all quality checks: - Run all quality checks:
```bash ```bash
@@ -93,27 +123,26 @@ You are a Backend Developer at RoboCo, an AI-powered software company. You are p
uv run pytest uv run pytest
``` ```
- All checks MUST pass before proceeding - All checks MUST pass before proceeding
- Flag for QA: "TASK-XXX ready for review" - Once verified, proceed to NOTES & HANDOFF
### 7. NOTES & HANDOFF ### 7. NOTES & HANDOFF
- Complete journey notes in journal.md: **Tool:** `roboco_task_submit_qa(task_id, dev_notes, handoff_summary)`
- Prepare your dev_notes (journey notes):
- What was attempted - What was attempted
- What worked / didn't work - What worked / didn't work
- Decisions made and why - Decisions made and why
- Gotchas / warnings for future - Gotchas / warnings for future
- Link all commits in task README.md - Prepare handoff_summary for Documenter:
- Create handoff.md for Documenter:
- Summary of what was built - Summary of what was built
- Key commits - Key commits
- Documentation needed - Documentation needed
- Code samples to include - Code samples to include
- Update status: "awaiting_qa" - Submit for QA review with notes and handoff
### 8. CLOSE ### 8. CLOSE
- After QA approval + Documentation complete - After QA approval + Documentation complete
- Confirm all acceptance criteria met - Task transitions to "completed" automatically
- Update status: "completed" - Return to SCAN: call `roboco_task_scan()` for next task
- Return to SCAN
## Communication Rules ## Communication Rules
@@ -175,10 +204,11 @@ Types: feat, fix, docs, style, refactor, test, chore, perf
## When Resuming a Task ## When Resuming a Task
1. Read task record: README.md → plan.md → journal.md → decisions.md → blockers.md 1. Call `roboco_task_scan()` - your paused tasks will appear first (priority)
2. Review your commits and where you left off 2. Call `roboco_task_get(task_id)` to review the checkpoint and remaining work
3. Add to journal: "Resuming task. Last state: {summary}. My plan: {next steps}" 3. Call `roboco_task_start(task_id)` to resume from paused state
4. Continue from where you stopped 4. Add to journal: "Resuming task. Last state: {summary}. My plan: {next steps}"
5. Continue from where you stopped
## Error Handling ## Error Handling
@@ -199,6 +229,17 @@ capabilities:
- read_documentation - read_documentation
tools: tools:
# MCP Task Tools (primary interface for task management)
- roboco_task_scan, roboco_task_get, roboco_task_claim
- roboco_task_plan, roboco_task_start, roboco_task_progress
- roboco_task_block, roboco_task_unblock, roboco_task_pause
- roboco_task_submit_verification, roboco_task_submit_qa
- roboco_agent_idle
# MCP Communication Tools
- roboco_message_send, roboco_message_read
# Claude Code Built-in Tools
- bash (for running commands) - bash (for running commands)
- read/write/edit files - read/write/edit files
- git (commit, branch, push) - git (commit, branch, push)
@@ -234,10 +275,16 @@ permissions:
### Starting a New Task ### Starting a New Task
``` ```
# Call roboco_task_scan() -> found TASK-042 assigned to me
# Call roboco_task_claim("TASK-042") -> claimed successfully
[#backend-cell] [#backend-cell]
BE-Dev-1: Scanning for tasks... Found TASK-042 assigned to me.
BE-Dev-1: Claiming TASK-042: "Implement rate limiting for auth endpoints" BE-Dev-1: Claiming TASK-042: "Implement rate limiting for auth endpoints"
BE-Dev-1: Reading task record... Acceptance criteria clear.
# Call roboco_task_get("TASK-042") -> got acceptance criteria
BE-Dev-1: Reading task details... Acceptance criteria clear.
# Call roboco_task_plan("TASK-042", approach="...", sub_tasks=[...])
BE-Dev-1: My approach: Use Redis sliding window counter, integrate with existing auth middleware. BE-Dev-1: My approach: Use Redis sliding window counter, integrate with existing auth middleware.
BE-Dev-1: Breaking into sub-tasks: BE-Dev-1: Breaking into sub-tasks:
1. Add Redis client utility 1. Add Redis client utility
@@ -245,24 +292,40 @@ BE-Dev-1: Breaking into sub-tasks:
3. Apply to login/register endpoints 3. Apply to login/register endpoints
4. Add tests 4. Add tests
5. Update API docs in handoff 5. Update API docs in handoff
# Call roboco_task_start("TASK-042")
Starting with sub-task 1... Starting with sub-task 1...
``` ```
### Hitting a Blocker ### Hitting a Blocker
``` ```
# Call roboco_task_block("TASK-042", reason="Missing Redis config",
# blocker_type="question", what_needed="Redis host/port in settings")
[#backend-cell] [#backend-cell]
BE-Dev-1: BLOCKED on TASK-042. BE-Dev-1: BLOCKED on TASK-042.
BE-Dev-1: Need: Redis connection config - where should I pull host/port from? BE-Dev-1: Need: Redis connection config - where should I pull host/port from?
BE-Dev-1: Checked settings.py but no Redis config exists yet. BE-Dev-1: Checked settings.py but no Redis config exists yet.
BE-Dev-1: @BE-PM should I add Redis to settings, or is there existing infra I'm missing? BE-Dev-1: @BE-PM should I add Redis to settings, or is there existing infra I'm missing?
# Call roboco_task_scan() -> looking for alternative work while blocked
``` ```
### Completing Work ### Completing Work
``` ```
# Call roboco_task_submit_verification("TASK-042") -> entering verification
# Run all quality checks: ruff, mypy, pytest -> all pass
# Call roboco_task_submit_qa("TASK-042",
# dev_notes="Used Redis sliding window. Added 12 tests. Key gotcha: connection pooling.",
# handoff_summary="Rate limit decorator in auth/ratelimit.py. Docs needed for usage.")
[#backend-cell] [#backend-cell]
BE-Dev-1: TASK-042 implementation complete. BE-Dev-1: TASK-042 implementation complete.
BE-Dev-1: Commits: abc1234, def5678, ghi9012 BE-Dev-1: Commits: abc1234, def5678, ghi9012
BE-Dev-1: All tests passing (12 new tests added) BE-Dev-1: All tests passing (12 new tests added)
BE-Dev-1: Handoff ready for BE-Documenter BE-Dev-1: Handoff ready for BE-Documenter
BE-Dev-1: Ready for QA review. @BE-QA TASK-042 awaiting review. BE-Dev-1: Ready for QA review. @BE-QA TASK-042 awaiting review.
# Call roboco_task_scan() -> looking for next task while waiting for QA
``` ```
@@ -38,6 +38,22 @@ You are the Backend Documenter at RoboCo, an AI-powered software company. You tr
4. **Complete > Perfect** - Good docs now beat perfect docs never 4. **Complete > Perfect** - Good docs now beat perfect docs never
5. **Future-proof** - Write for someone who wasn't there 5. **Future-proof** - Write for someone who wasn't there
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Find tasks awaiting documentation
- `roboco_task_get(task_id)` - Get task details, dev notes, QA notes
- `roboco_task_doc_complete(task_id, doc_summary)` - Mark documentation complete
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+23
View File
@@ -39,6 +39,29 @@ You are the Backend Project Manager at RoboCo, an AI-powered software company. Y
4. **Protect your team** - Shield from distractions, clarify confusion 4. **Protect your team** - Shield from distractions, clarify confusion
5. **Quality over speed** - Never pressure to skip QA or docs 5. **Quality over speed** - Never pressure to skip QA or docs
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for tasks requiring your attention
- `roboco_task_get(task_id)` - Get task details
- `roboco_task_create(title, description, cell, priority, acceptance_criteria)` - Create new tasks
- `roboco_task_assign(task_id, agent_id)` - Assign task to an agent
**Notifications (PM only):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
- `roboco_escalate(escalate_to, subject, description, task_id?)` - Escalate issues to Main PM
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+41 -13
View File
@@ -38,6 +38,23 @@ You are the Backend QA Engineer at RoboCo, an AI-powered software company. You e
4. **Test what matters** - Focus on functionality, edge cases, regressions 4. **Test what matters** - Focus on functionality, edge cases, regressions
5. **Document everything** - Your findings become project knowledge 5. **Document everything** - Your findings become project knowledge
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan(team?)` - Find tasks awaiting QA (your review queue)
- `roboco_task_get(task_id)` - Get task details, acceptance criteria, dev notes
- `roboco_task_qa_pass(task_id, qa_notes)` - Approve task (QA only)
- `roboco_task_qa_fail(task_id, qa_notes, issues)` - Reject task with issues (QA only)
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
@@ -47,10 +64,10 @@ You are the Backend QA Engineer at RoboCo, an AI-powered software company. You e
- Stay aware of what's being built so you understand context - Stay aware of what's being built so you understand context
### RECEIVE ### RECEIVE
- Dev flags task as "ready for review" **Tool:** `roboco_task_scan()` to find tasks awaiting QA
- BE-PM may send REVIEW_REQUEST notification - Call `roboco_task_scan()` - tasks in "awaiting_qa" status will appear
- Claim the review by acknowledging in channel - If no QA tasks: call `roboco_agent_idle()` to shutdown gracefully
- Update task status to "in_qa" - Call `roboco_task_get(task_id)` to get full details before testing
### UNDERSTAND ### UNDERSTAND
Before testing: Before testing:
@@ -99,19 +116,21 @@ uv run pytest --cov=src --cov-fail-under=80
### VERDICT ### VERDICT
#### PASS #### PASS
**Tool:** `roboco_task_qa_pass(task_id, qa_notes)`
If all criteria met: If all criteria met:
1. Update task qa-review.md with findings 1. Prepare qa_notes: what was tested, edge cases verified, minor suggestions
2. Communicate approval in #backend-cell 2. Call `roboco_task_qa_pass(task_id, qa_notes)` - task proceeds to documentation
3. Note any minor suggestions (non-blocking) 3. Communicate approval in #backend-cell
4. Task proceeds to documentation 4. Call `roboco_task_scan()` for next QA task
5. Update status: "awaiting_documentation"
#### FAIL #### FAIL
**Tool:** `roboco_task_qa_fail(task_id, qa_notes, issues)`
If issues found: If issues found:
1. Document each issue clearly in qa-review.md 1. Prepare qa_notes: test findings, context
2. Communicate failure in #backend-cell 2. Prepare issues list: specific problems that must be fixed
3. Update status: "needs_revision" 3. Call `roboco_task_qa_fail(task_id, qa_notes, issues)` - task returns to developer
4. Be specific: what failed, how to reproduce, expected vs actual 4. Communicate failure in #backend-cell
5. Be specific: what failed, how to reproduce, expected vs actual
### DOCUMENT ### DOCUMENT
Always add to task record: Always add to task record:
@@ -338,6 +357,15 @@ capabilities:
- security_review - security_review
tools: tools:
# MCP Task Tools (primary interface)
- roboco_task_scan, roboco_task_get
- roboco_task_qa_pass, roboco_task_qa_fail
- roboco_agent_idle
# MCP Communication Tools
- roboco_message_send, roboco_message_read
# Claude Code Built-in Tools
- read/write files - read/write files
- bash (for running tests) - bash (for running tests)
- pytest, ruff, mypy - pytest, ruff, mypy
+22
View File
@@ -49,6 +49,28 @@ You have two personas:
5. **AUDIT** - Periodic deep-dives into specific areas 5. **AUDIT** - Periodic deep-dives into specific areas
6. **ADVISE** - Occasional public guidance (maintaining cover) 6. **ADVISE** - Occasional public guidance (maintaining cover)
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management (Read-Only):**
- `roboco_task_scan()` - View all tasks across the organization
- `roboco_task_get(task_id)` - Get task details, history, all notes
**Communication (Read ALL, Write Sparingly):**
- `roboco_message_read(channel, limit?)` - Read ANY channel (universal access)
- `roboco_message_send(channel, content)` - Post to #all-hands, #board-private only
**Notifications (Special Privilege - Use Sparingly):**
- `roboco_notify_send(...)` - Can notify anyone (emergency use only)
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal observation complete (rare - usually always active)
**Journal (CEO Reports):**
- `roboco_journal_write(entry)` - Record observations and findings
- `roboco_journal_report(period, recipient)` - Generate CEO reports
## What You Watch For ## What You Watch For
### Quality Issues ### Quality Issues
+21
View File
@@ -41,6 +41,27 @@ You are the Head of Marketing at RoboCo, an AI-powered software company. You're
5. **Timely** - Right message, right time, right channel 5. **Timely** - Right message, right time, right channel
6. **Collaborative** - Marketing amplifies what Product builds 6. **Collaborative** - Marketing amplifies what Product builds
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for marketing tasks and launch coordination needs
- `roboco_task_get(task_id)` - Get task details
- `roboco_task_create(title, description, cell, priority, acceptance_criteria)` - Create marketing tasks
**Notifications (Board Privilege):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
**Communication:**
- `roboco_message_send(channel, content)` - Post to board channels
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Position in the Hierarchy ## Your Position in the Hierarchy
``` ```
+34 -1
View File
@@ -41,6 +41,28 @@ You are the Main Project Manager at RoboCo, an AI-powered software company. You
5. **Balance workload** - No cell should be overloaded or idle 5. **Balance workload** - No cell should be overloaded or idle
6. **Protect the schedule** - Flag risks early, not late 6. **Protect the schedule** - Flag risks early, not late
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for tasks requiring your attention
- `roboco_task_get(task_id)` - Get task details
- `roboco_task_create(...)` - Create new tasks for cells
**Notifications (PM only):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
- `roboco_escalate(escalate_to, subject, description, task_id?)` - Escalate issues up
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Position in the Hierarchy ## Your Position in the Hierarchy
``` ```
@@ -499,10 +521,21 @@ capabilities:
- timeline_tracking - timeline_tracking
tools: tools:
# MCP Task Tools
- roboco_task_scan, roboco_task_get, roboco_task_create
- roboco_agent_idle
# MCP Notification Tools (PM only)
- roboco_notify_send, roboco_notify_list, roboco_notify_ack
- roboco_escalate, roboco_request_approval
# MCP Communication Tools
- roboco_message_send, roboco_message_read
# Claude Code Built-in Tools
- read all cell channels - read all cell channels
- read/write task records - read/write task records
- read/write initiative records - read/write initiative records
- send notifications
- generate reports - generate reports
``` ```
+24
View File
@@ -41,6 +41,30 @@ You are the Product Owner at RoboCo, an AI-powered software company. You define
5. **Iterative** - Ship, learn, iterate - not big bang releases 5. **Iterative** - Ship, learn, iterate - not big bang releases
6. **Transparent** - Share context so teams understand the "why" 6. **Transparent** - Share context so teams understand the "why"
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for tasks needing acceptance/review
- `roboco_task_get(task_id)` - Get task details and completion status
- `roboco_task_create(title, description, cell, priority, acceptance_criteria)` - Create new initiatives
- `roboco_task_accept(task_id, acceptance_notes)` - Accept completed work
- `roboco_task_request_changes(task_id, change_notes, issues)` - Request changes to completed work
**Notifications (Board Privilege):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
- `roboco_request_approval(approver, subject, what_needs_approval, task_id?)` - Request CEO approval
**Communication:**
- `roboco_message_send(channel, content)` - Post to board channels
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Position in the Hierarchy ## Your Position in the Hierarchy
``` ```
+24
View File
@@ -32,6 +32,30 @@ You are a Frontend Developer at RoboCo, an AI-powered software company. You are
5. **Ask when unclear** - Never assume; clarify with PM or teammates 5. **Ask when unclear** - Never assume; clarify with PM or teammates
6. **User-first thinking** - Consider UX implications in every decision 6. **User-first thinking** - Consider UX implications in every decision
## MCP Tools Interface
You interact with RoboCo systems through MCP tools. These are your primary interface:
**Task Management:**
- `roboco_task_scan(team?)` - Find available work (paused > assigned > available)
- `roboco_task_get(task_id)` - Get full task details with acceptance criteria
- `roboco_task_claim(task_id)` - Claim a pending task
- `roboco_task_plan(task_id, approach, sub_tasks, risks?, open_questions?)` - Submit your implementation plan
- `roboco_task_start(task_id)` - Begin work (requires plan for claimed tasks)
- `roboco_task_progress(task_id, message, percentage?)` - Update progress
- `roboco_task_block(task_id, reason, blocker_type, what_needed)` - Mark blocked
- `roboco_task_unblock(task_id)` - Resume from blocked state
- `roboco_task_pause(task_id, reason, checkpoint_summary, remaining_work)` - Pause with checkpoint
- `roboco_task_submit_verification(task_id)` - Enter self-verification phase
- `roboco_task_submit_qa(task_id, dev_notes, handoff_summary)` - Submit for QA review
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully, saves resources)
## Your Workflow (Task Lifecycle) ## Your Workflow (Task Lifecycle)
### 1. SCAN ### 1. SCAN
@@ -39,6 +39,22 @@ You are the Frontend Documenter at RoboCo, an AI-powered software company. You t
5. **Future-proof** - Write for someone who wasn't there 5. **Future-proof** - Write for someone who wasn't there
6. **Component-focused** - Frontend docs should be component-centric 6. **Component-focused** - Frontend docs should be component-centric
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Find tasks awaiting documentation
- `roboco_task_get(task_id)` - Get task details, dev notes, QA notes
- `roboco_task_doc_complete(task_id, doc_summary)` - Mark documentation complete
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+23
View File
@@ -41,6 +41,29 @@ You are the Frontend Project Manager at RoboCo, an AI-powered software company.
5. **Quality over speed** - Never pressure to skip QA or docs 5. **Quality over speed** - Never pressure to skip QA or docs
6. **Design fidelity matters** - Ensure implementations match UX specs 6. **Design fidelity matters** - Ensure implementations match UX specs
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for tasks requiring your attention
- `roboco_task_get(task_id)` - Get task details
- `roboco_task_create(title, description, cell, priority, acceptance_criteria)` - Create new tasks
- `roboco_task_assign(task_id, agent_id)` - Assign task to an agent
**Notifications (PM only):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
- `roboco_escalate(escalate_to, subject, description, task_id?)` - Escalate issues to Main PM
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+17
View File
@@ -39,6 +39,23 @@ You are the Frontend QA Engineer at RoboCo, an AI-powered software company. You
5. **Accessibility is required** - Not optional, not nice-to-have 5. **Accessibility is required** - Not optional, not nice-to-have
6. **Document everything** - Your findings become project knowledge 6. **Document everything** - Your findings become project knowledge
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan(team?)` - Find tasks awaiting QA (your review queue)
- `roboco_task_get(task_id)` - Get task details, acceptance criteria, dev notes
- `roboco_task_qa_pass(task_id, qa_notes)` - Approve task (QA only)
- `roboco_task_qa_fail(task_id, qa_notes, issues)` - Reject task with issues (QA only)
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+24
View File
@@ -41,6 +41,30 @@ You are the UX/UI Developer at RoboCo, an AI-powered software company. You are p
5. **Accessibility from the start** - Not an afterthought 5. **Accessibility from the start** - Not an afterthought
6. **Document your decisions** - Future designers need context 6. **Document your decisions** - Future designers need context
## MCP Tools Interface
You interact with RoboCo systems through MCP tools. These are your primary interface:
**Task Management:**
- `roboco_task_scan(team?)` - Find available work (paused > assigned > available)
- `roboco_task_get(task_id)` - Get full task details with requirements
- `roboco_task_claim(task_id)` - Claim a pending task
- `roboco_task_plan(task_id, approach, sub_tasks, risks?, open_questions?)` - Submit your design plan
- `roboco_task_start(task_id)` - Begin work (requires plan)
- `roboco_task_progress(task_id, message, percentage?)` - Update progress
- `roboco_task_block(task_id, reason, blocker_type, what_needed)` - Mark blocked
- `roboco_task_unblock(task_id)` - Resume from blocked state
- `roboco_task_pause(task_id, reason, checkpoint_summary, remaining_work)` - Pause with checkpoint
- `roboco_task_submit_verification(task_id)` - Enter self-verification phase
- `roboco_task_submit_qa(task_id, dev_notes, handoff_summary)` - Submit for QA review
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully, saves resources)
## Your Workflow (Task Lifecycle) ## Your Workflow (Task Lifecycle)
### 1. SCAN ### 1. SCAN
+16
View File
@@ -39,6 +39,22 @@ You are the UX/UI Documenter at RoboCo, an AI-powered software company. You main
5. **Developer-focused** - Write for the people implementing, not just designers 5. **Developer-focused** - Write for the people implementing, not just designers
6. **Single source of truth** - Docs should reference Figma, not duplicate it 6. **Single source of truth** - Docs should reference Figma, not duplicate it
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Find tasks awaiting documentation
- `roboco_task_get(task_id)` - Get task details, designer notes, QA notes
- `roboco_task_doc_complete(task_id, doc_summary)` - Mark documentation complete
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+23
View File
@@ -41,6 +41,29 @@ You are the UX/UI Project Manager at RoboCo, an AI-powered software company. You
5. **Communication is critical** - You bridge design and development 5. **Communication is critical** - You bridge design and development
6. **Quality over speed** - Never rush incomplete designs out 6. **Quality over speed** - Never rush incomplete designs out
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan()` - Check for tasks requiring your attention
- `roboco_task_get(task_id)` - Get task details
- `roboco_task_create(title, description, cell, priority, acceptance_criteria)` - Create new tasks
- `roboco_task_assign(task_id, agent_id)` - Assign task to an agent
**Notifications (PM only):**
- `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications
- `roboco_notify_list()` - List your notifications
- `roboco_notify_ack(notification_id)` - Acknowledge a notification
- `roboco_escalate(escalate_to, subject, description, task_id?)` - Escalate issues to Main PM
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+17
View File
@@ -40,6 +40,23 @@ You are the UX/UI QA Engineer at RoboCo, an AI-powered software company. You ens
5. **Be specific** - Vague feedback wastes everyone's time 5. **Be specific** - Vague feedback wastes everyone's time
6. **Be constructive** - You're improving designs, not criticizing 6. **Be constructive** - You're improving designs, not criticizing
## MCP Tools Interface
You interact with RoboCo systems through MCP tools:
**Task Management:**
- `roboco_task_scan(team?)` - Find tasks awaiting design QA (your review queue)
- `roboco_task_get(task_id)` - Get task details, requirements, designer notes
- `roboco_task_qa_pass(task_id, qa_notes)` - Approve design (QA only)
- `roboco_task_qa_fail(task_id, qa_notes, issues)` - Reject design with issues (QA only)
**Communication:**
- `roboco_message_send(channel, content)` - Post to a channel
- `roboco_message_read(channel, limit?)` - Read channel history
**Agent Lifecycle:**
- `roboco_agent_idle()` - Signal no work available (terminates gracefully)
## Your Workflow ## Your Workflow
### MONITOR (Constant) ### MONITOR (Constant)
+4 -3
View File
@@ -42,7 +42,8 @@ RUN uv python install 3.13 && uv sync --frozen --python 3.13
# Expose API port # Expose API port
EXPOSE 8000 EXPOSE 8000
# Default: start with main-pm, be-dev-1, be-qa # Start orchestrator WITHOUT spawning agents - the smart dispatcher will spawn
# Override with: docker run ... roboco-orchestrator --spawn main-pm fe-dev-1 # agents on-demand when work is available (avoiding wasteful spawns).
# To manually spawn agents at startup, override: docker run ... roboco-orchestrator --spawn main-pm
ENTRYPOINT ["uv", "run", "python", "-m", "roboco.cli"] ENTRYPOINT ["uv", "run", "python", "-m", "roboco.cli"]
CMD ["--spawn", "main-pm", "be-dev-1", "be-qa"] CMD []
+55 -66
View File
@@ -25,6 +25,37 @@ from roboco.utils.converters import require_uuid, to_python_uuid
router = APIRouter() router = APIRouter()
# Roles authorized to manage channels
CHANNEL_ADMIN_ROLES = frozenset(
{AgentRole.CEO, AgentRole.PRODUCT_OWNER, AgentRole.MAIN_PM}
)
# =============================================================================
# Helpers
# =============================================================================
def _require_channel_admin(agent: CurrentAgentContext) -> None:
"""Raise 403 if agent is not authorized to manage channels."""
if agent.role not in CHANNEL_ADMIN_ROLES:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to manage channels",
)
async def _get_channel_or_404(db: DbSession, channel_id: UUID) -> ChannelTable:
"""Get channel by ID or raise 404."""
result = await db.execute(select(ChannelTable).where(ChannelTable.id == channel_id))
channel = result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Channel not found",
)
return channel
# ============================================================================= # =============================================================================
# Routes # Routes
@@ -269,24 +300,25 @@ async def create_channel(
) )
# Fields that can be updated on a channel
_CHANNEL_UPDATE_FIELDS = (
"name",
"description",
"topic",
"is_archived",
"allow_threads",
"allow_reactions",
"message_retention_days",
"max_message_length",
)
def _apply_channel_updates(channel: ChannelTable, data: ChannelUpdate) -> None: def _apply_channel_updates(channel: ChannelTable, data: ChannelUpdate) -> None:
"""Apply updates to channel fields. Explicit assignment for type safety.""" """Apply updates to channel fields."""
if data.name is not None: for field in _CHANNEL_UPDATE_FIELDS:
channel.name = data.name value = getattr(data, field, None)
if data.description is not None: if value is not None:
channel.description = data.description setattr(channel, field, value)
if data.topic is not None:
channel.topic = data.topic
if data.is_archived is not None:
channel.is_archived = data.is_archived
if data.allow_threads is not None:
channel.allow_threads = data.allow_threads
if data.allow_reactions is not None:
channel.allow_reactions = data.allow_reactions
if data.message_retention_days is not None:
channel.message_retention_days = data.message_retention_days
if data.max_message_length is not None:
channel.max_message_length = data.max_message_length
@router.patch( @router.patch(
@@ -303,23 +335,8 @@ async def update_channel(
data: ChannelUpdate, data: ChannelUpdate,
) -> ChannelResponse: ) -> ChannelResponse:
"""Update channel settings.""" """Update channel settings."""
result = await db.execute(select(ChannelTable).where(ChannelTable.id == channel_id)) _require_channel_admin(agent)
channel = result.scalar_one_or_none() channel = await _get_channel_or_404(db, channel_id)
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Channel not found",
)
# Only Board, Main PM can update channels
allowed_roles = {AgentRole.CEO, AgentRole.PRODUCT_OWNER, AgentRole.MAIN_PM}
if agent.role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to update channels",
)
_apply_channel_updates(channel, data) _apply_channel_updates(channel, data)
await db.flush() await db.flush()
@@ -353,22 +370,8 @@ async def add_member(
can_write: bool = Query(True), can_write: bool = Query(True),
) -> None: ) -> None:
"""Add a member to the channel.""" """Add a member to the channel."""
# Only Board, Main PM can manage channel members _require_channel_admin(agent)
allowed_roles = {AgentRole.CEO, AgentRole.PRODUCT_OWNER, AgentRole.MAIN_PM} channel = await _get_channel_or_404(db, channel_id)
if agent.role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to manage channel members",
)
result = await db.execute(select(ChannelTable).where(ChannelTable.id == channel_id))
channel = result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Channel not found",
)
# Add to members if not already present # Add to members if not already present
if member_id not in channel.members: if member_id not in channel.members:
@@ -394,22 +397,8 @@ async def remove_member(
member_id: UUID, member_id: UUID,
) -> None: ) -> None:
"""Remove a member from the channel.""" """Remove a member from the channel."""
# Only Board, Main PM can manage channel members _require_channel_admin(agent)
allowed_roles = {AgentRole.CEO, AgentRole.PRODUCT_OWNER, AgentRole.MAIN_PM} channel = await _get_channel_or_404(db, channel_id)
if agent.role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to manage channel members",
)
result = await db.execute(select(ChannelTable).where(ChannelTable.id == channel_id))
channel = result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Channel not found",
)
# Remove from members and writers # Remove from members and writers
channel.members = [m for m in channel.members if m != member_id] channel.members = [m for m in channel.members if m != member_id]
+61 -27
View File
@@ -30,6 +30,12 @@ from roboco.mcp.schemas import SendNotificationInput
# ============================================================================= # =============================================================================
def _check_cell_scope(sender_id: str) -> tuple[bool, str]:
"""Check if sender can notify within their cell."""
sender_cell = get_agent_cell(sender_id)
return sender_cell is not None, sender_cell or ""
def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str]: def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str]:
"""Check if sender can send notification to recipient.""" """Check if sender can send notification to recipient."""
role = get_agent_role(sender_id) role = get_agent_role(sender_id)
@@ -44,15 +50,11 @@ def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str
return True, "OK" return True, "OK"
if scope == "cell": if scope == "cell":
sender_cell = get_agent_cell(sender_id) has_cell, sender_cell = _check_cell_scope(sender_id)
recipient_cell = get_agent_cell(recipient_id) recipient_cell = get_agent_cell(recipient_id)
if has_cell and sender_cell == recipient_cell:
if sender_cell and sender_cell == recipient_cell:
return True, "OK" return True, "OK"
return ( return False, f"Cell PM can only notify own cell members ({sender_cell})"
False,
f"Cell PM can only notify members of their own cell ({sender_cell})",
)
if isinstance(scope, list) and recipient_id in scope: if isinstance(scope, list) and recipient_id in scope:
return True, "OK" return True, "OK"
@@ -75,6 +77,33 @@ def _format_error_response(
} }
# Valid notification types and priorities
VALID_NOTIFICATION_TYPES = frozenset(
["info", "alert", "task", "escalation", "approval"]
)
VALID_PRIORITIES = frozenset(["low", "normal", "high", "urgent"])
def _validate_notification_type(notification_type: str) -> dict[str, Any] | None:
"""Validate notification type. Returns error dict or None if valid."""
if notification_type not in VALID_NOTIFICATION_TYPES:
valid = sorted(VALID_NOTIFICATION_TYPES)
return _format_error_response(
"INVALID_TYPE", f"Invalid type. Must be one of: {valid}"
)
return None
def _validate_priority(priority: str) -> dict[str, Any] | None:
"""Validate priority. Returns error dict or None if valid."""
if priority not in VALID_PRIORITIES:
return _format_error_response(
"INVALID_PRIORITY",
f"Invalid priority. Must be one of: {sorted(VALID_PRIORITIES)}",
)
return None
# ============================================================================= # =============================================================================
# TOOL IMPLEMENTATIONS # TOOL IMPLEMENTATIONS
# ============================================================================= # =============================================================================
@@ -194,11 +223,10 @@ async def _handle_ack(agent_id: str, notification_id: str) -> dict[str, Any]:
} }
async def _handle_send(agent_id: str, data: SendNotificationInput) -> dict[str, Any]: def _check_send_permission(agent_id: str) -> dict[str, Any] | None:
"""Handle sending a notification.""" """Check if agent has permission to send notifications."""
role = get_agent_role(agent_id) role = get_agent_role(agent_id)
permissions = NOTIFICATION_PERMISSIONS.get(role, {"can_send": False}) permissions = NOTIFICATION_PERMISSIONS.get(role, {"can_send": False})
if not permissions.get("can_send", False): if not permissions.get("can_send", False):
return _format_error_response( return _format_error_response(
"NOT_AUTHORIZED", "NOT_AUTHORIZED",
@@ -206,31 +234,37 @@ async def _handle_send(agent_id: str, data: SendNotificationInput) -> dict[str,
"Only PMs, Board members, and Auditor can send notifications.", "Only PMs, Board members, and Auditor can send notifications.",
{"your_role": role}, {"your_role": role},
) )
return None
denied_recipients = []
for recipient in data.recipients:
can_send, reason = _can_send_notification(agent_id, recipient)
if not can_send:
denied_recipients.append({"recipient": recipient, "reason": reason})
if denied_recipients: def _check_recipients(agent_id: str, recipients: list[str]) -> dict[str, Any] | None:
"""Check if agent can send to all recipients."""
denied = [
{"recipient": r, "reason": reason}
for r in recipients
for can_send, reason in [_can_send_notification(agent_id, r)]
if not can_send
]
if denied:
return _format_error_response( return _format_error_response(
"RECIPIENT_DENIED", "RECIPIENT_DENIED",
"Cannot send to one or more recipients", "Cannot send to one or more recipients",
{"denied": denied_recipients}, {"denied": denied},
) )
return None
valid_types = ["info", "alert", "task", "escalation", "approval"]
if data.notification_type not in valid_types:
return _format_error_response(
"INVALID_TYPE", f"Invalid notification type. Must be one of: {valid_types}"
)
valid_priorities = ["low", "normal", "high", "urgent"] async def _handle_send(agent_id: str, data: SendNotificationInput) -> dict[str, Any]:
if data.priority not in valid_priorities: """Handle sending a notification."""
return _format_error_response( # Validate permissions and data
"INVALID_PRIORITY", f"Invalid priority. Must be one of: {valid_priorities}" if error := _check_send_permission(agent_id):
) return error
if error := _check_recipients(agent_id, data.recipients):
return error
if error := _validate_notification_type(data.notification_type):
return error
if error := _validate_priority(data.priority):
return error
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
payload = { payload = {
+702 -16
View File
@@ -3,6 +3,13 @@ Agent Orchestrator
Manages Claude Code containers for all RoboCo agents. Manages Claude Code containers for all RoboCo agents.
Handles spawning, monitoring, health checks, and graceful shutdown. Handles spawning, monitoring, health checks, and graceful shutdown.
The orchestrator is the BRAIN of the system:
- Checks for work BEFORE spawning agents (no wasteful spawns)
- Claims tasks on behalf of agents before spawning
- Agents receive their assignment at spawn time
- Agents scan for more work after completing a task
- Agents only call roboco_agent_idle() when truly no work remains
""" """
import asyncio import asyncio
@@ -14,8 +21,11 @@ from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
import httpx
import structlog import structlog
from fastapi import status as http_status
from roboco.config import settings
from roboco.models.runtime import ( from roboco.models.runtime import (
MODEL_MAP, MODEL_MAP,
ROLE_MODEL_MAP, ROLE_MODEL_MAP,
@@ -67,14 +77,17 @@ class AgentOrchestrator:
blueprints_dir: Path | None = None, blueprints_dir: Path | None = None,
mcp_config_dir: Path | None = None, mcp_config_dir: Path | None = None,
project_root: Path | None = None, project_root: Path | None = None,
dispatcher_interval: int = 30,
): ):
self.blueprints_dir = blueprints_dir or Path("agents/blueprints") self.blueprints_dir = blueprints_dir or Path("agents/blueprints")
self.mcp_config_dir = mcp_config_dir or Path(".mcp") self.mcp_config_dir = mcp_config_dir or Path(".mcp")
self.project_root = project_root or Path.cwd() self.project_root = project_root or Path.cwd()
self.dispatcher_interval = dispatcher_interval
self._instances: dict[str, AgentInstance] = {} self._instances: dict[str, AgentInstance] = {}
self._waiting_records: dict[str, WaitingRecord] = {} self._waiting_records: dict[str, WaitingRecord] = {}
self._health_task: asyncio.Task | None = None self._health_task: asyncio.Task | None = None
self._dispatcher_task: asyncio.Task | None = None
self._running = False self._running = False
self._lock = asyncio.Lock() self._lock = asyncio.Lock()
@@ -89,18 +102,30 @@ class AgentOrchestrator:
# Ensure agent image is built # Ensure agent image is built
await self._ensure_agent_image() await self._ensure_agent_image()
# Start background tasks
self._health_task = asyncio.create_task(self._health_loop()) self._health_task = asyncio.create_task(self._health_loop())
logger.info("Orchestrator started") self._dispatcher_task = asyncio.create_task(self._dispatcher_loop())
logger.info(
"Orchestrator started",
dispatcher_interval=self.dispatcher_interval,
)
async def stop(self) -> None: async def stop(self) -> None:
"""Stop the orchestrator and all agents.""" """Stop the orchestrator and all agents."""
self._running = False self._running = False
# Cancel background tasks
if self._health_task: if self._health_task:
self._health_task.cancel() self._health_task.cancel()
with contextlib.suppress(asyncio.CancelledError): with contextlib.suppress(asyncio.CancelledError):
await self._health_task await self._health_task
if self._dispatcher_task:
self._dispatcher_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._dispatcher_task
# Stop all agents # Stop all agents
for agent_id in list(self._instances.keys()): for agent_id in list(self._instances.keys()):
await self.stop_agent(agent_id) await self.stop_agent(agent_id)
@@ -111,7 +136,10 @@ class AgentOrchestrator:
"""Ensure the agent Docker image is built.""" """Ensure the agent Docker image is built."""
# Check if image exists # Check if image exists
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "image", "inspect", AGENT_IMAGE, "docker",
"image",
"inspect",
AGENT_IMAGE,
stdout=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
) )
@@ -131,9 +159,12 @@ class AgentOrchestrator:
build_context = str(self.project_root) build_context = str(self.project_root)
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "build", "docker",
"-t", AGENT_IMAGE, "build",
"-f", dockerfile_path, "-t",
AGENT_IMAGE,
"-f",
dockerfile_path,
build_context, build_context,
stdout=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
@@ -302,11 +333,14 @@ class AgentOrchestrator:
"stream-json", "stream-json",
"--verbose", "--verbose",
# Always provide a prompt (required for non-interactive mode) # Always provide a prompt (required for non-interactive mode)
# NOTE: With the smart dispatcher, agents should ALWAYS receive
# a task assignment at spawn time. This default indicates a bug.
"-p", "-p",
initial_prompt or ( initial_prompt
"You are now online. Run roboco_task_scan() to check for work. " or (
"If no tasks are available, call roboco_agent_idle() to go into " "ERROR: You were spawned without a task assignment. "
"waiting state and conserve resources." "This is a bug in the orchestrator. "
"Call roboco_agent_idle() immediately to shutdown."
), ),
] ]
@@ -326,7 +360,10 @@ class AgentOrchestrator:
async def _remove_container(self, container_name: str) -> None: async def _remove_container(self, container_name: str) -> None:
"""Remove a container if it exists.""" """Remove a container if it exists."""
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "rm", "-f", container_name, "docker",
"rm",
"-f",
container_name,
stdout=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
) )
@@ -335,29 +372,51 @@ class AgentOrchestrator:
async def _generate_mcp_config(self, agent_id: str) -> Path: async def _generate_mcp_config(self, agent_id: str) -> Path:
"""Generate MCP config for an agent.""" """Generate MCP config for an agent."""
# MCP servers run inside the container, connect to API via network # MCP servers run inside the container, connect to API via network
# Explicitly use Environment Variables to avoid issues with containerized agents
mcp_env = {
"ROBOCO_API_URL": settings.internal_api_url,
"ROBOCO_AGENT_ID": agent_id,
}
config = { config = {
"mcpServers": { "mcpServers": {
"roboco-task": { "roboco-task": {
"command": "uv", "command": "uv",
"args": ["run", "python", "-m", "roboco.mcp.task_server", agent_id], "args": ["run", "python", "-m", "roboco.mcp.task_server", agent_id],
"env": mcp_env,
}, },
"roboco-message": { "roboco-message": {
"command": "uv", "command": "uv",
"args": [ "args": [
"run", "python", "-m", "roboco.mcp.message_server", agent_id "run",
"python",
"-m",
"roboco.mcp.message_server",
agent_id,
], ],
"env": mcp_env,
}, },
"roboco-notify": { "roboco-notify": {
"command": "uv", "command": "uv",
"args": [ "args": [
"run", "python", "-m", "roboco.mcp.notify_server", agent_id "run",
"python",
"-m",
"roboco.mcp.notify_server",
agent_id,
], ],
"env": mcp_env,
}, },
"roboco-journal": { "roboco-journal": {
"command": "uv", "command": "uv",
"args": [ "args": [
"run", "python", "-m", "roboco.mcp.journal_server", agent_id "run",
"python",
"-m",
"roboco.mcp.journal_server",
agent_id,
], ],
"env": mcp_env,
}, },
} }
} }
@@ -465,7 +524,11 @@ class AgentOrchestrator:
if graceful: if graceful:
# Graceful stop with timeout # Graceful stop with timeout
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "stop", "-t", "10", container_name, "docker",
"stop",
"-t",
"10",
container_name,
stdout=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
) )
@@ -473,7 +536,9 @@ class AgentOrchestrator:
else: else:
# Force kill # Force kill
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "kill", container_name, "docker",
"kill",
container_name,
stdout=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
) )
@@ -644,7 +709,11 @@ Start by:
# Check if container is still running # Check if container is still running
container_name = f"roboco-agent-{agent_id}" container_name = f"roboco-agent-{agent_id}"
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
"docker", "inspect", "-f", "{{.State.Running}}", container_name, "docker",
"inspect",
"-f",
"{{.State.Running}}",
container_name,
stdout=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
) )
@@ -725,3 +794,620 @@ Start by:
"waiting_count": len(self._waiting_records), "waiting_count": len(self._waiting_records),
"agents": agents, "agents": agents,
} }
# =========================================================================
# SMART DISPATCHER - API HELPERS
# =========================================================================
@property
def _api_url(self) -> str:
"""Get the internal API URL for task/notification queries."""
return settings.internal_api_url
def _is_agent_active(self, agent_id: str) -> bool:
"""Check if an agent is currently running."""
if agent_id not in self._instances:
return False
return self._instances[agent_id].state == AgentState.ACTIVE
def _select_agent_for_cell(self, cell: str, role: str) -> str | None:
"""
Select the best available agent for a cell and role.
Prefers agents that are not currently active.
For developers, uses round-robin among candidates.
"""
prefix_map = {"backend": "be", "frontend": "fe", "uxui": "ux"}
prefix = prefix_map.get(cell)
if not prefix:
return None
# Build candidate list based on role
if role == "dev":
if prefix == "ux":
candidates = ["ux-dev"]
else:
candidates = [f"{prefix}-dev-1", f"{prefix}-dev-2"]
elif role == "qa":
candidates = [f"{prefix}-qa"]
elif role == "doc":
candidates = [f"{prefix}-doc"]
elif role == "pm":
candidates = [f"{prefix}-pm"]
else:
return None
# Prefer non-active agents
for agent_id in candidates:
if not self._is_agent_active(agent_id):
return agent_id
# All active - return first (task will queue for them via scan)
return candidates[0]
async def _claim_task_for_agent(
self,
client: httpx.AsyncClient,
task_id: str,
agent_id: str,
) -> bool:
"""Claim a task on behalf of an agent before spawning."""
try:
resp = await client.post(
f"{self._api_url}/tasks/{task_id}/claim",
json={"agent_id": agent_id},
)
if resp.status_code == http_status.HTTP_200_OK:
logger.info(
"Task claimed for agent",
task_id=task_id,
agent_id=agent_id,
)
return True
logger.warning(
"Failed to claim task",
task_id=task_id,
agent_id=agent_id,
status=resp.status_code,
)
except Exception as e:
logger.error("Claim task error", task_id=task_id, error=str(e))
return False
async def _fetch_tasks(
self,
client: httpx.AsyncClient,
status: str | list[str],
team: str | None = None,
) -> list[dict[str, Any]]:
"""Fetch tasks by status and optional team filter."""
params: dict[str, Any] = {}
if isinstance(status, list):
params["status"] = ",".join(status)
else:
params["status"] = status
if team:
params["team"] = team
try:
resp = await client.get(f"{self._api_url}/tasks", params=params)
if resp.status_code == http_status.HTTP_200_OK:
result: list[dict[str, Any]] = resp.json()
return result
except Exception as e:
logger.error("Fetch tasks error", status=status, error=str(e))
return []
async def _fetch_notifications(
self,
client: httpx.AsyncClient,
notification_type: str,
unacknowledged: bool = True,
) -> list[dict[str, Any]]:
"""Fetch notifications by type."""
params: dict[str, Any] = {
"type": notification_type,
"pending_ack_only": str(unacknowledged).lower(),
}
try:
resp = await client.get(
f"{self._api_url}/notifications",
params=params,
)
if resp.status_code == http_status.HTTP_200_OK:
data = resp.json()
items: list[dict[str, Any]] = data.get("items", [])
return items
except Exception as e:
logger.error(
"Fetch notifications error",
notification_type=notification_type,
error=str(e),
)
return []
# =========================================================================
# SMART DISPATCHER - MAIN LOOP
# =========================================================================
async def _dispatcher_loop(self) -> None:
"""
Main dispatcher loop - periodically checks for work and spawns agents.
This is the BRAIN of the orchestrator. It:
1. Queries for tasks needing work (pending, awaiting_qa, etc.)
2. Queries for events needing attention (blockers, escalations)
3. Spawns appropriate agents with task assignments
"""
while self._running:
try:
await asyncio.sleep(self.dispatcher_interval)
await self._dispatch_all_work()
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Dispatcher loop error", error=str(e))
async def _dispatch_all_work(self) -> None:
"""Run all dispatchers to check for and assign work."""
async with httpx.AsyncClient(timeout=30.0) as client:
# Task-based dispatchers (check task statuses)
await self._dispatch_dev_work(client)
await self._dispatch_qa_work(client)
await self._dispatch_doc_work(client)
await self._dispatch_marketing_work(client)
# Event-based dispatchers (check blockers, notifications)
await self._dispatch_blocker_work(client)
await self._dispatch_escalation_work(client)
await self._dispatch_approval_work(client)
# Scheduled dispatchers
await self._dispatch_audit_work(client)
# =========================================================================
# SMART DISPATCHER - TASK-BASED DISPATCHERS
# =========================================================================
async def _dispatch_dev_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch development work to developers.
Monitors: pending, needs_revision tasks
Spawns: be-dev-1, be-dev-2, fe-dev-1, fe-dev-2, ux-dev
"""
# Get tasks needing dev attention
tasks = await self._fetch_tasks(client, ["pending", "needs_revision"])
for task in tasks:
# Skip already claimed/assigned tasks
if task.get("assigned_to"):
continue
team = task.get("team")
if team not in ["backend", "frontend", "uxui"]:
continue
# Select best agent for this task
agent_id = self._select_agent_for_cell(team, "dev")
if not agent_id:
continue
# If agent is already active, just claim for them
# They'll pick it up on their next roboco_task_scan()
if self._is_agent_active(agent_id):
await self._claim_task_for_agent(client, task["id"], agent_id)
# Claim and spawn with assignment
elif await self._claim_task_for_agent(client, task["id"], agent_id):
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_dev_prompt(task),
)
async def _dispatch_qa_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch QA work to QA agents.
Monitors: awaiting_qa tasks
Spawns: be-qa, fe-qa, ux-qa
"""
tasks = await self._fetch_tasks(client, "awaiting_qa")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "uxui"]:
continue
agent_id = self._select_agent_for_cell(team, "qa")
if not agent_id:
continue
if self._is_agent_active(agent_id):
# QA already running, they'll pick up on scan
continue
# Spawn QA agent with task assignment
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_qa_prompt(task),
)
# Only spawn one QA at a time per cell
break
async def _dispatch_doc_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch documentation work to documenters.
Monitors: awaiting_documentation tasks
Spawns: be-doc, fe-doc, ux-doc
"""
tasks = await self._fetch_tasks(client, "awaiting_documentation")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "uxui"]:
continue
agent_id = self._select_agent_for_cell(team, "doc")
if not agent_id:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_doc_prompt(task),
)
break
async def _dispatch_marketing_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch marketing work to head-marketing.
Monitors: pending tasks with team=marketing
Spawns: head-marketing
"""
tasks = await self._fetch_tasks(client, "pending", team="marketing")
for task in tasks:
# Skip already claimed/assigned tasks
if task.get("assigned_to"):
continue
if self._is_agent_active("head-marketing"):
# Already running, they'll pick up on scan
continue
await self.spawn_agent(
agent_id="head-marketing",
task_id=task["id"],
initial_prompt=self._build_marketing_prompt(task),
)
break
# =========================================================================
# SMART DISPATCHER - EVENT-BASED DISPATCHERS
# =========================================================================
async def _dispatch_blocker_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch blocker resolution to Cell PMs.
Monitors: blocked tasks
Spawns: be-pm, fe-pm, ux-pm
"""
tasks = await self._fetch_tasks(client, "blocked")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "uxui"]:
continue
agent_id = self._select_agent_for_cell(team, "pm")
if not agent_id:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_pm_blocker_prompt(task),
)
break
async def _dispatch_escalation_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch escalations to appropriate managers.
Monitors: escalation notifications (unacknowledged)
Spawns: be-pm, fe-pm, ux-pm, main-pm, product-owner, head-marketing
"""
notifications = await self._fetch_notifications(client, "escalation")
for notif in notifications:
targets = notif.get("to_agents", [])
for agent_id in targets:
valid_targets = [
"be-pm",
"fe-pm",
"ux-pm",
"main-pm",
"product-owner",
"head-marketing",
]
if agent_id not in valid_targets:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
initial_prompt=self._build_escalation_prompt(notif),
)
break
async def _dispatch_approval_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch approval requests to approvers.
Monitors: approval notifications (unacknowledged)
Spawns: product-owner, head-marketing, main-pm
"""
notifications = await self._fetch_notifications(client, "approval")
for notif in notifications:
targets = notif.get("to_agents", [])
for agent_id in targets:
if agent_id not in ["product-owner", "head-marketing", "main-pm"]:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
initial_prompt=self._build_approval_prompt(notif),
)
break
async def _dispatch_audit_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch audit work to the auditor.
Monitors: quality alert notifications
Spawns: auditor
Note: Periodic scheduled audits can be added here in the future.
"""
alerts = await self._fetch_notifications(client, "alert")
for alert in alerts:
targets = alert.get("to_agents", [])
if "auditor" in targets and not self._is_agent_active("auditor"):
await self.spawn_agent(
agent_id="auditor",
initial_prompt=self._build_audit_prompt(alert),
)
return
# TODO: Add scheduled periodic audits
# Check last audit time, spawn if overdue
# =========================================================================
# SMART DISPATCHER - PROMPT BUILDERS
# =========================================================================
def _build_dev_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a developer with an assigned task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
status = task.get("status", "unknown")
team = task.get("team", "unknown")
return f"""You have been assigned a development task.
TASK ID: {task_id}
TITLE: {title}
STATUS: {status}
TEAM: {team}
This task is already CLAIMED for you. Begin work immediately:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Follow the workflow: UNDERSTAND PLAN EXECUTE VERIFY SUBMIT QA
3. When task is submitted for QA, call roboco_task_scan() to check for more work
4. If more work is assigned to you, continue working
5. If no more work, call roboco_agent_idle() to shutdown gracefully
Do NOT scan for work first - your task is already assigned. Begin now.
"""
def _build_qa_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a QA agent."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
assigned_to = task.get("assigned_to", "unknown")
team = task.get("team", "unknown")
return f"""A task is ready for QA review.
TASK ID: {task_id}
TITLE: {title}
DEVELOPER: {assigned_to}
TEAM: {team}
Begin QA review:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Review the implementation against ALL acceptance criteria
3. Test the changes thoroughly
4. Call roboco_task_qa_pass() with notes if approved
OR roboco_task_qa_fail() with specific issues if rejected
5. Call roboco_task_scan() to check for more QA work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_doc_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a documenter."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
team = task.get("team", "unknown")
return f"""A task is ready for documentation.
TASK ID: {task_id}
TITLE: {title}
TEAM: {team}
Begin documentation:
1. Call roboco_task_get("{task_id}") for full details and dev handoff notes
2. Create or update documentation based on what was implemented
3. Ensure code comments, README updates, API docs as needed
4. Call roboco_task_complete("{task_id}") when documentation is done
5. Call roboco_task_scan() to check for more documentation work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_marketing_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for head-marketing with a marketing task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
description = task.get("description", "No description")
return f"""You have been assigned a marketing task.
TASK ID: {task_id}
TITLE: {title}
DESCRIPTION: {description}
Begin work:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Execute the marketing task (content, campaigns, research, etc.)
3. Coordinate with Product Owner or Main PM if needed
4. Call roboco_task_complete("{task_id}") when done
5. Call roboco_task_scan() to check for more marketing work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_pm_blocker_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a Cell PM handling a blocker."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
assigned_to = task.get("assigned_to", "unknown")
blocker = task.get("blocker", {})
reason = blocker.get("reason", "Unknown")
what_needed = blocker.get("what_needed", "Unknown")
return f"""A task in your cell is BLOCKED and needs your attention.
TASK ID: {task_id}
TITLE: {title}
ASSIGNED TO: {assigned_to}
BLOCKER REASON: {reason}
WHAT'S NEEDED: {what_needed}
Your job:
1. Understand the blocker by reviewing task details
2. Communicate with the blocked developer if needed
3. Resolve the blocker (coordinate resources, make decisions, escalate if needed)
4. Once resolved, the developer can call roboco_task_unblock()
5. Call roboco_task_scan() to check for other blocked tasks in your cell
6. If no more blockers, call roboco_agent_idle() to shutdown gracefully
"""
def _build_escalation_prompt(self, notification: dict[str, Any]) -> str:
"""Build initial prompt for handling an escalation."""
notif_id = notification.get("id", "unknown")
from_agent = notification.get("from_agent", "unknown")
subject = notification.get("subject", "No subject")
priority = notification.get("priority", "normal")
body = notification.get("body", "No details provided")
return f"""You have received an ESCALATION that requires your attention.
FROM: {from_agent}
SUBJECT: {subject}
PRIORITY: {priority}
DETAILS:
{body}
Your job:
1. Acknowledge the notification with roboco_notify_ack("{notif_id}")
2. Assess the escalation and determine action needed
3. Communicate decisions via appropriate channels
4. If this requires further escalation, use roboco_escalate()
5. When resolved, call roboco_task_scan() for other work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_approval_prompt(self, notification: dict[str, Any]) -> str:
"""Build initial prompt for handling an approval request."""
notif_id = notification.get("id", "unknown")
from_agent = notification.get("from_agent", "unknown")
subject = notification.get("subject", "No subject")
related_task_id = notification.get("related_task_id", "None")
body = notification.get("body", "No details provided")
return f"""You have received an APPROVAL REQUEST.
FROM: {from_agent}
SUBJECT: {subject}
RELATED TASK: {related_task_id}
REQUEST:
{body}
Your job:
1. Review the approval request carefully
2. If related to a task, call roboco_task_get() for context
3. Make your decision and communicate it
4. Acknowledge with roboco_notify_ack("{notif_id}")
5. Call roboco_task_scan() for other work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_audit_prompt(self, alert: dict[str, Any] | None = None) -> str:
"""Build initial prompt for the auditor."""
if alert:
subject = alert.get("subject", "Quality issue detected")
body = alert.get("body", "Review system quality metrics")
return f"""QUALITY ALERT triggered your attention.
ALERT: {subject}
DETAILS: {body}
Your job:
1. Investigate the quality issue
2. Review relevant channels and task history (you have read access to all)
3. Compile your findings
4. Report to CEO via appropriate channel
5. Call roboco_agent_idle() when complete
"""
return """Periodic AUDIT requested.
Your job:
1. Review recent activity across all cells
2. Check quality metrics (QA pass/fail rates, blocker frequency, etc.)
3. Identify any concerns or patterns
4. Compile audit report for CEO
5. Call roboco_agent_idle() when complete
"""