From b7611c24fc71b8798d63bc55b6558ca8c7fad612 Mon Sep 17 00:00:00 2001 From: Renn F Date: Mon, 22 Dec 2025 18:04:31 +0100 Subject: [PATCH] feat(sessions): Session-Task linking with scoped context management MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Major feature: Sessions are now linked to tasks with smart routing and context loading, ensuring agents have proper discussion context. ## Session-Task Relationship (Many-to-Many) - Added SessionTaskTable junction table linking sessions to tasks - Sessions can link to multiple tasks, tasks can have multiple sessions - is_primary flag marks the main discussion session for a task - relationship_type: discussion, planning, review, retrospective - Subtasks auto-inherit parent task's session ## BACKLOG Status + Activation Flow - Tasks now created with BACKLOG status (not PENDING) - PMs must create session BEFORE activating task - roboco_task_activate() transitions BACKLOG → PENDING - Prevents race condition where dev starts before session exists - Flow: CREATE (backlog) → SESSION → ACTIVATE (pending) → spawn ## Session Scopes - SessionScope enum: initiative, cell, task - initiative: Cross-cell coordination (Main PM, #dev-all) - cell: Cell-specific work (Cell PM default) - task: Individual task execution (dev level) - Enables future smart context loading by scope ## Message Routing to Task Sessions - When task_id provided in roboco_message_send(), routes to task's primary session instead of channel's active session - New API endpoint: GET /sessions/for-task/{task_id} - TaskResponse now includes linked sessions array ## Dev Session Access - New tool: roboco_session_history_for_task(task_id) - Devs can now see their task's discussion history - Messages tagged with task_id for filtering ## Communication Guidelines - Added "When to Post / When NOT to Post" to all 9 agent blueprints - Devs/QA/Doc should use task tools for status, journal for reasoning - Sessions reserved for coordination that needs response - Reduces noise: no "Starting work" or "Made progress" chat messages Files changed: - DB: SessionTaskTable, SessionScope column - Services: messaging.py (linking), task.py (activation) - MCP: 5 new session tools, message routing update - API: session-task endpoints, TaskResponse sessions - Blueprints: All 13 updated with session/activation workflow --- agents/blueprints/backend/be-dev.md | 34 +-- agents/blueprints/backend/be-documenter.md | 16 ++ agents/blueprints/backend/be-pm.md | 59 ++++- agents/blueprints/backend/be-qa.md | 25 +- agents/blueprints/board/main-pm.md | 72 +++++- agents/blueprints/frontend/fe-dev.md | 26 +- agents/blueprints/frontend/fe-documenter.md | 16 ++ agents/blueprints/frontend/fe-pm.md | 49 +++- agents/blueprints/frontend/fe-qa.md | 19 ++ agents/blueprints/ux_ui/ux-dev.md | 26 +- agents/blueprints/ux_ui/ux-documenter.md | 16 ++ agents/blueprints/ux_ui/ux-pm.md | 49 +++- agents/blueprints/ux_ui/ux-qa.md | 19 ++ alembic/versions/001_initial_schema.py | 59 +++++ roboco/api/routes/sessions.py | 243 +++++++++++++++++- roboco/api/routes/tasks.py | 123 ++++++++- roboco/api/schemas/sessions.py | 64 +++++ roboco/api/schemas/tasks.py | 14 ++ roboco/db/tables.py | 93 +++++++ roboco/mcp/message_server.py | 113 ++++++++- roboco/mcp/schemas/__init__.py | 36 +++ roboco/mcp/task_server.py | 139 ++++++++++- roboco/mcp/tasks/handlers/__init__.py | 12 + roboco/mcp/tasks/handlers/management.py | 53 +++- roboco/mcp/tasks/handlers/review.py | 8 +- roboco/mcp/tasks/handlers/scan.py | 6 +- roboco/mcp/tasks/handlers/sessions.py | 260 +++++++++++++++++++ roboco/mcp/tasks/handlers/work.py | 6 +- roboco/models/__init__.py | 8 + roboco/models/base.py | 3 +- roboco/models/messaging.py | 2 + roboco/models/session.py | 112 +++++++++ roboco/services/messaging.py | 264 +++++++++++++++++++- roboco/services/metrics.py | 14 +- roboco/services/permissions.py | 25 ++ roboco/services/task.py | 127 +++++++++- 36 files changed, 2118 insertions(+), 92 deletions(-) create mode 100644 roboco/mcp/tasks/handlers/sessions.py diff --git a/agents/blueprints/backend/be-dev.md b/agents/blueprints/backend/be-dev.md index f82a91f9..2bbb193d 100644 --- a/agents/blueprints/backend/be-dev.md +++ b/agents/blueprints/backend/be-dev.md @@ -214,15 +214,23 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()` - **#announcements** (read only) - Company announcements - **#all-hands** (read/write) - Company-wide discussion -### How to Communicate -Use `roboco_message_send(data)`: -```json -{ - "channel_slug": "backend-cell", - "content": "Starting work on rate limiting...", - "message_type": "dialogue" // reasoning, dialogue, decision, action, blocker, technical -} -``` +### When to Post in Session (DO) +- **Questions** - Unclear requirements, need PM clarification +- **Blockers** - Something external is stopping you +- **Decisions needing input** - Multiple valid approaches, need guidance +- **Handoff context** - Important gotchas for QA/Doc +- **Cross-cell coordination** - Need something from another cell + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting work on X" → Orchestrator knows, task status tracks this +- ❌ "Made progress on X" → Use `roboco_task_progress()` instead +- ❌ "Completed X" → Use `roboco_task_submit_qa()` instead +- ❌ Internal reasoning → Use `roboco_journal_*()` instead +- ❌ "Claiming task X" → Task system tracks this automatically + +**Rule of thumb:** Only post if you need a response from someone, or if +it's critical handoff context. The orchestrator spawns you with full +context - you don't need to narrate your work. ### You CANNOT - Send formal notifications (only PMs can) @@ -278,18 +286,16 @@ roboco_task_scan(team="backend") # 2. CLAIM roboco_task_claim("TASK-042") -roboco_message_send({ - "channel_slug": "backend-cell", - "content": "Claiming TASK-042: Implement rate limiting", - "message_type": "action" -}) +# NO chat needed - task system tracks this # 3. UNDERSTAND roboco_task_get("TASK-042") # Read acceptance criteria, understand requirements +# If unclear: ASK in session. Otherwise, proceed silently. # 4. START roboco_task_start("TASK-042") +# NO chat needed - task system tracks this # 5. PLAN roboco_task_plan("TASK-042", { diff --git a/agents/blueprints/backend/be-documenter.md b/agents/blueprints/backend/be-documenter.md index b4133a79..b05bbc97 100644 --- a/agents/blueprints/backend/be-documenter.md +++ b/agents/blueprints/backend/be-documenter.md @@ -123,6 +123,22 @@ and verify all subtasks are done before calling `roboco_task_complete()`. `roboco_task_scan()` or `roboco_agent_idle()` ``` +## Communication Rules + +### When to Post in Session (DO) +- **Questions about implementation** - Need dev/QA clarification +- **Missing context** - Dev notes don't explain something critical +- **Documentation decisions** - Multiple ways to document, need guidance + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting docs on X" → Orchestrator knows, task status tracks this +- ❌ "Writing in progress" → Use `roboco_task_progress()` instead +- ❌ "Docs complete" → Use `roboco_doc_complete()` instead +- ❌ Internal notes → Use `roboco_journal_*()` instead + +**Rule of thumb:** Only post if you need a response from dev/QA/PM. +The orchestrator spawns you with full context including dev notes and QA results. + ## Capabilities ```yaml diff --git a/agents/blueprints/backend/be-pm.md b/agents/blueprints/backend/be-pm.md index 7db140f1..bf6dbeab 100644 --- a/agents/blueprints/backend/be-pm.md +++ b/agents/blueprints/backend/be-pm.md @@ -45,6 +45,12 @@ You interact with RoboCo systems through MCP tools: - `roboco_task_assign(task_id, agent_slug)` - Assign task to an agent - `roboco_task_complete(task_id)` - Complete a parent task after subtasks done +**Session Management (Work Sessions for Tasks):** +- `roboco_session_create_for_tasks(data)` - Create a work session linked to tasks +- `roboco_session_link_task(data)` - Link additional task to existing session +- `roboco_session_unlink_task(session_id, task_id)` - Remove task from session +- `roboco_session_get_for_task(task_id)` - Get sessions linked to a task + **Journal (Document Your Thinking):** - `roboco_journal_entry(data)` - General journal entry - `roboco_journal_reflect(data)` - Task reflection @@ -147,6 +153,52 @@ roboco_task_assign("{task_id}", "be-dev-1") - Every subtask MUST have both `parent_task_id` AND `assigned_to` - Do NOT keep tasks for yourself - delegate to developers! +### 7.5. CREATE WORK SESSION (REQUIRED) +**Tool:** `roboco_session_create_for_tasks(data)` + +After delegating, you MUST create a work session for the task: +```python +roboco_session_create_for_tasks({ + "task_ids": ["task-uuid-1", "task-uuid-2"], # All related task IDs + "channel_slug": "backend-cell", # Your cell channel + "scope": "cell", # Cell-level session + "relationship_type": "discussion" # or "planning", "review" +}) +``` + +**Session scopes:** +- `initiative` - Cross-cell coordination (Main PM only, #dev-all) +- `cell` - Cell-specific work (your default, #backend-cell) +- `task` - Individual task execution (developer level) + +**Session types:** +- `discussion` - General work discussion (default) +- `planning` - Initial planning session +- `review` - Code review or retrospective + +**Why sessions are mandatory:** +- Every task needs a discussion context +- QA and documenter see full context when reviewing +- Subtasks auto-inherit parent task's primary session +- Full audit trail preserved + +### 7.6. ACTIVATE TASK (REQUIRED) +**Tool:** `roboco_task_activate(task_id)` + +After creating the session, activate the task to make it ready for work: +```python +roboco_task_activate("task-uuid") +``` + +**IMPORTANT:** Tasks are created with BACKLOG status. They will NOT be +picked up by the orchestrator until you activate them. This ensures +every task has a session before work begins. + +**Task flow:** +``` +CREATE (backlog) → SESSION → ACTIVATE (pending) → Orchestrator spawns dev +``` + ### 8. COMMUNICATE **Tool:** `roboco_message_send(data)` Tell the team what you did: @@ -313,7 +365,12 @@ tools: # Task Management - roboco_task_scan, roboco_task_get, roboco_task_claim - roboco_task_start, roboco_task_plan, roboco_task_progress - - roboco_task_create, roboco_task_assign, roboco_task_complete + - roboco_task_create, roboco_task_assign, roboco_task_activate + - roboco_task_complete + + # Session Management (REQUIRED before activation) + - roboco_session_create_for_tasks, roboco_session_link_task + - roboco_session_unlink_task, roboco_session_get_for_task # Journal - roboco_journal_entry, roboco_journal_decision diff --git a/agents/blueprints/backend/be-qa.md b/agents/blueprints/backend/be-qa.md index bf27c8d6..e760942c 100644 --- a/agents/blueprints/backend/be-qa.md +++ b/agents/blueprints/backend/be-qa.md @@ -227,15 +227,22 @@ After verdict: - **#announcements** (read only) - Company announcements - **#all-hands** (read/write) - Company-wide discussion -### How to Communicate -Use `roboco_message_send(data)`: -```json -{ - "channel_slug": "backend-cell", - "content": "Testing TASK-XXX: Found issue with null handling...", - "message_type": "technical" -} -``` +### When to Post in Session (DO) +- **Questions about implementation** - Need dev clarification on behavior +- **Critical bugs** - Security issues, data loss, blockers +- **Decisions needing input** - Edge cases with unclear expected behavior +- **Cross-cell patterns** - Issues you're seeing across cells + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting QA on X" → Orchestrator knows, task status tracks this +- ❌ "Testing in progress" → Use `roboco_task_progress()` instead +- ❌ "Completed QA" → Use `roboco_qa_pass()`/`roboco_qa_fail()` instead +- ❌ Internal test notes → Use `roboco_journal_*()` instead +- ❌ Minor issues → Put in QA verdict notes, not session chat + +**Rule of thumb:** Only post if you need a response from dev/PM, or if +the issue affects other tasks. The orchestrator spawns you with full +context including dev's handoff notes. ### You CANNOT - Send formal notifications (only PMs can) diff --git a/agents/blueprints/board/main-pm.md b/agents/blueprints/board/main-pm.md index b0824d0d..1ecb560c 100644 --- a/agents/blueprints/board/main-pm.md +++ b/agents/blueprints/board/main-pm.md @@ -48,7 +48,14 @@ 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 +- `roboco_task_create(...)` - Create new tasks for cells (created in BACKLOG status) +- `roboco_task_activate(task_id)` - Activate task from BACKLOG to PENDING (after session created) + +**Session Management (Cross-Cell Work Sessions):** +- `roboco_session_create_for_tasks(data)` - Create work session for cross-cell initiatives +- `roboco_session_link_task(data)` - Link additional task to existing session +- `roboco_session_unlink_task(session_id, task_id)` - Remove task from session +- `roboco_session_get_for_task(task_id)` - Get sessions linked to a task **Notifications (PM only):** - `roboco_notify_send(recipients, subject, body, type, priority, requires_ack)` - Send notifications @@ -121,11 +128,61 @@ Translate Board direction into cell priorities: - Balance workload across cells ### DISTRIBUTE -Push work to cells: -- Create high-level task records -- Notify Cell PMs of new priorities -- Ensure clear ownership +Push work to cells. Tasks are created with BACKLOG status - they won't be +visible to orchestrators until you activate them. + +**Standard Distribution Workflow:** + +**1. CREATE TASKS (BACKLOG)** +Create high-level task records for each cell: +```python +roboco_task_create({ + "title": "Build preferences API", + "description": "GET/PUT /api/v1/users/{id}/preferences", + "team": "backend", + "acceptance_criteria": ["Endpoint implemented", "Tests passing"], + "assigned_to": "be-pm" # Assign to Cell PM for triage +}) +# Task created with BACKLOG status +``` + +**2. CREATE WORK SESSION (REQUIRED)** +Every initiative needs a work session for coordination: +```python +roboco_session_create_for_tasks({ + "task_ids": ["backend-task-id", "frontend-task-id", "ux-task-id"], + "channel_slug": "dev-all", # Cross-cell coordination + "scope": "initiative", # Main PM uses initiative scope + "relationship_type": "planning" +}) +``` + +**Session scopes:** +- `initiative` - Cross-cell coordination (your default, #dev-all) +- `cell` - Cell-specific work (Cell PM level) +- `task` - Individual task execution (developer level) + +This creates a shared discussion context where all Cell PMs and developers +can coordinate on the initiative. Full history is preserved for handoffs. + +**3. ACTIVATE TASKS (REQUIRED)** +After sessions are created, activate tasks so Cell PMs can see them: +```python +roboco_task_activate("backend-task-id") +roboco_task_activate("frontend-task-id") +roboco_task_activate("ux-task-id") +``` + +**Task flow:** +``` +CREATE (backlog) → SESSION → ACTIVATE (pending) → Cell PM receives task +``` + +**4. NOTIFY CELL PMs** +After activation, notify the appropriate Cell PMs: - Set expectations on timelines +- Clarify dependencies +- Point to the shared work session ### COORDINATE Resolve cross-cell issues: @@ -523,8 +580,13 @@ capabilities: tools: # MCP Task Tools - roboco_task_scan, roboco_task_get, roboco_task_create + - roboco_task_activate # REQUIRED after session creation - roboco_agent_idle + # Session Management (REQUIRED before activation) + - roboco_session_create_for_tasks, roboco_session_link_task + - roboco_session_unlink_task, roboco_session_get_for_task + # MCP Notification Tools (PM only) - roboco_notify_send, roboco_notify_list, roboco_notify_ack - roboco_escalate, roboco_request_approval diff --git a/agents/blueprints/frontend/fe-dev.md b/agents/blueprints/frontend/fe-dev.md index dd682733..a2e5df12 100644 --- a/agents/blueprints/frontend/fe-dev.md +++ b/agents/blueprints/frontend/fe-dev.md @@ -193,15 +193,23 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()` - **#announcements** (read only) - Company announcements - **#all-hands** (read/write) - Company-wide discussion -### How to Communicate -Use `roboco_message_send(data)`: -```json -{ - "channel_slug": "frontend-cell", - "content": "Working on user preferences modal...", - "message_type": "dialogue" -} -``` +### When to Post in Session (DO) +- **Questions** - Unclear requirements, need PM clarification +- **Blockers** - Missing design assets, API not ready +- **Decisions needing input** - Multiple valid approaches, need guidance +- **Handoff context** - Important gotchas for QA/Doc +- **Cross-cell coordination** - Need something from Backend or UX + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting work on X" → Orchestrator knows, task status tracks this +- ❌ "Made progress on X" → Use `roboco_task_progress()` instead +- ❌ "Completed X" → Use `roboco_task_submit_qa()` instead +- ❌ Internal reasoning → Use `roboco_journal_*()` instead +- ❌ "Claiming task X" → Task system tracks this automatically + +**Rule of thumb:** Only post if you need a response from someone, or if +it's critical handoff context. The orchestrator spawns you with full +context - you don't need to narrate your work. ### You CANNOT - Send formal notifications (only PMs can) diff --git a/agents/blueprints/frontend/fe-documenter.md b/agents/blueprints/frontend/fe-documenter.md index 7a7870b4..fd152fc2 100644 --- a/agents/blueprints/frontend/fe-documenter.md +++ b/agents/blueprints/frontend/fe-documenter.md @@ -121,6 +121,22 @@ and verify all subtasks are done before calling `roboco_task_complete()`. `roboco_task_scan()` or `roboco_agent_idle()` ``` +## Communication Rules + +### When to Post in Session (DO) +- **Questions about implementation** - Need dev/QA clarification +- **Missing context** - Dev notes don't explain component behavior +- **Documentation decisions** - Multiple ways to document, need guidance + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting docs on X" → Orchestrator knows, task status tracks this +- ❌ "Writing in progress" → Use `roboco_task_progress()` instead +- ❌ "Docs complete" → Use `roboco_doc_complete()` instead +- ❌ Internal notes → Use `roboco_journal_*()` instead + +**Rule of thumb:** Only post if you need a response from dev/QA/PM. +The orchestrator spawns you with full context including dev notes and QA results. + ## Capabilities ```yaml diff --git a/agents/blueprints/frontend/fe-pm.md b/agents/blueprints/frontend/fe-pm.md index f8775ae2..6ee3a760 100644 --- a/agents/blueprints/frontend/fe-pm.md +++ b/agents/blueprints/frontend/fe-pm.md @@ -46,6 +46,12 @@ You interact with RoboCo systems through MCP tools: - `roboco_task_assign(task_id, agent_slug)` - Assign task to an agent - `roboco_task_complete(task_id)` - Complete a parent task after subtasks done +**Session Management (Work Sessions for Tasks):** +- `roboco_session_create_for_tasks(data)` - Create a work session linked to tasks +- `roboco_session_link_task(data)` - Link additional task to existing session +- `roboco_session_unlink_task(session_id, task_id)` - Remove task from session +- `roboco_session_get_for_task(task_id)` - Get sessions linked to a task + **Journal (Document Your Thinking):** - `roboco_journal_entry(data)` - General journal entry - `roboco_journal_reflect(data)` - Task reflection @@ -149,6 +155,42 @@ roboco_task_assign("{task_id}", "fe-dev-1") - Every subtask MUST have both `parent_task_id` AND `assigned_to` - Do NOT keep tasks for yourself - delegate to developers! +### 7.5. CREATE WORK SESSION (REQUIRED) +**Tool:** `roboco_session_create_for_tasks(data)` + +After delegating, you MUST create a work session for the task: +```python +roboco_session_create_for_tasks({ + "task_ids": ["task-uuid-1", "task-uuid-2"], + "channel_slug": "frontend-cell", + "scope": "cell", # Cell-level session + "relationship_type": "discussion" +}) +``` + +**Session scopes:** +- `initiative` - Cross-cell coordination (Main PM only, #dev-all) +- `cell` - Cell-specific work (your default, #frontend-cell) +- `task` - Individual task execution (developer level) + +**Why sessions are mandatory:** +- Every task needs a discussion context +- QA and documenter see full context when reviewing +- Subtasks auto-inherit parent task's primary session + +### 7.6. ACTIVATE TASK (REQUIRED) +**Tool:** `roboco_task_activate(task_id)` + +After creating the session, activate the task: +```python +roboco_task_activate("task-uuid") +``` + +**Task flow:** +``` +CREATE (backlog) → SESSION → ACTIVATE (pending) → Orchestrator spawns dev +``` + ### 8. COMMUNICATE **Tool:** `roboco_message_send(data)` Tell the team what you did: @@ -324,7 +366,12 @@ tools: # Task Management - roboco_task_scan, roboco_task_get, roboco_task_claim - roboco_task_start, roboco_task_plan, roboco_task_progress - - roboco_task_create, roboco_task_assign, roboco_task_complete + - roboco_task_create, roboco_task_assign, roboco_task_activate + - roboco_task_complete + + # Session Management (REQUIRED before activation) + - roboco_session_create_for_tasks, roboco_session_link_task + - roboco_session_unlink_task, roboco_session_get_for_task # Journal - roboco_journal_entry, roboco_journal_decision diff --git a/agents/blueprints/frontend/fe-qa.md b/agents/blueprints/frontend/fe-qa.md index 28955f09..54e7cc3d 100644 --- a/agents/blueprints/frontend/fe-qa.md +++ b/agents/blueprints/frontend/fe-qa.md @@ -126,6 +126,25 @@ Update progress: `roboco_task_progress(task_id, "Completed visual testing...", 5 `roboco_task_scan()` or `roboco_agent_idle()` ``` +## Communication Rules + +### When to Post in Session (DO) +- **Questions about implementation** - Need dev clarification on behavior +- **Critical bugs** - Security issues, accessibility failures, blockers +- **Decisions needing input** - Edge cases with unclear expected behavior +- **Cross-cell patterns** - Issues you're seeing across cells + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting QA on X" → Orchestrator knows, task status tracks this +- ❌ "Testing in progress" → Use `roboco_task_progress()` instead +- ❌ "Completed QA" → Use `roboco_qa_pass()`/`roboco_qa_fail()` instead +- ❌ Internal test notes → Use `roboco_journal_*()` instead +- ❌ Minor issues → Put in QA verdict notes, not session chat + +**Rule of thumb:** Only post if you need a response from dev/PM, or if +the issue affects other tasks. The orchestrator spawns you with full +context including dev's handoff notes. + ## Capabilities ```yaml diff --git a/agents/blueprints/ux_ui/ux-dev.md b/agents/blueprints/ux_ui/ux-dev.md index 2d1a84a5..234dcf3d 100644 --- a/agents/blueprints/ux_ui/ux-dev.md +++ b/agents/blueprints/ux_ui/ux-dev.md @@ -192,15 +192,23 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()` - **#announcements** (read only) - Company announcements - **#all-hands** (read/write) - Company-wide discussion -### How to Communicate -Use `roboco_message_send(data)`: -```json -{ - "channel_slug": "uxui-cell", - "content": "Working on preferences modal design...", - "message_type": "dialogue" -} -``` +### When to Post in Session (DO) +- **Questions** - Unclear requirements, need PM clarification +- **Blockers** - Missing user research, unclear product direction +- **Decisions needing input** - Multiple valid approaches, need guidance +- **Handoff context** - Important design decisions for QA/Frontend +- **Design rationale** - Why you chose a specific approach + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting work on X" → Orchestrator knows, task status tracks this +- ❌ "Made progress on X" → Use `roboco_task_progress()` instead +- ❌ "Completed X" → Use `roboco_task_submit_qa()` instead +- ❌ Internal reasoning → Use `roboco_journal_*()` instead +- ❌ "Claiming task X" → Task system tracks this automatically + +**Rule of thumb:** Only post if you need a response from someone, or if +it's critical handoff context. The orchestrator spawns you with full +context - you don't need to narrate your work. ### You CANNOT - Send formal notifications (only PMs can) diff --git a/agents/blueprints/ux_ui/ux-documenter.md b/agents/blueprints/ux_ui/ux-documenter.md index ff7d6ae8..64bd769f 100644 --- a/agents/blueprints/ux_ui/ux-documenter.md +++ b/agents/blueprints/ux_ui/ux-documenter.md @@ -121,6 +121,22 @@ and verify all subtasks are done before calling `roboco_task_complete()`. `roboco_task_scan()` or `roboco_agent_idle()` ``` +## Communication Rules + +### When to Post in Session (DO) +- **Questions about design decisions** - Need designer/QA clarification +- **Missing context** - Design notes don't explain rationale +- **Documentation decisions** - Multiple ways to document, need guidance + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting docs on X" → Orchestrator knows, task status tracks this +- ❌ "Writing in progress" → Use `roboco_task_progress()` instead +- ❌ "Docs complete" → Use `roboco_doc_complete()` instead +- ❌ Internal notes → Use `roboco_journal_*()` instead + +**Rule of thumb:** Only post if you need a response from designer/QA/PM. +The orchestrator spawns you with full context including design notes and QA results. + ## Capabilities ```yaml diff --git a/agents/blueprints/ux_ui/ux-pm.md b/agents/blueprints/ux_ui/ux-pm.md index 3e5cd771..4f38c5ab 100644 --- a/agents/blueprints/ux_ui/ux-pm.md +++ b/agents/blueprints/ux_ui/ux-pm.md @@ -46,6 +46,12 @@ You interact with RoboCo systems through MCP tools: - `roboco_task_assign(task_id, agent_slug)` - Assign task to an agent - `roboco_task_complete(task_id)` - Complete a parent task after subtasks done +**Session Management (Work Sessions for Tasks):** +- `roboco_session_create_for_tasks(data)` - Create a work session linked to tasks +- `roboco_session_link_task(data)` - Link additional task to existing session +- `roboco_session_unlink_task(session_id, task_id)` - Remove task from session +- `roboco_session_get_for_task(task_id)` - Get sessions linked to a task + **Journal (Document Your Thinking):** - `roboco_journal_entry(data)` - General journal entry - `roboco_journal_reflect(data)` - Task reflection @@ -149,6 +155,42 @@ roboco_task_assign("{task_id}", "ux-dev") - Every subtask MUST have both `parent_task_id` AND `assigned_to` - Do NOT keep tasks for yourself - delegate to designers! +### 7.5. CREATE WORK SESSION (REQUIRED) +**Tool:** `roboco_session_create_for_tasks(data)` + +After delegating, you MUST create a work session for the task: +```python +roboco_session_create_for_tasks({ + "task_ids": ["task-uuid-1"], + "channel_slug": "uxui-cell", + "scope": "cell", # Cell-level session + "relationship_type": "discussion" +}) +``` + +**Session scopes:** +- `initiative` - Cross-cell coordination (Main PM only, #dev-all) +- `cell` - Cell-specific work (your default, #uxui-cell) +- `task` - Individual task execution (developer level) + +**Why sessions are mandatory:** +- Every task needs a discussion context +- QA and documenter see design context +- Frontend can review design discussion history + +### 7.6. ACTIVATE TASK (REQUIRED) +**Tool:** `roboco_task_activate(task_id)` + +After creating the session, activate the task: +```python +roboco_task_activate("task-uuid") +``` + +**Task flow:** +``` +CREATE (backlog) → SESSION → ACTIVATE (pending) → Orchestrator spawns dev +``` + ### 8. COMMUNICATE **Tool:** `roboco_message_send(data)` Tell the team what you did: @@ -324,7 +366,12 @@ tools: # Task Management - roboco_task_scan, roboco_task_get, roboco_task_claim - roboco_task_start, roboco_task_plan, roboco_task_progress - - roboco_task_create, roboco_task_assign, roboco_task_complete + - roboco_task_create, roboco_task_assign, roboco_task_activate + - roboco_task_complete + + # Session Management (REQUIRED before activation) + - roboco_session_create_for_tasks, roboco_session_link_task + - roboco_session_unlink_task, roboco_session_get_for_task # Journal - roboco_journal_entry, roboco_journal_decision diff --git a/agents/blueprints/ux_ui/ux-qa.md b/agents/blueprints/ux_ui/ux-qa.md index 27b97464..b1658082 100644 --- a/agents/blueprints/ux_ui/ux-qa.md +++ b/agents/blueprints/ux_ui/ux-qa.md @@ -125,6 +125,25 @@ If dev_notes is empty or no Figma link provided, that's a valid FAIL reason. `roboco_task_scan()` or `roboco_agent_idle()` ``` +## Communication Rules + +### When to Post in Session (DO) +- **Questions about design intent** - Need designer clarification +- **Critical issues** - Accessibility failures, missing states +- **Decisions needing input** - Edge cases with unclear expected behavior +- **Cross-cell patterns** - Issues you're seeing across cells + +### When NOT to Post (USE OTHER TOOLS) +- ❌ "Starting QA on X" → Orchestrator knows, task status tracks this +- ❌ "Reviewing in progress" → Use `roboco_task_progress()` instead +- ❌ "Completed QA" → Use `roboco_qa_pass()`/`roboco_qa_fail()` instead +- ❌ Internal review notes → Use `roboco_journal_*()` instead +- ❌ Minor feedback → Put in QA verdict notes, not session chat + +**Rule of thumb:** Only post if you need a response from designer/PM, or if +the issue affects other tasks. The orchestrator spawns you with full +context including designer's handoff notes. + ## Capabilities ```yaml diff --git a/alembic/versions/001_initial_schema.py b/alembic/versions/001_initial_schema.py index ea85b254..b6f0b50d 100644 --- a/alembic/versions/001_initial_schema.py +++ b/alembic/versions/001_initial_schema.py @@ -10,6 +10,7 @@ Creates all tables for the RoboCo AI Agents Company system: - channels: Communication channels - groups: Role-based groups within channels - sessions: Bounded message sessions +- session_tasks: Many-to-many session-task links (PM work sessions) - messages: Extracted messages - notifications: Formal notifications - journals: Agent personal logs @@ -292,6 +293,13 @@ def upgrade() -> None: server_default="active", index=True, ), + sa.Column( + "scope", + sa.Enum("initiative", "cell", "task", name="sessionscope"), + nullable=False, + server_default="task", + index=True, + ), sa.Column( "started_at", sa.DateTime(), nullable=False, server_default=sa.func.now() ), @@ -309,6 +317,54 @@ def upgrade() -> None: ), ) + # ========================================================================== + # SESSION_TASKS TABLE (Many-to-Many Junction) + # ========================================================================== + op.create_table( + "session_tasks", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "session_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("sessions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column( + "task_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("tasks.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("is_primary", sa.Boolean(), nullable=False, server_default="false"), + sa.Column( + "relationship_type", + sa.String(50), + nullable=False, + server_default="discussion", + ), + sa.Column( + "added_at", sa.DateTime(), nullable=False, server_default=sa.func.now() + ), + sa.Column( + "added_by", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("agents.id", ondelete="SET NULL"), + nullable=True, + ), + sa.UniqueConstraint("session_id", "task_id", name="uq_session_task"), + ) + + # Partial unique index: only one primary session per task + op.execute( + """ + CREATE UNIQUE INDEX ix_session_tasks_primary_per_task + ON session_tasks (task_id) + WHERE is_primary = true + """ + ) + # ========================================================================== # MESSAGES TABLE # ========================================================================== @@ -651,6 +707,8 @@ def downgrade() -> None: op.drop_table("journals") op.drop_table("notifications") op.drop_table("messages") + op.drop_index("ix_session_tasks_primary_per_task", table_name="session_tasks") + op.drop_table("session_tasks") op.drop_table("sessions") op.drop_table("groups") op.drop_table("channels") @@ -668,6 +726,7 @@ def downgrade() -> None: op.execute("DROP TYPE IF EXISTS notificationtype") op.execute("DROP TYPE IF EXISTS messagetype") op.execute("DROP TYPE IF EXISTS sessionstatus") + op.execute("DROP TYPE IF EXISTS sessionscope") op.execute("DROP TYPE IF EXISTS channeltype") op.execute("DROP TYPE IF EXISTS complexity") op.execute("DROP TYPE IF EXISTS taskstatus") diff --git a/roboco/api/routes/sessions.py b/roboco/api/routes/sessions.py index f0552aed..5bb69bbd 100644 --- a/roboco/api/routes/sessions.py +++ b/roboco/api/routes/sessions.py @@ -17,12 +17,27 @@ from roboco.api.deps import CurrentAgentId, DbSession from roboco.api.schemas.sessions import ( ListSessionsParams, SessionCreateRequest, + SessionForTasksCreateRequest, SessionListResponse, SessionResponse, + SessionTaskLinkRequest, + SessionTaskLinkResponse, + SessionTaskLinksResponse, +) +from roboco.db.tables import ( + ChannelTable, + GroupTable, + SessionTable, + SessionTaskTable, ) -from roboco.db.tables import GroupTable, SessionTable from roboco.models import SessionStatus -from roboco.services.permissions import has_privileged_access +from roboco.models.session import ( + SessionForTasksCreate, + SessionTaskRelationshipType, +) +from roboco.services import ConflictError, NotFoundError +from roboco.services.messaging import get_messaging_service +from roboco.services.permissions import has_privileged_access, is_pm_role from roboco.utils.converters import require_uuid router = APIRouter() @@ -88,6 +103,7 @@ async def list_sessions( id=require_uuid(s.id), group_id=require_uuid(s.group_id), status=s.status, + scope=s.scope, message_count=s.message_count, total_content_length=s.total_content_length, started_at=s.started_at, @@ -134,6 +150,7 @@ async def get_session( id=require_uuid(session.id), group_id=require_uuid(session.group_id), status=session.status, + scope=session.scope, message_count=session.message_count, total_content_length=session.total_content_length, started_at=session.started_at, @@ -219,6 +236,7 @@ async def create_session( id=require_uuid(session.id), group_id=require_uuid(session.group_id), status=session.status, + scope=session.scope, message_count=session.message_count, total_content_length=session.total_content_length, started_at=session.started_at, @@ -271,9 +289,230 @@ async def close_session( id=require_uuid(session.id), group_id=require_uuid(session.group_id), status=session.status, + scope=session.scope, message_count=session.message_count, total_content_length=session.total_content_length, started_at=session.started_at, last_activity_at=session.last_activity_at, closed_at=session.closed_at, ) + + +# ============================================================================= +# SESSION-TASK ROUTES +# ============================================================================= + + +def _link_to_response(link: SessionTaskTable) -> SessionTaskLinkResponse: + """Convert SessionTaskTable to response model.""" + return SessionTaskLinkResponse( + id=require_uuid(link.id), + session_id=require_uuid(link.session_id), + task_id=require_uuid(link.task_id), + is_primary=link.is_primary, + relationship_type=link.relationship_type, + added_at=link.added_at, + added_by=require_uuid(link.added_by) if link.added_by else None, + ) + + +@router.get( + "/for-task/{task_id}", + response_model=list[SessionTaskLinkResponse], + summary="Get sessions for task", + description="Get all sessions linked to a specific task.", +) +async def get_sessions_for_task( + task_id: UUID, + db: DbSession, +) -> list[SessionTaskLinkResponse]: + """Get sessions linked to a task. + + Returns session links with session_id, is_primary, and relationship_type. + Any agent assigned to the task can access this. + """ + messaging = get_messaging_service(db) + links = await messaging.get_sessions_for_task(task_id) + return [_link_to_response(link) for link in links] + + +@router.post( + "/for-tasks", + response_model=SessionTaskLinksResponse, + status_code=status.HTTP_201_CREATED, + summary="Create session for tasks", + description="Create a work session linked to one or more tasks (PM only).", +) +async def create_session_for_tasks( + db: DbSession, + agent_id: CurrentAgentId, + data: SessionForTasksCreateRequest, +) -> SessionTaskLinksResponse: + """Create a session linked to tasks (PM only).""" + # Verify PM permission + if not await is_pm_role(db, agent_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only PMs can create task-linked sessions", + ) + + # Verify channel exists + channel_result = await db.execute( + select(ChannelTable).where(ChannelTable.slug == data.channel_slug) + ) + channel = channel_result.scalar_one_or_none() + if not channel: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Channel '{data.channel_slug}' not found", + ) + + # Create session with links using service + messaging = get_messaging_service(db) + + try: + rel_type = SessionTaskRelationshipType(data.relationship_type) + except ValueError: + rel_type = SessionTaskRelationshipType.DISCUSSION + + req = SessionForTasksCreate( + task_ids=data.task_ids, + channel_slug=data.channel_slug, + relationship_type=rel_type, + ) + + try: + session, links = await messaging.create_session_for_tasks(req, agent_id) + except NotFoundError as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) from e + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + session_response = SessionResponse( + id=require_uuid(session.id), + group_id=require_uuid(session.group_id), + status=session.status, + scope=session.scope, + message_count=session.message_count, + total_content_length=session.total_content_length, + started_at=session.started_at, + last_activity_at=session.last_activity_at, + closed_at=session.closed_at, + ) + + return SessionTaskLinksResponse( + session=session_response, + links=[_link_to_response(link) for link in links], + ) + + +@router.post( + "/{session_id}/tasks", + response_model=SessionTaskLinkResponse, + status_code=status.HTTP_201_CREATED, + summary="Link task to session", + description="Link a task to an existing session (PM only).", +) +async def link_task_to_session( + db: DbSession, + agent_id: CurrentAgentId, + session_id: UUID, + data: SessionTaskLinkRequest, +) -> SessionTaskLinkResponse: + """Link a task to a session (PM only).""" + if not await is_pm_role(db, agent_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only PMs can link tasks to sessions", + ) + + messaging = get_messaging_service(db) + + try: + rel_type = SessionTaskRelationshipType(data.relationship_type) + except ValueError: + rel_type = SessionTaskRelationshipType.DISCUSSION + + try: + link = await messaging.link_session_to_task( + session_id=session_id, + task_id=data.task_id, + added_by=agent_id, + is_primary=data.is_primary, + relationship_type=rel_type, + ) + except NotFoundError as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) from e + except ConflictError as e: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(e), + ) from e + + return _link_to_response(link) + + +@router.delete( + "/{session_id}/tasks/{task_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Unlink task from session", + description="Remove a task from a session (PM only).", +) +async def unlink_task_from_session( + db: DbSession, + agent_id: CurrentAgentId, + session_id: UUID, + task_id: UUID, +) -> None: + """Unlink a task from a session (PM only).""" + if not await is_pm_role(db, agent_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only PMs can unlink tasks from sessions", + ) + + messaging = get_messaging_service(db) + removed = await messaging.unlink_session_from_task(session_id, task_id) + + if not removed: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Session-task link not found", + ) + + +@router.get( + "/{session_id}/tasks", + response_model=list[SessionTaskLinkResponse], + summary="Get tasks for session", + description="Get all tasks linked to a session.", +) +async def get_tasks_for_session( + db: DbSession, + _agent_id: CurrentAgentId, + session_id: UUID, +) -> list[SessionTaskLinkResponse]: + """Get all tasks linked to a session.""" + # Verify session exists + session_result = await db.execute( + select(SessionTable).where(SessionTable.id == session_id) + ) + if not session_result.scalar_one_or_none(): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Session not found", + ) + + messaging = get_messaging_service(db) + links = await messaging.get_tasks_for_session(session_id) + + return [_link_to_response(link) for link in links] diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 6e002101..f0fdb574 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -15,6 +15,10 @@ from roboco.api.deps import ( DbSession, PermissionServiceDep, ) +from roboco.api.schemas.sessions import ( + SessionTaskLinkResponse, + TaskSessionsResponse, +) from roboco.api.schemas.tasks import ( CheckpointRequest, ClaimRequest, @@ -25,6 +29,7 @@ from roboco.api.schemas.tasks import ( SoftBlockRequest, TaskCountResponse, TaskResponse, + TaskSessionLinkResponse, TaskUpdate, TeamTasksQuery, task_list_to_response, @@ -35,12 +40,14 @@ from roboco.db.tables import AgentTable from roboco.models.base import TaskStatus, Team from roboco.models.task import TaskCreate from roboco.services.audit import get_audit_service +from roboco.services.messaging import get_messaging_service from roboco.services.permissions import TaskAction from roboco.services.task import ( TaskCreateRequest, extract_original_developer, get_task_service, ) +from roboco.utils.converters import require_uuid router = APIRouter() @@ -283,14 +290,33 @@ async def get_task( task_id: UUID, db: DbSession, ) -> TaskResponse: - """Get a specific task.""" + """Get a specific task with linked sessions.""" service = get_task_service(db) task = await service.get(task_id) if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" ) - return task_to_response(task) + + # Get linked sessions for this task + messaging = get_messaging_service(db) + session_links = await messaging.get_sessions_for_task(task_id) + + # Build response with sessions + response = task_to_response(task) + response.sessions = [ + TaskSessionLinkResponse( + session_id=require_uuid(link.session_id), + channel_slug=link.session.group.channel.slug, + scope=link.session.scope, + is_primary=link.is_primary, + relationship_type=link.relationship_type, + ) + for link in session_links + if link.session and link.session.group and link.session.group.channel + ] + + return response @router.put("/{task_id}", response_model=TaskResponse) @@ -396,9 +422,7 @@ async def get_subtasks( # ============================================================================= -async def _resolve_claim_agent_id( - db: DbSession, agent_id_str: str -) -> UUID: +async def _resolve_claim_agent_id(db: DbSession, agent_id_str: str) -> UUID: """Resolve agent ID from UUID string or slug.""" try: return UUID(agent_id_str) @@ -1068,3 +1092,92 @@ async def add_commit( ) await db.commit() return task_to_response(task) + + +# ============================================================================= +# TASK ACTIVATION (PM ONLY) +# ============================================================================= + + +@router.post("/{task_id}/activate", response_model=TaskResponse) +async def activate_task( + task_id: UUID, + db: DbSession, + agent: CurrentAgentContext, + permissions: PermissionServiceDep, +) -> TaskResponse: + """ + Activate a task from BACKLOG to PENDING status (PM only). + + This is the final step in PM setup. After creating a session and + linking the task, the PM activates it to make it ready for work. + + REQUIRES: Task must have at least one linked session. + """ + # Check PM permission + if not permissions.can_perform_task_action(agent, "create_tasks"): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only PMs and management can activate tasks", + ) + + service = get_task_service(db) + + try: + task = await service.activate(task_id) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + await db.commit() + return task_to_response(task) + + +# ============================================================================= +# SESSION-TASK ENDPOINTS +# ============================================================================= + + +@router.get("/{task_id}/sessions", response_model=TaskSessionsResponse) +async def get_sessions_for_task( + task_id: UUID, + db: DbSession, + _agent: CurrentAgentContext, # Kept for auth dependency +) -> TaskSessionsResponse: + """Get all sessions linked to a task.""" + # Verify task exists + service = get_task_service(db) + task = await service.get(task_id) + if not task: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Task not found", + ) + + # Get sessions + messaging = get_messaging_service(db) + links = await messaging.get_sessions_for_task(task_id) + + # Find primary session + primary_link = next((link for link in links if link.is_primary), None) + + return TaskSessionsResponse( + task_id=task_id, + sessions=[ + SessionTaskLinkResponse( + id=require_uuid(link.id), + session_id=require_uuid(link.session_id), + task_id=require_uuid(link.task_id), + is_primary=link.is_primary, + relationship_type=link.relationship_type, + added_at=link.added_at, + added_by=require_uuid(link.added_by) if link.added_by else None, + ) + for link in links + ], + primary_session_id=( + require_uuid(primary_link.session_id) if primary_link else None + ), + ) diff --git a/roboco/api/schemas/sessions.py b/roboco/api/schemas/sessions.py index 1c381f7c..93f68382 100644 --- a/roboco/api/schemas/sessions.py +++ b/roboco/api/schemas/sessions.py @@ -10,6 +10,7 @@ from uuid import UUID from pydantic import BaseModel, Field from roboco.models import SessionStatus +from roboco.models.session import SessionScope class ListSessionsParams(BaseModel): @@ -26,6 +27,7 @@ class SessionResponse(BaseModel): id: UUID group_id: UUID status: SessionStatus + scope: SessionScope message_count: int total_content_length: int started_at: datetime @@ -48,3 +50,65 @@ class SessionCreateRequest(BaseModel): max_message_count: int | None = 100 max_content_length: int | None = 50000 timeout_seconds: int = 300 + + +# ============================================================================= +# SESSION-TASK SCHEMAS +# ============================================================================= + + +class SessionForTasksCreateRequest(BaseModel): + """Request to create a session linked to tasks (PM only).""" + + task_ids: list[UUID] = Field(..., min_length=1) + channel_slug: str + scope: SessionScope = Field( + default=SessionScope.CELL, + description="Scope level: initiative (Main PM), cell (Cell PM), task (dev)", + ) + relationship_type: str = Field( + default="discussion", + description="Type: discussion, planning, review, retrospective", + ) + max_time_window_minutes: int | None = 30 + max_message_count: int | None = 100 + max_content_length: int | None = 50000 + timeout_seconds: int = 300 + + +class SessionTaskLinkRequest(BaseModel): + """Request to link a task to a session.""" + + task_id: UUID + is_primary: bool = False + relationship_type: str = Field( + default="discussion", + description="Type: discussion, planning, review, retrospective", + ) + + +class SessionTaskLinkResponse(BaseModel): + """Response for a session-task link.""" + + id: UUID + session_id: UUID + task_id: UUID + is_primary: bool + relationship_type: str + added_at: datetime + added_by: UUID | None + + +class SessionTaskLinksResponse(BaseModel): + """Response containing session with its task links.""" + + session: SessionResponse + links: list[SessionTaskLinkResponse] + + +class TaskSessionsResponse(BaseModel): + """Response containing sessions linked to a task.""" + + task_id: UUID + sessions: list[SessionTaskLinkResponse] + primary_session_id: UUID | None diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index 072a0e7e..154d3e58 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -11,6 +11,7 @@ from uuid import UUID from pydantic import BaseModel, Field from roboco.models.base import Complexity, TaskStatus, Team +from roboco.models.session import SessionScope from roboco.utils.converters import require_uuid, to_python_uuid, to_python_uuid_list if TYPE_CHECKING: @@ -50,6 +51,16 @@ class CommitRefResponse(BaseModel): author_agent_id: UUID | None = None +class TaskSessionLinkResponse(BaseModel): + """A session linked to this task.""" + + session_id: UUID + channel_slug: str + scope: SessionScope + is_primary: bool + relationship_type: str + + class SubTaskResponse(BaseModel): """A sub-task within a task plan.""" @@ -230,6 +241,9 @@ class TaskResponse(BaseModel): self_verified: bool qa_verified: bool | None + # Linked Sessions (for agent context) + sessions: list[TaskSessionLinkResponse] = [] + class Config: from_attributes = True diff --git a/roboco/db/tables.py b/roboco/db/tables.py index 4ac44c49..356cef10 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -41,6 +41,7 @@ from roboco.models.base import ( TaskStatus, Team, ) +from roboco.models.session import SessionScope # ============================================================================= # AGENT TABLE @@ -211,6 +212,10 @@ class TaskTable(Base): parent_task: Mapped["TaskTable | None"] = relationship( "TaskTable", remote_side=[id], lazy="select" ) + # Session links (many-to-many via SessionTaskTable) + session_links: Mapped[list["SessionTaskTable"]] = relationship( + "SessionTaskTable", back_populates="task", lazy="select" + ) __table_args__ = ( # Composite indexes for common queries @@ -390,6 +395,11 @@ class SessionTable(Base): Enum(SessionStatus), nullable=False, default=SessionStatus.ACTIVE, index=True ) + # Scope (for smart context loading) + scope: Mapped[SessionScope] = mapped_column( + Enum(SessionScope), nullable=False, default=SessionScope.TASK, index=True + ) + # Timestamps started_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=datetime.now(UTC), nullable=False @@ -415,6 +425,10 @@ class SessionTable(Base): messages: Mapped[list["MessageTable"]] = relationship( "MessageTable", back_populates="session", lazy="select" ) + # Task links (many-to-many via SessionTaskTable) + task_links: Mapped[list["SessionTaskTable"]] = relationship( + "SessionTaskTable", back_populates="session", lazy="select" + ) __table_args__ = ( # Composite indexes for common queries @@ -423,6 +437,85 @@ class SessionTable(Base): ) +# ============================================================================= +# SESSION-TASK JUNCTION TABLE +# ============================================================================= + + +class SessionTaskTable(Base): + """ + Junction table for many-to-many Session ↔ Task relationship. + + Enables PMs to create work sessions as discussion contexts for tasks. + A task can have multiple sessions (planning, review, retrospective). + A session can discuss multiple related tasks. + """ + + __tablename__ = "session_tasks" + + # Identity + id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid4 + ) + + # Foreign Keys (indexes defined in __table_args__) + session_id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("sessions.id", ondelete="CASCADE"), + nullable=False, + ) + task_id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="CASCADE"), + nullable=False, + ) + + # Relationship Metadata + is_primary: Mapped[bool] = mapped_column( + Boolean, default=False, nullable=False + ) # Primary discussion session for this task + relationship_type: Mapped[str] = mapped_column( + String(50), default="discussion", nullable=False + ) # "discussion", "planning", "review", "retrospective" + + # Audit + added_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=datetime.now(UTC), nullable=False + ) + added_by: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("agents.id", ondelete="SET NULL"), + nullable=True, # Allow NULL if PM is deleted + ) + + # Relationships + session: Mapped["SessionTable"] = relationship( + "SessionTable", back_populates="task_links", lazy="joined" + ) + task: Mapped["TaskTable"] = relationship( + "TaskTable", back_populates="session_links", lazy="joined" + ) + added_by_agent: Mapped["AgentTable | None"] = relationship( + "AgentTable", lazy="select" + ) + + __table_args__ = ( + # Each session-task pair is unique + UniqueConstraint("session_id", "task_id", name="uq_session_task"), + # Partial unique index: only one primary session per task + Index( + "ix_session_tasks_primary_per_task", + "task_id", + unique=True, + postgresql_where=(is_primary == True), # noqa: E712 + ), + # Fast lookups + Index("ix_session_tasks_task_id", "task_id"), + Index("ix_session_tasks_session_id", "session_id"), + Index("ix_session_tasks_type", "relationship_type"), + ) + + # ============================================================================= # MESSAGE TABLE # ============================================================================= diff --git a/roboco/mcp/message_server.py b/roboco/mcp/message_server.py index 002b2f9b..cf214efa 100644 --- a/roboco/mcp/message_server.py +++ b/roboco/mcp/message_server.py @@ -304,27 +304,61 @@ async def _handle_channel_history( } +async def _get_task_primary_session(client: ApiClient, task_id: str) -> str | None: + """Get the primary session ID for a task, if one exists.""" + resp = await client.get(f"/sessions/for-task/{task_id}") + if not resp.ok: + return None + + sessions = resp.json() + if not sessions: + return None + + # Find primary session + for session in sessions: + if session.get("is_primary"): + return str(session.get("session_id")) + + # Fall back to first session if no primary marked + return str(sessions[0].get("session_id")) if sessions else None + + async def _handle_message_send( client: ApiClient, agent_id: str, data: SendMessageInput, ) -> dict[str, Any]: - """Handle message sending.""" + """Handle message sending. + + If task_id is provided, routes to that task's primary session. + Otherwise, routes to the channel's current active session. + """ if validation_error := _validate_message_send( agent_id, data.channel_slug, data.content, data.message_type ): return validation_error - # Get channel by slug - channel_result = await _get_channel_by_slug(client, data.channel_slug) - if isinstance(channel_result, dict): - return channel_result # Error response - channel_id = channel_result + session_id: str | None = None + routed_to_task_session = False - session_result = await _get_or_create_session(client, channel_id) - if isinstance(session_result, dict): - return session_result - session_id = session_result + # If task_id provided, try to route to task's primary session + if data.task_id: + session_id = await _get_task_primary_session(client, data.task_id) + if session_id: + routed_to_task_session = True + + # Fall back to channel's active session + if not session_id: + # Get channel by slug + channel_result = await _get_channel_by_slug(client, data.channel_slug) + if isinstance(channel_result, dict): + return channel_result # Error response + channel_id = channel_result + + session_result = await _get_or_create_session(client, channel_id) + if isinstance(session_result, dict): + return session_result + session_id = session_result # Resolve mentions (slugs) to UUIDs using shared cache resolved_mentions: list[str] = [] @@ -352,11 +386,16 @@ async def _handle_message_send( "SEND_FAILED", "Failed to send message", {"api_error": resp.text} ) + guidance = "Message sent successfully." + if routed_to_task_session: + guidance = f"Message sent to task {data.task_id}'s session." + return { "status": "sent", "message": resp.json(), "channel": data.channel_slug, - "guidance": "Message sent successfully.", + "routed_to_task_session": routed_to_task_session, + "guidance": guidance, } @@ -529,6 +568,58 @@ def create_message_mcp_server(agent_id: str) -> FastMCP: ) return await _handle_report_blocker(client, agent_id, data) + @mcp.tool() + async def roboco_session_history_for_task( + task_id: str, + limit: int = 50, + ) -> dict[str, Any]: + """ + Get message history from your task's work session. + + Use this to see the discussion context for a task you're working on. + Returns messages from the task's primary session. + + Args: + task_id: The task ID to get session history for + limit: Maximum number of messages to return (default 50) + """ + # Get task's primary session + session_id = await _get_task_primary_session(client, task_id) + if not session_id: + return { + "error": "NO_SESSION", + "message": f"Task {task_id} has no linked session.", + "guidance": ( + "This task doesn't have a work session yet. " + "The PM should create one before work begins." + ), + } + + # Get messages from the session + resp = await client.get( + "/messages", + params={"session_id": session_id, "limit": limit}, + ) + + if not resp.ok: + return format_error_response( + "FETCH_FAILED", + "Failed to fetch session history", + {"api_error": resp.text}, + ) + + messages = resp.json() + return { + "task_id": task_id, + "session_id": session_id, + "message_count": len(messages), + "messages": messages, + "guidance": ( + "This is the discussion history for your task. " + "Use roboco_message_send with task_id to add to it." + ), + } + return mcp diff --git a/roboco/mcp/schemas/__init__.py b/roboco/mcp/schemas/__init__.py index dc63b507..d03240f3 100644 --- a/roboco/mcp/schemas/__init__.py +++ b/roboco/mcp/schemas/__init__.py @@ -216,3 +216,39 @@ class TaskPauseInput(BaseModel): remaining_work: list[str] = Field( default_factory=list, description="List of remaining sub-tasks" ) + + +# ============================================================================= +# SESSION-TASK SCHEMAS (PM Tools) +# ============================================================================= + + +class SessionCreateForTasksInput(BaseModel): + """Input for creating a session linked to tasks (PM only).""" + + task_ids: list[str] = Field( + ..., min_length=1, description="Task IDs to link to the session" + ) + channel_slug: str = Field(..., description="Channel where session is created") + scope: str = Field( + default="cell", + description="Scope level: initiative (Main PM), cell (Cell PM), task (dev)", + ) + relationship_type: str = Field( + default="discussion", + description="Type: discussion, planning, review, retrospective", + ) + + +class SessionLinkTaskInput(BaseModel): + """Input for linking a session to a task (PM only).""" + + session_id: str = Field(..., description="Session ID to link") + task_id: str = Field(..., description="Task ID to link") + is_primary: bool = Field( + default=False, description="Mark as primary session for this task" + ) + relationship_type: str = Field( + default="discussion", + description="Type: discussion, planning, review, retrospective", + ) diff --git a/roboco/mcp/task_server.py b/roboco/mcp/task_server.py index d06f7525..3093e215 100644 --- a/roboco/mcp/task_server.py +++ b/roboco/mcp/task_server.py @@ -23,6 +23,10 @@ Tools: - roboco_task_assign: Assign task to agent (PM only) - roboco_task_cancel: Cancel a task (PM/Board only) - roboco_task_escalate: Escalate task up hierarchy (all agents) +- roboco_session_create_for_tasks: Create work session for tasks (PM only) +- roboco_session_link_task: Link session to task (PM only) +- roboco_session_unlink_task: Unlink session from task (PM only) +- roboco_session_get_for_task: Get sessions for a task (all agents) """ from typing import Any @@ -30,6 +34,8 @@ from typing import Any from mcp.server.fastmcp import FastMCP from roboco.mcp.schemas import ( + SessionCreateForTasksInput, + SessionLinkTaskInput, TaskAssignInput, TaskBlockInput, TaskCreateInput, @@ -39,6 +45,11 @@ from roboco.mcp.schemas import ( from roboco.mcp.tasks.handlers import ( handle_agent_idle, handle_docs_complete, + handle_session_create_for_tasks, + handle_session_get_for_task, + handle_session_link_task, + handle_session_unlink_task, + handle_task_activate, handle_task_assign, handle_task_block, handle_task_cancel, @@ -61,7 +72,7 @@ from roboco.mcp.tasks.handlers import ( from roboco.mcp.utils import ApiClient -def create_task_mcp_server(agent_id: str) -> FastMCP: +def create_task_mcp_server(agent_id: str) -> FastMCP: # noqa: PLR0915 """ Create a Task MCP server for a specific agent. @@ -522,6 +533,132 @@ def create_task_mcp_server(agent_id: str) -> FastMCP: ) return await handle_task_escalate(client, input_data, agent_id) + # ========================================================================= + # PM SESSION TOOLS + # ========================================================================= + + @mcp.tool() + async def roboco_session_create_for_tasks( + data: SessionCreateForTasksInput, + ) -> dict[str, Any]: + """ + Create a work session linked to one or more tasks (PM only). + + Use this to: + - Create a discussion context for a task or set of related tasks + - Enable assigned agents to communicate about the work + - Set up planning/review sessions for complex tasks + + SCOPE LEVELS: + - "initiative": Cross-cell sessions in #dev-all (Main PM only) + - "cell": Cell-specific sessions in team channel (Cell PM default) + - "task": Individual task execution (Developer level) + + ENFORCEMENT: + - Only PMs and management can create task-linked sessions + - Cell PMs can only create sessions in their team's channel + + Args: + data: SessionCreateForTasksInput with task_ids, channel_slug, + scope (initiative/cell/task), and relationship_type + + Returns: + Created session with task links + """ + return await handle_session_create_for_tasks(client, data, agent_id) + + @mcp.tool() + async def roboco_session_link_task( + data: SessionLinkTaskInput, + ) -> dict[str, Any]: + """ + Link an existing session to a task (PM only). + + Use this to: + - Add additional tasks to an existing session + - Link related tasks to the same discussion context + - Mark a session as primary for a specific task + + ENFORCEMENT: + - Only PMs and management can link sessions to tasks + - One primary session per task (use is_primary carefully) + + Args: + data: SessionLinkTaskInput with session_id, task_id, + optional is_primary and relationship_type + + Returns: + Created link confirmation + """ + return await handle_session_link_task(client, data, agent_id) + + @mcp.tool() + async def roboco_session_unlink_task( + session_id: str, + task_id: str, + ) -> dict[str, Any]: + """ + Remove a task from a session (PM only). + + Use this to: + - Remove tasks that are no longer relevant to the session + - Clean up session-task links after task completion + + ENFORCEMENT: + - Only PMs and management can unlink sessions from tasks + + Args: + session_id: Session ID to unlink from + task_id: Task ID to unlink + + Returns: + Unlink confirmation + """ + return await handle_session_unlink_task(client, session_id, task_id, agent_id) + + @mcp.tool() + async def roboco_session_get_for_task(task_id: str) -> dict[str, Any]: + """ + Get all sessions linked to a task. + + Use this to: + - Find the discussion context for a task you're working on + - Check if a task has a primary session + - See all related sessions (planning, review, etc.) + + Args: + task_id: Task ID to query sessions for + + Returns: + List of sessions with their relationship types + """ + return await handle_session_get_for_task(client, task_id, agent_id) + + @mcp.tool() + async def roboco_task_activate(task_id: str) -> dict[str, Any]: + """ + Activate a task from BACKLOG to PENDING status (PM only). + + This is the FINAL STEP in task setup. After creating and assigning + a task, you MUST: + 1. Create a session: roboco_session_create_for_tasks() + 2. Activate the task: roboco_task_activate() + + Only after activation will the orchestrator spawn agents to work on it. + + ENFORCEMENT: + - Only PMs and management can activate tasks + - Task must be in BACKLOG status + - Task MUST have at least one linked session + + Args: + task_id: The task UUID to activate + + Returns: + Activated task with PENDING status + """ + return await handle_task_activate(client, task_id, agent_id) + return mcp diff --git a/roboco/mcp/tasks/handlers/__init__.py b/roboco/mcp/tasks/handlers/__init__.py index fb5287e5..1af576ea 100644 --- a/roboco/mcp/tasks/handlers/__init__.py +++ b/roboco/mcp/tasks/handlers/__init__.py @@ -17,6 +17,7 @@ from roboco.mcp.tasks.handlers.lifecycle import ( handle_task_complete, ) from roboco.mcp.tasks.handlers.management import ( + handle_task_activate, handle_task_assign, handle_task_create, handle_task_escalate, @@ -28,6 +29,12 @@ from roboco.mcp.tasks.handlers.review import ( handle_task_submit_verification, ) from roboco.mcp.tasks.handlers.scan import handle_task_get, handle_task_scan +from roboco.mcp.tasks.handlers.sessions import ( + handle_session_create_for_tasks, + handle_session_get_for_task, + handle_session_link_task, + handle_session_unlink_task, +) from roboco.mcp.tasks.handlers.work import ( handle_task_plan, handle_task_progress, @@ -37,6 +44,11 @@ from roboco.mcp.tasks.handlers.work import ( __all__ = [ "handle_agent_idle", "handle_docs_complete", + "handle_session_create_for_tasks", + "handle_session_get_for_task", + "handle_session_link_task", + "handle_session_unlink_task", + "handle_task_activate", "handle_task_assign", "handle_task_block", "handle_task_cancel", diff --git a/roboco/mcp/tasks/handlers/management.py b/roboco/mcp/tasks/handlers/management.py index 61a0cc7e..0f78492e 100644 --- a/roboco/mcp/tasks/handlers/management.py +++ b/roboco/mcp/tasks/handlers/management.py @@ -132,9 +132,7 @@ def _validate_create_permissions(agent_id: str) -> dict[str, Any] | None: return None -def _validate_cell_pm_team( - agent_id: str, requested_team: str -) -> dict[str, Any] | None: +def _validate_cell_pm_team(agent_id: str, requested_team: str) -> dict[str, Any] | None: """Validate Cell PM team restrictions for task creation. Returns error or None.""" role = get_agent_role(agent_id) agent_team = get_agent_team(agent_id) @@ -326,3 +324,52 @@ async def handle_task_escalate( "They will be notified and can reassign or provide guidance." ) return format_task_response(task, "ESCALATED", guidance) + + +async def handle_task_activate( + client: ApiClient, task_id: str, agent_id: str +) -> dict[str, Any]: + """ + Handle task activation from BACKLOG to PENDING (PM only). + + This is the final step in PM setup. After creating a session and + linking the task, the PM activates it to make it ready for work. + The orchestrator will then spawn agents to claim and work on it. + + REQUIRES: Task must have at least one linked session. + """ + if not can_create_tasks(agent_id): + return format_error_response( + "PERMISSION_DENIED", + "Only PMs and management can activate tasks", + {"role": get_agent_role(agent_id)}, + ) + + try: + resp = await client.post(f"/tasks/{task_id}/activate") + except Exception as e: + return format_error_response( + "CONNECTION_ERROR", + f"Failed to connect to API: {type(e).__name__}", + ) + + if resp.is_status(status.HTTP_404_NOT_FOUND): + return format_error_response("NOT_FOUND", f"Task {task_id} not found") + + if resp.is_status(status.HTTP_400_BAD_REQUEST): + detail = resp.json().get("detail", "Activation failed") + return format_error_response("ACTIVATION_FAILED", detail) + + if not resp.ok: + return format_error_response( + "ACTIVATION_FAILED", + "Failed to activate task", + {"status_code": resp.status_code, "detail": resp.text}, + ) + + task = resp.json() + guidance = ( + "Task activated. Status is now PENDING. " + "Orchestrator will spawn agents to work on it." + ) + return format_task_response(task, "ACTIVATED", guidance) diff --git a/roboco/mcp/tasks/handlers/review.py b/roboco/mcp/tasks/handlers/review.py index 5b59637e..e08b4dc5 100644 --- a/roboco/mcp/tasks/handlers/review.py +++ b/roboco/mcp/tasks/handlers/review.py @@ -20,9 +20,7 @@ from roboco.services.task import extract_original_developer def _has_work_evidence(task: dict[str, Any]) -> bool: """Check if task has evidence of work done.""" return bool( - task.get("commits") - or task.get("progress_updates") - or task.get("checkpoints") + task.get("commits") or task.get("progress_updates") or task.get("checkpoints") ) @@ -186,9 +184,7 @@ async def handle_task_qa_pass( if error := await _check_self_review(task, agent_id, client): return error - pass_resp = await client.post( - f"/tasks/{task_id}/pass-qa", json={"notes": qa_notes} - ) + pass_resp = await client.post(f"/tasks/{task_id}/pass-qa", json={"notes": qa_notes}) if not pass_resp.ok: return format_error_response( "QA_FAILED", diff --git a/roboco/mcp/tasks/handlers/scan.py b/roboco/mcp/tasks/handlers/scan.py index 2ebd79fd..f6a7a9a2 100644 --- a/roboco/mcp/tasks/handlers/scan.py +++ b/roboco/mcp/tasks/handlers/scan.py @@ -27,7 +27,11 @@ async def handle_task_scan( assigned_resp = await client.get("/tasks/my") assigned_data = assigned_resp.json() if assigned_resp.ok else [] active_statuses = { - "pending", "claimed", "in_progress", "verifying", "needs_revision" + "pending", + "claimed", + "in_progress", + "verifying", + "needs_revision", } assigned_tasks = [t for t in assigned_data if t.get("status") in active_statuses] diff --git a/roboco/mcp/tasks/handlers/sessions.py b/roboco/mcp/tasks/handlers/sessions.py new file mode 100644 index 00000000..f6eac9b8 --- /dev/null +++ b/roboco/mcp/tasks/handlers/sessions.py @@ -0,0 +1,260 @@ +""" +Task MCP Server Session Handlers + +PM-specific session-task handlers for the Task MCP server. +Enables PMs to create work sessions linked to tasks. +""" + +from typing import Any + +from fastapi import status + +from roboco.agents_config import ( + can_create_tasks, + get_agent_role, + get_agent_team, +) +from roboco.mcp.schemas import SessionCreateForTasksInput, SessionLinkTaskInput +from roboco.mcp.utils import ApiClient, format_error_response + +# ============================================================================= +# HELPER FUNCTIONS +# ============================================================================= + + +def _validate_pm_permissions(agent_id: str) -> dict[str, Any] | None: + """Validate agent has PM permissions. Returns error or None.""" + if not can_create_tasks(agent_id): + return format_error_response( + "PERMISSION_DENIED", + "Only PMs and management can manage session-task links", + {"role": get_agent_role(agent_id)}, + ) + return None + + +def _validate_channel_access(agent_id: str, channel_slug: str) -> dict[str, Any] | None: + """Validate Cell PM channel restrictions. Returns error or None.""" + role = get_agent_role(agent_id) + agent_team = get_agent_team(agent_id) + + if role != "cell_pm": + return None # Main PM and board can access any channel + + # Cell PM channel restrictions + team_channels = { + "backend": ["backend-cell"], + "frontend": ["frontend-cell"], + "ux_ui": ["uxui-cell"], + } + + allowed = team_channels.get(agent_team or "", []) + if channel_slug not in allowed: + return format_error_response( + "CHANNEL_ACCESS_DENIED", + "Cell PM can only create sessions in their team channel", + {"channel": channel_slug, "allowed": allowed}, + ) + + return None + + +def _format_session_response( + session: dict[str, Any], + links: list[dict[str, Any]], + status_code: str, + guidance: str, +) -> dict[str, Any]: + """Format session response with guidance.""" + return { + "status": status_code, + "session": session, + "task_links": links, + "guidance": guidance, + } + + +# ============================================================================= +# SESSION HANDLERS +# ============================================================================= + + +async def handle_session_create_for_tasks( + client: ApiClient, + input_data: SessionCreateForTasksInput, + agent_id: str, +) -> dict[str, Any]: + """Handle session creation linked to tasks (PM only).""" + if error := _validate_pm_permissions(agent_id): + return error + + if error := _validate_channel_access(agent_id, input_data.channel_slug): + return error + + payload = { + "task_ids": input_data.task_ids, + "channel_slug": input_data.channel_slug, + "scope": input_data.scope, + "relationship_type": input_data.relationship_type, + } + + try: + resp = await client.post("/sessions/for-tasks", json=payload) + except Exception as e: + return format_error_response( + "CONNECTION_ERROR", + f"Failed to connect to API: {type(e).__name__}", + ) + + if not resp.is_status(status.HTTP_201_CREATED): + return format_error_response( + "CREATE_FAILED", + "Failed to create session for tasks", + {"status_code": resp.status_code, "detail": resp.text}, + ) + + data = resp.json() + session = data.get("session", {}) + links = data.get("links", []) + + guidance = ( + f"Work session created. Session ID: {session.get('id', 'unknown')}. " + f"Linked to {len(links)} task(s). " + f"First task is marked as primary. " + "Assigned agents can now discuss in this session." + ) + + return _format_session_response(session, links, "CREATED", guidance) + + +async def handle_session_link_task( + client: ApiClient, + input_data: SessionLinkTaskInput, + agent_id: str, +) -> dict[str, Any]: + """Handle linking a session to a task (PM only).""" + if error := _validate_pm_permissions(agent_id): + return error + + payload = { + "task_id": input_data.task_id, + "is_primary": input_data.is_primary, + "relationship_type": input_data.relationship_type, + } + + try: + resp = await client.post( + f"/sessions/{input_data.session_id}/tasks", json=payload + ) + except Exception as e: + return format_error_response( + "CONNECTION_ERROR", + f"Failed to connect to API: {type(e).__name__}", + ) + + if resp.is_status(status.HTTP_409_CONFLICT): + return format_error_response( + "ALREADY_LINKED", + "Session is already linked to this task", + {"session_id": input_data.session_id, "task_id": input_data.task_id}, + ) + + if not resp.ok: + return format_error_response( + "LINK_FAILED", + "Failed to link session to task", + {"status_code": resp.status_code, "detail": resp.text}, + ) + + link = resp.json() + primary_note = " (marked as primary)" if input_data.is_primary else "" + + return { + "status": "LINKED", + "link": link, + "guidance": ( + f"Session linked to task{primary_note}. " + "Task's assigned agent can now access this session." + ), + } + + +async def handle_session_unlink_task( + client: ApiClient, + session_id: str, + task_id: str, + agent_id: str, +) -> dict[str, Any]: + """Handle unlinking a session from a task (PM only).""" + if error := _validate_pm_permissions(agent_id): + return error + + try: + resp = await client.delete(f"/sessions/{session_id}/tasks/{task_id}") + except Exception as e: + return format_error_response( + "CONNECTION_ERROR", + f"Failed to connect to API: {type(e).__name__}", + ) + + if resp.is_status(status.HTTP_404_NOT_FOUND): + return format_error_response( + "NOT_FOUND", + "Session-task link not found", + {"session_id": session_id, "task_id": task_id}, + ) + + if not resp.ok: + return format_error_response( + "UNLINK_FAILED", + "Failed to unlink session from task", + {"status_code": resp.status_code, "detail": resp.text}, + ) + + return { + "status": "UNLINKED", + "guidance": "Session unlinked from task. Task agent no longer has access.", + } + + +async def handle_session_get_for_task( + client: ApiClient, + task_id: str, + _agent_id: str, # Kept for handler signature consistency +) -> dict[str, Any]: + """Handle getting sessions for a task.""" + # Any agent can query sessions for tasks they have access to + try: + resp = await client.get(f"/tasks/{task_id}/sessions") + except Exception as e: + return format_error_response( + "CONNECTION_ERROR", + f"Failed to connect to API: {type(e).__name__}", + ) + + if resp.is_status(status.HTTP_404_NOT_FOUND): + return format_error_response("NOT_FOUND", f"Task {task_id} not found") + + if not resp.ok: + return format_error_response( + "FETCH_FAILED", + "Failed to fetch sessions for task", + {"status_code": resp.status_code, "detail": resp.text}, + ) + + data = resp.json() + sessions = data.get("sessions", []) + primary = next((s for s in sessions if s.get("is_primary")), None) + + guidance = f"Found {len(sessions)} session(s) for this task." + if primary: + guidance += f" Primary session: {primary.get('session_id', 'unknown')}." + else: + guidance += " No primary session set." + + return { + "status": "OK", + "sessions": sessions, + "primary_session_id": primary.get("session_id") if primary else None, + "guidance": guidance, + } diff --git a/roboco/mcp/tasks/handlers/work.py b/roboco/mcp/tasks/handlers/work.py index abe7433c..955e71a4 100644 --- a/roboco/mcp/tasks/handlers/work.py +++ b/roboco/mcp/tasks/handlers/work.py @@ -16,9 +16,9 @@ from roboco.mcp.tasks.handlers._helpers import ( ) from roboco.mcp.utils import ApiClient, format_error_response -ACTIVE_PROGRESS_STATUSES = frozenset({ - "in_progress", "verifying", "awaiting_qa", "awaiting_documentation" -}) +ACTIVE_PROGRESS_STATUSES = frozenset( + {"in_progress", "verifying", "awaiting_qa", "awaiting_documentation"} +) def _format_plan_response( diff --git a/roboco/models/__init__.py b/roboco/models/__init__.py index bb1d0831..91ac29cd 100644 --- a/roboco/models/__init__.py +++ b/roboco/models/__init__.py @@ -70,6 +70,10 @@ from roboco.models.session import ( Session, SessionConfig, SessionCreate, + SessionForTasksCreate, + SessionTaskLink, + SessionTaskLinkCreate, + SessionTaskRelationshipType, ) from roboco.models.task import ( Checkpoint, @@ -131,7 +135,11 @@ __all__ = [ "Session", "SessionConfig", "SessionCreate", + "SessionForTasksCreate", "SessionStatus", + "SessionTaskLink", + "SessionTaskLinkCreate", + "SessionTaskRelationshipType", "Task", "TaskCreate", "TaskPlan", diff --git a/roboco/models/base.py b/roboco/models/base.py index d1eda229..f8eda8b3 100644 --- a/roboco/models/base.py +++ b/roboco/models/base.py @@ -19,7 +19,8 @@ from pydantic import BaseModel, ConfigDict, Field class TaskStatus(str, Enum): """Task lifecycle states.""" - PENDING = "pending" + BACKLOG = "backlog" # PM setup phase - session must be created before activation + PENDING = "pending" # Ready for work - orchestrator can spawn agents CLAIMED = "claimed" IN_PROGRESS = "in_progress" BLOCKED = "blocked" diff --git a/roboco/models/messaging.py b/roboco/models/messaging.py index c3f03d9f..5685a1fd 100644 --- a/roboco/models/messaging.py +++ b/roboco/models/messaging.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from uuid import UUID from roboco.models.base import AgentRole, ChannelType, MessageType +from roboco.models.session import SessionScope @dataclass @@ -43,6 +44,7 @@ class SessionCreateRequest: max_message_count: int | None = 100 max_content_length: int | None = 50000 timeout_seconds: int = 300 + scope: SessionScope = SessionScope.TASK @dataclass diff --git a/roboco/models/session.py b/roboco/models/session.py index 10cbfa07..a052b7f6 100644 --- a/roboco/models/session.py +++ b/roboco/models/session.py @@ -3,9 +3,15 @@ Session Model Sessions group messages within boundaries (time, count, content length). They are automatically created and closed based on configuration. + +Session-Task Relationships: + PMs can create work sessions as discussion contexts for tasks. + A session can discuss multiple related tasks. + A task can have multiple sessions (planning, review, retrospective). """ from datetime import UTC, datetime, timedelta +from enum import StrEnum from uuid import UUID, uuid4 from pydantic import Field @@ -16,6 +22,44 @@ from roboco.models.base import ( TimestampMixin, ) +# ============================================================================= +# SESSION SCOPE (Context Level) +# ============================================================================= + + +class SessionScope(StrEnum): + """ + Scope level for sessions - determines context loading strategy. + + Sessions at different scopes serve different purposes: + - INITIATIVE: Cross-cell coordination (Main PM, #dev-all) + - CELL: Cell-specific work (Cell PM, #backend-cell) + - TASK: Individual task execution (Developer level) + + When loading context for an agent: + - Load their scope's sessions fully + - Load parent scope sessions as summaries/references + """ + + INITIATIVE = "initiative" # Cross-cell, Main PM level + CELL = "cell" # Cell-specific, Cell PM level + TASK = "task" # Individual task execution + + +# ============================================================================= +# SESSION-TASK RELATIONSHIP TYPES +# ============================================================================= + + +class SessionTaskRelationshipType(StrEnum): + """Type of relationship between a session and a task.""" + + DISCUSSION = "discussion" # General discussion about the task + PLANNING = "planning" # Planning session for the task + REVIEW = "review" # Review/retrospective session + RETROSPECTIVE = "retrospective" # Post-completion reflection + + # ============================================================================= # SUPPORTING MODELS # ============================================================================= @@ -72,6 +116,12 @@ class Session(TimestampMixin): # State status: SessionStatus = Field(default=SessionStatus.ACTIVE) + # Scope (for smart context loading) + scope: SessionScope = Field( + default=SessionScope.TASK, + description="Session scope level - initiative, cell, or task", + ) + # Timestamps started_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) last_activity_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) @@ -97,3 +147,65 @@ class SessionCreate(RobocoBase): group_id: UUID config: SessionConfig | None = None + + +# ============================================================================= +# SESSION-TASK LINK MODELS +# ============================================================================= + + +class SessionTaskLink(TimestampMixin): + """ + Represents a link between a session and a task. + + Used for reading/displaying session-task relationships. + """ + + id: UUID = Field(default_factory=uuid4, description="Link ID") + session_id: UUID = Field(..., description="Session ID") + task_id: UUID = Field(..., description="Task ID") + + # Relationship metadata + is_primary: bool = Field( + default=False, description="Is this the primary discussion session for the task" + ) + relationship_type: SessionTaskRelationshipType = Field( + default=SessionTaskRelationshipType.DISCUSSION, + description="Type of session-task relationship", + ) + + # Audit + added_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + added_by: UUID | None = Field(default=None, description="PM who created the link") + + +class SessionTaskLinkCreate(RobocoBase): + """Schema for creating a session-task link.""" + + session_id: UUID = Field(..., description="Session to link") + task_id: UUID = Field(..., description="Task to link") + is_primary: bool = Field( + default=False, description="Mark as primary session for this task" + ) + relationship_type: SessionTaskRelationshipType = Field( + default=SessionTaskRelationshipType.DISCUSSION, + description="Type of relationship", + ) + + +class SessionForTasksCreate(RobocoBase): + """Schema for PM creating a session linked to multiple tasks.""" + + task_ids: list[UUID] = Field(..., min_length=1, description="Tasks to link") + channel_slug: str = Field(..., description="Channel where session is created") + scope: SessionScope = Field( + default=SessionScope.CELL, + description="Session scope level for context loading strategy", + ) + config: SessionConfig | None = Field( + default=None, description="Session boundary configuration" + ) + relationship_type: SessionTaskRelationshipType = Field( + default=SessionTaskRelationshipType.DISCUSSION, + description="Relationship type for all links", + ) diff --git a/roboco/services/messaging.py b/roboco/services/messaging.py index 862c17f3..a0b52cbf 100644 --- a/roboco/services/messaging.py +++ b/roboco/services/messaging.py @@ -16,12 +16,14 @@ from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import joinedload from roboco.db.tables import ( ChannelTable, GroupTable, MessageTable, SessionTable, + SessionTaskTable, ) from roboco.enforcement import validate_channel_access from roboco.events import Event, EventType, get_event_bus @@ -35,7 +37,11 @@ from roboco.models.messaging import ( MessageCreateRequest, SessionCreateRequest, ) -from roboco.services.base import BaseService +from roboco.models.session import ( + SessionForTasksCreate, + SessionTaskRelationshipType, +) +from roboco.services.base import BaseService, ConflictError, NotFoundError # ============================================================================= # MESSAGING SERVICE @@ -281,6 +287,7 @@ class MessagingService(BaseService): max_content_length=req.max_content_length, timeout_seconds=req.timeout_seconds, status=SessionStatus.ACTIVE, + scope=req.scope, ) self.session.add(session) @@ -386,6 +393,261 @@ class MessagingService(BaseService): # Create new session return await self.create_session(SessionCreateRequest(group_id=group_id)) + # ========================================================================= + # SESSION-TASK LINKING OPERATIONS + # ========================================================================= + + async def link_session_to_task( + self, + session_id: UUID, + task_id: UUID, + added_by: UUID, + is_primary: bool = False, + relationship_type: SessionTaskRelationshipType = ( + SessionTaskRelationshipType.DISCUSSION + ), + ) -> SessionTaskTable: + """ + Link a session to a task. + + Args: + session_id: Session to link + task_id: Task to link + added_by: PM who created this link + is_primary: Mark as primary discussion session for this task + relationship_type: Type of relationship + + Returns: + Created link + + Raises: + NotFoundError: If session not found + ConflictError: If link already exists or primary constraint violated + """ + # Verify session exists + session = await self.get_session(session_id) + if not session: + raise NotFoundError(f"Session {session_id} not found") + + # Check if link already exists + existing = await self.session.execute( + select(SessionTaskTable).where( + SessionTaskTable.session_id == session_id, + SessionTaskTable.task_id == task_id, + ) + ) + if existing.scalar_one_or_none(): + raise ConflictError( + f"Session {session_id} is already linked to task {task_id}" + ) + + # If marking as primary, check if task already has a primary session + if is_primary: + existing_primary = await self.session.execute( + select(SessionTaskTable).where( + SessionTaskTable.task_id == task_id, + SessionTaskTable.is_primary.is_(True), + ) + ) + if existing_primary.scalar_one_or_none(): + raise ConflictError(f"Task {task_id} already has a primary session") + + link = SessionTaskTable( + session_id=session_id, + task_id=task_id, + is_primary=is_primary, + relationship_type=relationship_type.value, + added_by=added_by, + ) + + self.session.add(link) + await self.session.flush() + + self.log.info( + "Session linked to task", + session_id=str(session_id), + task_id=str(task_id), + is_primary=is_primary, + relationship_type=relationship_type.value, + ) + return link + + async def unlink_session_from_task( + self, + session_id: UUID, + task_id: UUID, + ) -> bool: + """ + Remove a session-task link. + + Args: + session_id: Session to unlink + task_id: Task to unlink + + Returns: + True if link was removed, False if not found + """ + result = await self.session.execute( + select(SessionTaskTable).where( + SessionTaskTable.session_id == session_id, + SessionTaskTable.task_id == task_id, + ) + ) + link = result.scalar_one_or_none() + + if not link: + return False + + await self.session.delete(link) + await self.session.flush() + + self.log.info( + "Session unlinked from task", + session_id=str(session_id), + task_id=str(task_id), + ) + return True + + async def get_sessions_for_task( + self, + task_id: UUID, + relationship_type: SessionTaskRelationshipType | None = None, + ) -> list[SessionTaskTable]: + """ + Get all sessions linked to a task. + + Args: + task_id: Task to get sessions for + relationship_type: Filter by relationship type + + Returns: + List of session-task links (with session→group→channel loaded) + """ + query = ( + select(SessionTaskTable) + .where(SessionTaskTable.task_id == task_id) + .options( + joinedload(SessionTaskTable.session) + .joinedload(SessionTable.group) + .joinedload(GroupTable.channel) + ) + ) + + if relationship_type: + query = query.where( + SessionTaskTable.relationship_type == relationship_type.value + ) + + query = query.order_by(SessionTaskTable.added_at.desc()) + result = await self.session.execute(query) + return list(result.scalars().unique().all()) + + async def get_primary_session_for_task( + self, + task_id: UUID, + ) -> SessionTaskTable | None: + """ + Get the primary session for a task. + + Args: + task_id: Task to get primary session for + + Returns: + Primary session-task link, or None if no primary session + """ + result = await self.session.execute( + select(SessionTaskTable).where( + SessionTaskTable.task_id == task_id, + SessionTaskTable.is_primary.is_(True), + ) + ) + return result.scalar_one_or_none() + + async def get_tasks_for_session( + self, + session_id: UUID, + ) -> list[SessionTaskTable]: + """ + Get all tasks linked to a session. + + Args: + session_id: Session to get tasks for + + Returns: + List of session-task links (with task relationship loaded) + """ + result = await self.session.execute( + select(SessionTaskTable) + .where(SessionTaskTable.session_id == session_id) + .order_by(SessionTaskTable.added_at.desc()) + ) + return list(result.scalars().all()) + + async def create_session_for_tasks( + self, + req: SessionForTasksCreate, + pm_agent_id: UUID, + ) -> tuple[SessionTable, list[SessionTaskTable]]: + """ + Create a new session linked to one or more tasks (PM operation). + + This is the main entry point for PMs to create work sessions. + + Args: + req: Session creation request with task IDs + pm_agent_id: PM agent creating the session + + Returns: + Tuple of (created session, list of created links) + + Raises: + NotFoundError: If channel not found + ValueError: If no groups found in channel + """ + # Get channel + channel = await self.get_channel_by_slug(req.channel_slug) + if not channel: + raise NotFoundError(f"Channel '{req.channel_slug}' not found") + + # Get first group in channel (or create default) + groups = await self.list_groups_in_channel(cast("UUID", channel.id)) + if not groups: + raise ValueError(f"No groups found in channel '{req.channel_slug}'") + group = groups[0] + + # Create session with config and scope + session_req = SessionCreateRequest( + group_id=cast("UUID", group.id), + max_message_count=(req.config.max_message_count if req.config else None), + max_content_length=(req.config.max_content_length if req.config else None), + timeout_seconds=(req.config.timeout_seconds if req.config else 300), + scope=req.scope, + ) + session = await self.create_session(session_req) + + # Link all tasks + links: list[SessionTaskTable] = [] + for i, task_id in enumerate(req.task_ids): + # First task is primary + is_primary = i == 0 + link = await self.link_session_to_task( + session_id=cast("UUID", session.id), + task_id=task_id, + added_by=pm_agent_id, + is_primary=is_primary, + relationship_type=req.relationship_type, + ) + links.append(link) + + self.log.info( + "Session created for tasks", + session_id=str(session.id), + task_count=len(req.task_ids), + channel_slug=req.channel_slug, + pm_agent_id=str(pm_agent_id), + ) + return session, links + def _check_session_boundaries(self, session: SessionTable) -> bool: """Check if session has exceeded boundaries. Returns True if should close.""" # Check message count diff --git a/roboco/services/metrics.py b/roboco/services/metrics.py index 95637ca9..f3f2873a 100644 --- a/roboco/services/metrics.py +++ b/roboco/services/metrics.py @@ -30,12 +30,14 @@ HOURS_PER_DAY = 24 SECONDS_PER_HOUR = 3600 # Active task statuses for queries -ACTIVE_STATUSES = frozenset({ - TaskStatus.CLAIMED, - TaskStatus.IN_PROGRESS, - TaskStatus.VERIFYING, - TaskStatus.AWAITING_QA, -}) +ACTIVE_STATUSES = frozenset( + { + TaskStatus.CLAIMED, + TaskStatus.IN_PROGRESS, + TaskStatus.VERIFYING, + TaskStatus.AWAITING_QA, + } +) # Health thresholds CRITICAL_BLOCKED_RATIO = 0.3 diff --git a/roboco/services/permissions.py b/roboco/services/permissions.py index ab30e9d3..d7a0cf67 100644 --- a/roboco/services/permissions.py +++ b/roboco/services/permissions.py @@ -431,3 +431,28 @@ async def has_privileged_access(db: "AsyncSession", agent_id: UUID) -> bool: ) role = result.scalar_one_or_none() return role in PRIVILEGED_ROLES if role else False + + +# Roles that can manage tasks and sessions (PMs and board) +PM_ROLES = frozenset({AgentRole.CELL_PM, AgentRole.MAIN_PM}) +MANAGEMENT_ROLES = frozenset( + {AgentRole.CEO, AgentRole.PRODUCT_OWNER, AgentRole.CELL_PM, AgentRole.MAIN_PM} +) + + +async def is_pm_role(db: "AsyncSession", agent_id: UUID) -> bool: + """ + Check if agent has a PM or management role. + + PM roles (Cell PM, Main PM) and management roles (CEO, Product Owner) + can create task-linked sessions and assign work. + """ + from roboco.db.tables import AgentTable + + result = await db.execute( + select(AgentTable.role).where( + (AgentTable.id == agent_id) | (AgentTable.slug == str(agent_id)) + ) + ) + role = result.scalar_one_or_none() + return role in MANAGEMENT_ROLES if role else False diff --git a/roboco/services/task.py b/roboco/services/task.py index 624ef2f3..92d2478c 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -12,7 +12,7 @@ from uuid import UUID from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession -from roboco.db.tables import AgentTable, TaskTable +from roboco.db.tables import AgentTable, SessionTaskTable, TaskTable from roboco.enforcement import ( TaskOwnershipError, validate_task_ownership, @@ -115,7 +115,15 @@ class TaskService(BaseService): # ========================================================================= async def create(self, req: TaskCreateRequest) -> TaskTable: - """Create a new task.""" + """ + Create a new task. + + Tasks are created with BACKLOG status by default. PM must: + 1. Create a session for the task + 2. Call activate() to transition to PENDING + + This ensures every task has a session before work begins. + """ task = TaskTable( title=req.title, description=req.description, @@ -126,11 +134,19 @@ class TaskService(BaseService): parent_task_id=req.parent_task_id, target_date=req.target_date, estimated_complexity=req.estimated_complexity, - status=TaskStatus.PENDING, + status=TaskStatus.BACKLOG, ) self.session.add(task) await self.session.flush() + # Inherit parent task's primary session for subtasks + if req.parent_task_id: + await self._inherit_parent_session( + task_id=cast("UUID", task.id), + parent_task_id=req.parent_task_id, + created_by=req.created_by, + ) + self.log.info( "Task created", task_id=str(task.id), @@ -139,6 +155,111 @@ class TaskService(BaseService): ) return task + async def _inherit_parent_session( + self, + task_id: UUID, + parent_task_id: UUID, + created_by: UUID, + ) -> SessionTaskTable | None: + """ + Inherit the parent task's primary session for a subtask. + + When a subtask is created, it automatically joins the parent task's + primary discussion session (if one exists). This enables context + continuity across the task hierarchy. + + Args: + task_id: The new subtask ID + parent_task_id: The parent task ID + created_by: PM who created the subtask + + Returns: + Created link if parent had a primary session, None otherwise + """ + # Find parent's primary session + result = await self.session.execute( + select(SessionTaskTable).where( + SessionTaskTable.task_id == parent_task_id, + SessionTaskTable.is_primary.is_(True), + ) + ) + parent_link = result.scalar_one_or_none() + + if not parent_link: + return None + + # Create link for subtask (not primary - parent owns the primary) + link = SessionTaskTable( + session_id=parent_link.session_id, + task_id=task_id, + is_primary=False, # Subtasks don't become primary + relationship_type=parent_link.relationship_type, + added_by=created_by, + ) + + self.session.add(link) + await self.session.flush() + + self.log.info( + "Subtask inherited parent session", + task_id=str(task_id), + parent_task_id=str(parent_task_id), + session_id=str(parent_link.session_id), + ) + return link + + async def activate(self, task_id: UUID) -> TaskTable: + """ + Activate a task from BACKLOG to PENDING status. + + This is a PM-only operation that transitions a task from setup + phase to ready-for-work phase. The orchestrator will then spawn + agents to work on it. + + REQUIRES: Task must have at least one linked session. + + Args: + task_id: The task to activate + + Returns: + The activated task + + Raises: + ValueError: If task not found, not in BACKLOG, or has no session + """ + task = await self.get(task_id) + if not task: + raise ValueError(f"Task {task_id} not found") + + if task.status != TaskStatus.BACKLOG: + raise ValueError( + f"Task {task_id} is not in BACKLOG status (current: {task.status})" + ) + + # Check if task has at least one linked session + result = await self.session.execute( + select(SessionTaskTable).where(SessionTaskTable.task_id == task_id).limit(1) + ) + session_link = result.scalar_one_or_none() + + if not session_link: + raise ValueError( + f"Task {task_id} has no linked session. " + "Create a session with roboco_session_create_for_tasks " + "before activating." + ) + + # Transition to PENDING + task.status = TaskStatus.PENDING + await self.session.flush() + + self.log.info( + "Task activated", + task_id=str(task_id), + session_id=str(session_link.session_id), + ) + return task + async def get(self, task_id: UUID) -> TaskTable | None: """Get a task by ID.""" result = await self.session.execute(