feat(sessions): Session-Task linking with scoped context management

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
2026-08-03 07:23:24 +02:00 · 2025-12-22 18:04:31 +01:00
parent 4e06bdb842
commit b7611c24fc
36 changed files with 2118 additions and 92 deletions
@@ -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", {
@@ -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
@@ -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
@@ -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)
@@ -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
@@ -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)
@@ -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
@@ -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
@@ -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
@@ -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)
@@ -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
@@ -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
@@ -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
@@ -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")
@@ -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]
@@ -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
+        ),
+    )
@@ -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
@@ -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

@@ -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
 # =============================================================================
@@ -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


@@ -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",
+    )
@@ -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


@@ -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",
@@ -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)
@@ -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",
@@ -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]

@@ -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,
+    }
@@ -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(
@@ -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",
@@ -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"
@@ -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
@@ -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",
+    )
@@ -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
@@ -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
@@ -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
@@ -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(