1. Blueprint updates - Added NO_GROUPS escalation docs to all 12 agent blueprints

2. Multi-image Docker architecture - Created role-specific Dockerfiles:
  - agent-base.Dockerfile (shared foundation)
  - agent-pm.Dockerfile, agent-dev-be.Dockerfile, agent-dev-fe.Dockerfile
  - agent-qa-be.Dockerfile, agent-qa-fe.Dockerfile, agent-doc.Dockerfile, agent-ux.Dockerfile
3. Orchestrator updates - roboco/runtime/orchestrator.py:
  - Added AGENT_IMAGES mapping and get_agent_image() function
  - Updated _ensure_agent_image() to build base + specialized images
  - Updated _spawn_container() to use role-specific image
  - Made _generate_mcp_config() role-aware (though kept notify for all since they need to receive)

agents/blueprints/backend/be-dev.md

@@ -212,6 +212,13 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()`
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (be-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### Channels You Access
 - **#backend-cell** (read/write) - Your primary workspace
 - **#dev-all** (read/write) - Cross-cell dev discussion

agents/blueprints/backend/be-documenter.md

@@ -125,6 +125,13 @@ and verify all subtasks are done before calling `roboco_task_complete()`.
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (be-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### When to Post in Session (DO)
 - **Questions about implementation** - Need dev/QA clarification
 - **Missing context** - Dev notes don't explain something critical

agents/blueprints/backend/be-pm.md

@@ -187,6 +187,21 @@ roboco_session_create_for_tasks({
 - Subtasks auto-inherit parent task's primary session
 - Full audit trail preserved
 
+**Handling NO_GROUPS Error:**
+If you get a NO_GROUPS error when creating a session, it means the channel
+doesn't have a group for this initiative yet. Groups are created by Main PM.
+
+Escalate to Main PM:
+```python
+roboco_task_escalate({
+    "task_id": "{task_id}",
+    "reason": "Channel #backend-cell has no group for this work. Need group created.",
+    "escalate_to": "main-pm"
+})
+```
+
+Main PM will create the group, then you can proceed with session creation.
+
 ### 7b. ACTIVATE TASK (REQUIRED)
 **Tool:** `roboco_task_activate(task_id)`
 

agents/blueprints/backend/be-qa.md

@@ -226,6 +226,13 @@ After verdict:
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (be-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### Channels You Access
 - **#backend-cell** (read/write) - Your primary workspace
 - **#qa-all** (read/write) - Cross-cell QA discussion

agents/blueprints/board/main-pm.md

@@ -51,6 +51,9 @@ You interact with RoboCo systems through MCP tools:
 - `roboco_task_create(...)` - Create new tasks for cells (pass `status: "backlog"` for setup phase)
 - `roboco_task_activate(task_id)` - Activate task from BACKLOG to PENDING (after session created)
 
+**Group Management (Feature/Initiative Scopes):**
+- `roboco_group_create(data)` - Create a group for a feature/initiative in a channel
+
 **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
@@ -146,7 +149,21 @@ roboco_task_create({
 })
 ```
 
-**2. CREATE WORK SESSION (REQUIRED)**
+**2. CREATE GROUP (if needed)**
+Before creating sessions, ensure the channel has a group for this initiative:
+```python
+roboco_group_create({
+    "channel_slug": "dev-all",      # Or "backend-cell" for cell-specific work
+    "name": "User Preferences Feature",
+    "hierarchy_level": 4            # 0=CEO, 1=Board, 2=Main PM, 3=Cell PM, 4=Members
+})
+```
+
+Groups organize work into feature/initiative scopes. Cell PMs then create
+sessions within groups for actual work items. If a Cell PM escalates to you
+with a NO_GROUPS error, create the group and notify them.
+
+**3. CREATE WORK SESSION (REQUIRED)**
 Every initiative needs a work session for coordination:
 ```python
 roboco_session_create_for_tasks({
@@ -165,7 +182,7 @@ roboco_session_create_for_tasks({
 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)**
+**4. ACTIVATE TASKS (REQUIRED)**
 After sessions are created, activate tasks so Cell PMs can see them:
 ```python
 roboco_task_activate("backend-task-id")
@@ -175,7 +192,7 @@ roboco_task_activate("ux-task-id")
 
 **Task flow:**
 ```
-CREATE (backlog) → SESSION → ACTIVATE (pending) → Cell PM receives task
+CREATE (backlog) → GROUP (if needed) → SESSION → ACTIVATE (pending) → Cell PM receives
 ```
 
 **4. NOTIFY CELL PMs**

agents/blueprints/frontend/fe-dev.md

@@ -191,6 +191,13 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()`
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (fe-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### Channels You Access
 - **#frontend-cell** (read/write) - Your primary workspace
 - **#dev-all** (read/write) - Cross-cell dev discussion

agents/blueprints/frontend/fe-documenter.md

@@ -123,6 +123,13 @@ and verify all subtasks are done before calling `roboco_task_complete()`.
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (fe-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### When to Post in Session (DO)
 - **Questions about implementation** - Need dev/QA clarification
 - **Missing context** - Dev notes don't explain component behavior

agents/blueprints/frontend/fe-pm.md

@@ -183,6 +183,21 @@ roboco_session_create_for_tasks({
 - QA and documenter see full context when reviewing
 - Subtasks auto-inherit parent task's primary session
 
+**Handling NO_GROUPS Error:**
+If you get a NO_GROUPS error when creating a session, it means the channel
+doesn't have a group for this initiative yet. Groups are created by Main PM.
+
+Escalate to Main PM:
+```python
+roboco_task_escalate({
+    "task_id": "{task_id}",
+    "reason": "Channel #frontend-cell has no group for this work. Need group created.",
+    "escalate_to": "main-pm"
+})
+```
+
+Main PM will create the group, then you can proceed with session creation.
+
 ### 7b. ACTIVATE TASK (REQUIRED)
 **Tool:** `roboco_task_activate(task_id)`
 

agents/blueprints/frontend/fe-qa.md

@@ -130,6 +130,13 @@ Update progress: `roboco_task_progress(task_id, "Completed visual testing...", 5
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (fe-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### When to Post in Session (DO)
 - **Questions about implementation** - Need dev clarification on behavior
 - **Critical bugs** - Security issues, accessibility failures, blockers

agents/blueprints/ux_ui/ux-dev.md

@@ -190,6 +190,13 @@ Return to SCAN: `roboco_task_scan()` or `roboco_agent_idle()`
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (ux-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### Channels You Access
 - **#uxui-cell** (read/write) - Your primary workspace
 - **#dev-all** (read) - See what frontend is building

agents/blueprints/ux_ui/ux-documenter.md

@@ -123,6 +123,13 @@ and verify all subtasks are done before calling `roboco_task_complete()`.
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (ux-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### When to Post in Session (DO)
 - **Questions about design decisions** - Need designer/QA clarification
 - **Missing context** - Design notes don't explain rationale

agents/blueprints/ux_ui/ux-pm.md

@@ -183,6 +183,21 @@ roboco_session_create_for_tasks({
 - QA and documenter see design context
 - Frontend can review design discussion history
 
+**Handling NO_GROUPS Error:**
+If you get a NO_GROUPS error when creating a session, it means the channel
+doesn't have a group for this initiative yet. Groups are created by Main PM.
+
+Escalate to Main PM:
+```python
+roboco_task_escalate({
+    "task_id": "{task_id}",
+    "reason": "Channel #uxui-cell has no group for this work. Need group created.",
+    "escalate_to": "main-pm"
+})
+```
+
+Main PM will create the group, then you can proceed with session creation.
+
 ### 7b. ACTIVATE TASK (REQUIRED)
 **Tool:** `roboco_task_activate(task_id)`
 

agents/blueprints/ux_ui/ux-qa.md

@@ -129,6 +129,13 @@ If dev_notes is empty or no Figma link provided, that's a valid FAIL reason.
 
 ## Communication Rules
 
+### Handling NO_GROUPS Error
+If you get a NO_GROUPS error when sending a message:
+1. This means the channel hasn't been set up for this work yet
+2. Escalate to your Cell PM (ux-pm) using `roboco_task_escalate`
+3. Include the channel and task context in your escalation
+4. If you have a task_id, always include it in message calls (routes to task session)
+
 ### When to Post in Session (DO)
 - **Questions about design intent** - Need designer clarification
 - **Critical issues** - Accessibility failures, missing states

docker/agent-base.Dockerfile

@@ -36,7 +36,6 @@ COPY --chown=agent:agent pyproject.toml uv.lock README.md /app/
 USER agent
 
 # Install Python dependencies for MCP servers (as agent)
-# Increase timeout for large NVIDIA packages (674MB cudnn, 858MB torch)
 ENV UV_HTTP_TIMEOUT=300
 ENV UV_CONCURRENT_DOWNLOADS=4
 RUN uv python install 3.13 && uv sync --frozen --python 3.13

docker/agent-dev-be.Dockerfile

@@ -0,0 +1,17 @@
+# Backend Developer Agent
+# Python/FastAPI development tools
+
+FROM roboco-agent-base
+
+USER root
+
+# Backend-specific tools
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    postgresql-client \
+    redis-tools \
+    && rm -rf /var/lib/apt/lists/*
+
+USER agent
+
+LABEL role="backend-developer"
+LABEL description="Backend developer agent - Python, FastAPI, databases"

docker/agent-dev-fe.Dockerfile

@@ -0,0 +1,36 @@
+# Frontend Developer Agent
+# React/TypeScript development with browser automation
+
+FROM roboco-agent-base
+
+USER root
+
+# Playwright system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libnss3 \
+    libnspr4 \
+    libatk1.0-0 \
+    libatk-bridge2.0-0 \
+    libcups2 \
+    libdrm2 \
+    libxkbcommon0 \
+    libxcomposite1 \
+    libxdamage1 \
+    libxfixes3 \
+    libxrandr2 \
+    libgbm1 \
+    libasound2 \
+    libpango-1.0-0 \
+    libcairo2 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install pnpm globally
+RUN npm install -g pnpm
+
+USER agent
+
+# Install Playwright (browsers will be installed on first run or can be cached)
+RUN npx playwright install chromium
+
+LABEL role="frontend-developer"
+LABEL description="Frontend developer agent - React, TypeScript, Playwright"

docker/agent-doc.Dockerfile

@@ -0,0 +1,11 @@
+# Documenter Agent
+# Lightweight - documentation doesn't need heavy tools
+
+FROM roboco-agent-base
+
+# No additional tools needed
+# Documenters write markdown, update READMEs, changelogs
+# They use the mounted /app/docs directory
+
+LABEL role="documenter"
+LABEL description="Documenter agent - technical writing, API docs, changelogs"

docker/agent-pm.Dockerfile

@@ -0,0 +1,10 @@
+# PM Agent - Lightweight coordinator
+# PMs don't code, they coordinate and delegate
+
+FROM roboco-agent-base
+
+# No additional tools needed - PMs use MCP tools only
+# They get: task management, messaging, notifications, journaling
+
+LABEL role="pm"
+LABEL description="Project Manager agent - coordinates work, delegates to developers"

docker/agent-qa-be.Dockerfile

@@ -0,0 +1,18 @@
+# Backend QA Agent
+# Testing and quality assurance tools for backend
+
+FROM roboco-agent-base
+
+USER root
+
+# QA tools for backend
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    postgresql-client \
+    && rm -rf /var/lib/apt/lists/*
+
+USER agent
+
+# Python testing tools are already in pyproject.toml (pytest, coverage, etc.)
+
+LABEL role="backend-qa"
+LABEL description="Backend QA agent - testing, code review, quality assurance"

docker/agent-qa-fe.Dockerfile

@@ -0,0 +1,35 @@
+# Frontend QA Agent
+# Browser testing and accessibility tools
+
+FROM roboco-agent-base
+
+USER root
+
+# Playwright system dependencies (same as fe-dev)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libnss3 \
+    libnspr4 \
+    libatk1.0-0 \
+    libatk-bridge2.0-0 \
+    libcups2 \
+    libdrm2 \
+    libxkbcommon0 \
+    libxcomposite1 \
+    libxdamage1 \
+    libxfixes3 \
+    libxrandr2 \
+    libgbm1 \
+    libasound2 \
+    libpango-1.0-0 \
+    libcairo2 \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN npm install -g pnpm
+
+USER agent
+
+# Playwright for browser testing
+RUN npx playwright install chromium
+
+LABEL role="frontend-qa"
+LABEL description="Frontend QA agent - browser testing, accessibility, visual regression"

docker/agent-ux.Dockerfile

@@ -0,0 +1,12 @@
+# UX/UI Agent
+# Design tools - future: Figma MCP, image generation
+
+FROM roboco-agent-base
+
+# Future additions:
+# - Figma MCP server integration
+# - Image generation tools
+# - Design token management
+
+LABEL role="ux-designer"
+LABEL description="UX/UI agent - design, prototyping, design system"

roboco/agents/base.py

@@ -338,39 +338,73 @@ class Agent(ABC):
 
     async def send_message(
         self,
-        channel_id: UUID,
+        session_id: UUID | None,
         content: str,
         message_type: str = "dialogue",
+        task_id: UUID | None = None,
     ) -> None:
         """
-        Send a message to a channel.
+        Send a message to a session.
 
         Args:
-            channel_id: Target channel
+            session_id: Target session (from ctx.session_id)
             content: Message content
             message_type: Type of message (reasoning, dialogue, action, etc.)
+            task_id: Optional task context for the message
         """
+        if session_id is None:
+            # No session - escalate to PM
+            self.log.warning(
+                "Cannot send message: no session_id provided",
+                task_id=str(task_id) if task_id else None,
+                content_preview=content[:50],
+            )
+            if task_id:
+                await self._escalate_no_session(task_id)
+            return
+
         self.state.messages_sent += 1
         self.state.last_activity = datetime.now(UTC)
 
-        url = f"http://{settings.host}:{settings.port}/api/v1/messages"
+        try:
-        async with httpx.AsyncClient() as client:
+            await self._api_call(
-            await client.post(
+                "POST",
-                url,
+                "/messages",
                 json={
-                    "channel_id": str(channel_id),
+                    "session_id": str(session_id),
-                    "agent_id": str(self.id),
+                    "type": message_type,
                     "content": content,
-                    "message_type": message_type,
+                    "task_id": str(task_id) if task_id else None,
                 },
             )
+            self.log.debug(
+                "Message sent",
+                session_id=str(session_id),
+                message_type=message_type,
+                content_length=len(content),
+            )
+        except Exception as e:
+            self.log.warning("Failed to send message", error=str(e))
 
-        self.log.debug(
+    async def _escalate_no_session(self, task_id: UUID) -> None:
-            "Message sent",
+        """Escalate when no session is available for a task."""
-            channel_id=str(channel_id),
+        self.log.info(
-            message_type=message_type,
+            "Escalating: task has no session",
-            content_length=len(content),
+            task_id=str(task_id),
+            agent_role=self.role.value if self.role else "unknown",
         )
+        # Record escalation via API (orchestrator handles routing)
+        try:
+            await self._api_call(
+                "POST",
+                f"/tasks/{task_id}/escalate",
+                json={
+                    "reason": "Task has no linked session for communication",
+                    "agent_id": str(self.id),
+                },
+            )
+        except Exception as e:
+            self.log.warning("Failed to record escalation", error=str(e))
 
     async def stream_reasoning(self, content: str) -> None:
         """
@@ -512,13 +546,38 @@ class Agent(ABC):
 
     async def _get_task_title(self, task_id: UUID) -> str:
         """Get task title from API."""
+        title, _ = await self._get_task_info(task_id)
+        return title
+
+    async def _get_task_info(self, task_id: UUID) -> tuple[str, UUID | None]:
+        """
+        Get task info including title and primary session_id.
+
+        The task response includes linked sessions. We extract the primary
+        session_id so it can be stored in the context for message routing.
+
+        Args:
+            task_id: Task to fetch
+
+        Returns:
+            Tuple of (title, session_id). session_id is None if no primary session.
+        """
         try:
             result = await self._api_call("GET", f"/tasks/{task_id}")
             title: str = result.get("title", f"Task {str(task_id)[:8]}")
-            return title
+
+            # Extract primary session from linked sessions
+            session_id: UUID | None = None
+            sessions = result.get("sessions", [])
+            for session in sessions:
+                if session.get("is_primary"):
+                    session_id = UUID(session["session_id"])
+                    break
+
+            return title, session_id
         except Exception as e:
-            self.log.warning("Failed to get task title", error=str(e))
+            self.log.warning("Failed to get task info", error=str(e))
-            return f"Task {str(task_id)[:8]}"
+            return f"Task {str(task_id)[:8]}", None
 
     async def _read_task_requirements(self, task_id: UUID) -> str:
         """Read task requirements from task record."""
@@ -780,25 +839,25 @@ class Agent(ABC):
         task_id: UUID,
         message: str,
         percentage: int,
-        channel_id: UUID | None = None,
+        session_id: UUID | None = None,
     ) -> None:
         """
-        Report progress: save to task AND send channel message.
+        Report progress: save to task AND send session message.
 
         Args:
             task_id: Task to update
             message: Progress message
             percentage: Completion percentage (0-100)
-            channel_id: Channel to notify (uses cell_channel_id or task_id)
+            session_id: Session to notify (from context)
         """
         await self._add_progress(task_id, message, percentage)
 
-        target_channel = channel_id or self.cell_channel_id or task_id
         task_ref = str(task_id)[:8]
         await self.send_message(
-            target_channel,
+            session_id,
             f"TASK-{task_ref} ({percentage}%) {message}",
             message_type="action",
+            task_id=task_id,
         )
 
     # =========================================================================

roboco/agents/developer.py

@@ -13,7 +13,7 @@ import structlog
 
 from roboco.agents.base import Agent, AgentConfig
 from roboco.agents.mixins import PhaseConfig, PhaseEngine
-from roboco.models import TaskStatus
+from roboco.models import AgentStatus, TaskStatus
 from roboco.models.agents import DevTaskPhase, TaskContext
 
 logger = structlog.get_logger()
@@ -146,9 +146,11 @@ class DeveloperAgent(Agent, PhaseEngine[DevTaskPhase, TaskContext]):
         """
         # Initialize or restore task context
         if self._task_context is None or self._task_context.task_id != task_id:
+            title, session_id = await self._get_task_info(task_id)
             self._task_context = TaskContext(
                 task_id=task_id,
-                title=await self._get_task_title(task_id),
+                title=title,
+                session_id=session_id,
             )
 
         ctx = self._task_context
@@ -190,11 +192,12 @@ class DeveloperAgent(Agent, PhaseEngine[DevTaskPhase, TaskContext]):
         # Update task status
         await self._update_task_status(ctx.task_id, TaskStatus.CLAIMED)
 
-        # Announce in channel
+        # Announce in session
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"Claiming TASK-{str(ctx.task_id)[:8]}: {ctx.title}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         # Journal entry
@@ -245,12 +248,13 @@ If clarification needed, respond with: "QUESTION: [your question]"
             )
             return True
         else:
-            # Ask question in channel
+            # Ask question in session
             question = response.replace("QUESTION:", "").strip()
             await self.send_message(
-                self._cell_channel_id or ctx.task_id,
+                ctx.session_id,
                 f"Question about TASK-{str(ctx.task_id)[:8]}: {question}",
                 message_type="dialogue",
+                task_id=ctx.task_id,
             )
             return False
 
@@ -313,9 +317,10 @@ Add unit tests,tests/test_main.py,small
 
         # Announce plan
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"TASK-{str(ctx.task_id)[:8]} plan ready: {len(ctx.subtasks)} subtasks",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
     async def _phase_execute(self, ctx: TaskContext) -> bool:
@@ -385,9 +390,10 @@ Respond with the implementation.
         await self._add_progress(ctx.task_id, progress_msg, percentage)
 
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"TASK-{str(ctx.task_id)[:8]} ({percentage}%) {progress_msg}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         ctx.current_subtask += 1
@@ -424,10 +430,11 @@ Respond with the implementation.
         if all_passed:
             # Flag for QA
             await self.send_message(
-                self._cell_channel_id or ctx.task_id,
+                ctx.session_id,
                 f"TASK-{str(ctx.task_id)[:8]} ready for QA review. "
                 f"Commits: {', '.join(ctx.commits)}",
                 message_type="action",
+                task_id=ctx.task_id,
             )
             ctx.journal_entries.append(
                 f"[{datetime.now(UTC).isoformat()}] VERIFY PASSED. Flagged for QA."
@@ -495,9 +502,10 @@ Summarize in 2-3 sentences what documentation is needed.
         if ctx.blockers:
             blocker = ctx.blockers[-1]
             await self.send_message(
-                self._cell_channel_id or ctx.task_id,
+                ctx.session_id,
                 f"BLOCKED on TASK-{str(ctx.task_id)[:8]}: {blocker}",
                 message_type="blocker",
+                task_id=ctx.task_id,
             )
             await self._update_task_status(ctx.task_id, TaskStatus.BLOCKED)
 
@@ -541,12 +549,12 @@ Summarize in 2-3 sentences what documentation is needed.
             return None
 
     async def _signal_availability(self) -> None:
-        """Signal availability to PM."""
+        """Signal availability to orchestrator (no task context, so use API)."""
-        await self.send_message(
+        self.log.info("Signaling availability", agent_name=self.name)
-            self._cell_channel_id or self.id,
+        # No task/session context - signal via state update instead of message
-            f"{self.name} available for new tasks",
+        self.state.status = AgentStatus.IDLE
-            message_type="dialogue",
+        self.state.current_task_id = None
-        )
+        self.state.current_session_id = None
 
     async def _submit_for_qa(
         self, task_id: UUID, dev_notes: str, handoff_summary: str

roboco/agents/documenter.py

@@ -134,9 +134,11 @@ class DocumenterAgent(Agent, PhaseEngine[DocTaskPhase, DocContext]):
         Returns True when documentation is complete.
         """
         if self._doc_context is None or self._doc_context.task_id != task_id:
+            title, session_id = await self._get_task_info(task_id)
             self._doc_context = DocContext(
                 task_id=task_id,
-                title=await self._get_task_title(task_id),
+                title=title,
+                session_id=session_id,
             )
 
         ctx = self._doc_context
@@ -177,9 +179,10 @@ class DocumenterAgent(Agent, PhaseEngine[DocTaskPhase, DocContext]):
         self.log.info("RECEIVE phase", task_id=str(ctx.task_id))
 
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"Starting documentation for TASK-{str(ctx.task_id)[:8]}: {ctx.title}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         ctx.notes.append(f"[{datetime.now(UTC).isoformat()}] Documentation started")
@@ -338,9 +341,10 @@ Format appropriately for the document type.
 
         progress = f"{ctx.current_doc}/{len(ctx.documents_needed)}"
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"TASK-{str(ctx.task_id)[:8]} doc {progress}: {doc_spec.title}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         return ctx.current_doc >= len(ctx.documents_needed)
@@ -415,10 +419,11 @@ good,complete,clear,helpful,None
         await self._mark_awaiting_pm_review(ctx.task_id)
 
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"TASK-{str(ctx.task_id)[:8]} documentation complete, awaiting PM review\n"
             f"Published: {', '.join(ctx.written_docs)}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         ctx.notes.append(f"[{datetime.now(UTC).isoformat()}] Documentation published")

roboco/agents/mixins.py

@@ -370,26 +370,30 @@ class ProgressTracker:
 
     @abstractmethod
     async def send_message(
-        self, channel_id: UUID, content: str, message_type: str
+        self,
+        session_id: UUID | None,
+        content: str,
+        message_type: str,
+        task_id: UUID | None = None,
     ) -> None:
-        """Send message to channel (implemented in base Agent)."""
+        """Send message to session (implemented in base Agent)."""
         ...
 
     async def _report_progress(
         self,
         task_id: UUID,
-        channel_id: UUID | None,
+        session_id: UUID | None,
         message: str,
         percentage: int,
     ) -> None:
         """
         Report progress for a task.
 
-        Saves to task record AND sends channel message.
+        Saves to task record AND sends message to session.
 
         Args:
             task_id: Task to update
-            channel_id: Channel to notify (uses task_id if None)
+            session_id: Session to notify (from context)
             message: Progress message
             percentage: Completion percentage (0-100)
         """
@@ -401,10 +405,11 @@ class ProgressTracker:
                 json={"message": message, "percentage": percentage},
             )
 
-        # Send channel message
+        # Send session message
         task_ref = str(task_id)[:8]
         await self.send_message(
-            channel_id or task_id,
+            session_id,
             f"TASK-{task_ref} ({percentage}%) {message}",
             message_type="action",
+            task_id=task_id,
         )

roboco/agents/pm.py

@@ -221,10 +221,13 @@ medium,TASK-abc123,P1,backend-dev-1
             if resolved:
                 success = await self._unblock_task(task_id)
                 if success:
+                    # Get session from task
+                    _, session_id = await self._get_task_info(task_id)
                     await self.send_message(
-                        self._cell_channel_id or self.id,
+                        session_id,
                         f"TASK-{str(task_id)[:8]} unblocked - blocker resolved",
                         message_type="action",
+                        task_id=task_id,
                     )
 
         # Check for pending questions in channel
@@ -249,10 +252,11 @@ As the Cell PM, provide:
 Be helpful and unblock the team.
 """
             response = await self.think(prompt)
-            await self.send_message(
+            # TODO: Questions should include task_id for routing
-                self._cell_channel_id or self.id,
+            # For now, log that we can't route without session context
-                response,
+            self.log.info(
-                message_type="dialogue",
+                "PM response (no session context - need task_id in questions)",
+                response_preview=response[:100],
             )
 
     async def _phase_escalate(self) -> None:

roboco/agents/qa.py

@@ -133,9 +133,11 @@ class QAAgent(Agent, PhaseEngine[QATaskPhase, ReviewContext]):
         Returns True when review is complete.
         """
         if self._review_context is None or self._review_context.task_id != task_id:
+            title, session_id = await self._get_task_info(task_id)
             self._review_context = ReviewContext(
                 task_id=task_id,
-                title=await self._get_task_title(task_id),
+                title=title,
+                session_id=session_id,
             )
 
         ctx = self._review_context
@@ -181,9 +183,10 @@ class QAAgent(Agent, PhaseEngine[QATaskPhase, ReviewContext]):
         self.log.info("RECEIVE phase", task_id=str(ctx.task_id))
 
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             f"Starting review of TASK-{str(ctx.task_id)[:8]}: {ctx.title}",
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         ctx.notes.append(f"[{datetime.now(UTC).isoformat()}] Review started")
@@ -321,9 +324,10 @@ PASS,All criteria verified successfully,No issues found
         task_ref = str(ctx.task_id)[:8]
         msg = f"TASK-{task_ref} test {progress}: {test_case.name} - {result_str}"
         await self.send_message(
-            self._cell_channel_id or ctx.task_id,
+            ctx.session_id,
             msg,
             message_type="action",
+            task_id=ctx.task_id,
         )
 
         return ctx.current_test >= len(ctx.test_cases)
@@ -350,11 +354,12 @@ PASS,All criteria verified successfully,No issues found
             )
 
             await self.send_message(
-                self._cell_channel_id or ctx.task_id,
+                ctx.session_id,
                 f"TASK-{str(ctx.task_id)[:8]} QA FAILED\n\n"
                 f"Issues found:\n{failure_summary}\n\n"
                 f"Task returned to developer for fixes.",
                 message_type="decision",
+                task_id=ctx.task_id,
             )
 
             # Update task status
@@ -364,11 +369,12 @@ PASS,All criteria verified successfully,No issues found
             ctx.verdict = TestResult.PASS
 
             await self.send_message(
-                self._cell_channel_id or ctx.task_id,
+                ctx.session_id,
                 f"TASK-{str(ctx.task_id)[:8]} QA APPROVED\n\n"
                 f"All {len(ctx.test_cases)} tests passed.\n"
                 f"Ready for documentation.",
                 message_type="decision",
+                task_id=ctx.task_id,
             )
 
             # Update task status

roboco/api/routes/groups.py

@@ -0,0 +1,140 @@
+"""
+Groups API Routes
+
+Endpoints for managing groups within channels.
+Groups are created by Main PM to organize work into feature/initiative scopes.
+Cell PMs then create sessions within groups for actual work items.
+"""
+
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException, status
+
+from roboco.api.deps import CurrentAgentContext, DbSession
+from roboco.api.schemas.groups import (
+    GroupCreateRequest,
+    GroupDetailResponse,
+    GroupResponse,
+)
+from roboco.models.base import AgentRole
+from roboco.models.messaging import GroupCreateRequest as ServiceGroupCreate
+from roboco.services.messaging import get_messaging_service
+from roboco.utils.converters import require_uuid, to_python_uuid
+
+router = APIRouter()
+
+# Roles authorized to create groups
+GROUP_ADMIN_ROLES = frozenset({AgentRole.CEO, AgentRole.MAIN_PM, AgentRole.AUDITOR})
+
+
+@router.post(
+    "",
+    response_model=GroupResponse,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create group",
+    description=(
+        "Create a new group in a channel. Groups organize work into "
+        "feature/initiative scopes. Only Main PM can create groups."
+    ),
+)
+async def create_group(
+    db: DbSession,
+    agent: CurrentAgentContext,
+    data: GroupCreateRequest,
+) -> GroupResponse:
+    """
+    Create a new group in a channel.
+
+    Groups are the organizational unit between channels and sessions:
+    - Main PM creates Groups for features/initiatives
+    - Cell PMs create Sessions within Groups for work items
+    - Developers communicate within Sessions
+    """
+    if agent.role not in GROUP_ADMIN_ROLES:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=(
+                "Only Main PM, CEO, or Auditor can create groups. "
+                "If you need a group created, escalate to Main PM."
+            ),
+        )
+
+    service = get_messaging_service(db)
+
+    # Get channel by slug
+    channel = await service.get_channel_by_slug(data.channel_slug)
+    if not channel:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Channel not found: {data.channel_slug}",
+        )
+
+    # Create group via service
+    try:
+        group = await service.create_group(
+            ServiceGroupCreate(
+                name=data.name,
+                channel_id=require_uuid(channel.id),
+                allowed_roles=data.allowed_roles,
+                hierarchy_level=data.hierarchy_level,
+            )
+        )
+        await db.commit()
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        ) from e
+
+    return GroupResponse(
+        id=require_uuid(group.id),
+        name=group.name,
+        channel_id=require_uuid(group.channel_id),
+        channel_slug=data.channel_slug,
+        hierarchy_level=group.hierarchy_level,
+        is_active=group.is_active,
+        total_sessions=group.total_sessions,
+        total_messages=group.total_messages,
+        active_session_id=to_python_uuid(group.active_session_id),
+    )
+
+
+@router.get(
+    "/{group_id}",
+    response_model=GroupDetailResponse,
+    summary="Get group",
+    description="Get detailed information about a group.",
+)
+async def get_group(
+    db: DbSession,
+    group_id: UUID,
+) -> GroupDetailResponse:
+    """Get a group by ID."""
+    service = get_messaging_service(db)
+
+    group = await service.get_group(group_id)
+    if not group:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Group not found",
+        )
+
+    # Get channel slug for response
+    channel = await service.get_channel(require_uuid(group.channel_id))
+    channel_slug = channel.slug if channel else "unknown"
+
+    return GroupDetailResponse(
+        id=require_uuid(group.id),
+        name=group.name,
+        channel_id=require_uuid(group.channel_id),
+        channel_slug=channel_slug,
+        hierarchy_level=group.hierarchy_level,
+        is_active=group.is_active,
+        total_sessions=group.total_sessions,
+        total_messages=group.total_messages,
+        active_session_id=to_python_uuid(group.active_session_id),
+        allowed_roles=[
+            r.value if hasattr(r, "value") else r for r in group.allowed_roles
+        ],
+        members=list(group.members) if group.members else [],
+    )

roboco/api/schemas/groups.py

@@ -0,0 +1,61 @@
+"""
+Groups API Schemas
+
+Request/response models for group endpoints.
+"""
+
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+from roboco.models.base import AgentRole
+
+
+class GroupCreateRequest(BaseModel):
+    """Request to create a group in a channel."""
+
+    channel_slug: str = Field(
+        ...,
+        description="Channel slug where group will be created",
+        examples=["backend-cell", "dev-all"],
+    )
+    name: str = Field(
+        ...,
+        min_length=1,
+        max_length=100,
+        description="Group name",
+        examples=["User Preferences Feature", "Sprint 12 Work"],
+    )
+    hierarchy_level: int = Field(
+        default=4,
+        ge=0,
+        le=4,
+        description=(
+            "Access level: 0=CEO, 1=Board, 2=Main PM, 3=Cell PM, 4=Cell Members"
+        ),
+    )
+    allowed_roles: list[AgentRole] | None = Field(
+        default=None,
+        description="Specific roles that can access (overrides hierarchy)",
+    )
+
+
+class GroupResponse(BaseModel):
+    """Response for a created/retrieved group."""
+
+    id: UUID
+    name: str
+    channel_id: UUID
+    channel_slug: str
+    hierarchy_level: int
+    is_active: bool
+    total_sessions: int = 0
+    total_messages: int = 0
+    active_session_id: UUID | None = None
+
+
+class GroupDetailResponse(GroupResponse):
+    """Detailed group response with additional fields."""
+
+    allowed_roles: list[str]
+    members: list[UUID]

roboco/mcp/message_server.py

@@ -18,7 +18,7 @@ from typing import Any
 from fastapi import status
 from mcp.server.fastmcp import FastMCP
 
-from roboco.agents_config import CHANNEL_ACCESS
+from roboco.agents_config import CHANNEL_ACCESS, get_agent_role
 from roboco.llm import ToonAdapter
 from roboco.mcp.schemas import (
     AskQuestionInput,
@@ -40,6 +40,36 @@ _toon = ToonAdapter()
 # =============================================================================
 
 
+def _format_no_groups_error(agent_id: str, channel_slug: str) -> dict[str, Any]:
+    """Format NO_GROUPS error with role-aware escalation guidance."""
+    role = get_agent_role(agent_id)
+
+    if role == "main_pm":
+        guidance = (
+            f"Channel #{channel_slug} has no groups. "
+            "Use roboco_group_create to create a group for this channel."
+        )
+    elif role == "cell_pm":
+        guidance = (
+            f"Channel #{channel_slug} has no groups. "
+            "Groups are created by Main PM. "
+            "Use roboco_task_escalate to request group creation."
+        )
+    else:
+        # Developer/QA/Documenter
+        guidance = (
+            f"Channel #{channel_slug} has no groups yet. "
+            "Escalate to your Cell PM. "
+            "If you have a task_id, include it in your message call."
+        )
+
+    return format_error_response(
+        "NO_GROUPS",
+        f"Channel #{channel_slug} has no groups.",
+        {"guidance": guidance, "channel": channel_slug, "role": role},
+    )
+
+
 def _check_channel_access(agent_id: str, channel_slug: str, action: str) -> bool:
     """Check if agent has access to channel for the given action."""
     channel = CHANNEL_ACCESS.get(channel_slug, {})
@@ -102,8 +132,14 @@ def _validate_message_send(
 async def _get_default_group(
     client: ApiClient,
     channel_id: str,
+    agent_id: str | None = None,
+    channel_slug: str | None = None,
 ) -> str | dict[str, Any]:
-    """Get the default (first) group for a channel. Returns group_id or error dict."""
+    """Get the default (first) group for a channel. Returns group_id or error dict.
+
+    If agent_id and channel_slug are provided, NO_GROUPS error includes
+    role-aware escalation guidance.
+    """
     resp = await client.get(f"/channels/{channel_id}/groups")
 
     if not resp.ok:
@@ -115,6 +151,9 @@ async def _get_default_group(
 
     groups = resp.json()
     if not groups:
+        # Return role-aware guidance if we have context
+        if agent_id and channel_slug:
+            return _format_no_groups_error(agent_id, channel_slug)
         return format_error_response("NO_GROUPS", "Channel has no groups")
 
     # Return first active group, or first group if none are active
@@ -127,6 +166,8 @@ async def _get_default_group(
 async def _get_active_session(
     client: ApiClient,
     channel_id: str,
+    agent_id: str | None = None,
+    channel_slug: str | None = None,
 ) -> str | dict[str, Any]:
     """Get active session for channel. Returns session_id or error dict.
 
@@ -135,9 +176,11 @@ async def _get_active_session(
     returns an error guiding the agent to use task_id or ask their PM.
     """
     # First get the default group for this channel
-    group_result = await _get_default_group(client, channel_id)
+    group_result = await _get_default_group(
+        client, channel_id, agent_id=agent_id, channel_slug=channel_slug
+    )
     if isinstance(group_result, dict):
-        return group_result  # Error response
+        return group_result  # Error response (NO_GROUPS with guidance)
     group_id = group_result
 
     # Check if group has an active session
@@ -281,8 +324,10 @@ async def _handle_channel_history(
         return channel_result
     channel_id = channel_result
 
-    # Get group
+    # Get group (with role-aware guidance if NO_GROUPS)
-    group_result = await _get_default_group(client, channel_id)
+    group_result = await _get_default_group(
+        client, channel_id, agent_id=agent_id, channel_slug=channel_slug
+    )
     if isinstance(group_result, dict):
         return group_result
     group_id = group_result
@@ -366,9 +411,11 @@ async def _handle_message_send(
             return channel_result  # Error response
         channel_id = channel_result
 
-        session_result = await _get_active_session(client, channel_id)
+        session_result = await _get_active_session(
+            client, channel_id, agent_id=agent_id, channel_slug=data.channel_slug
+        )
         if isinstance(session_result, dict):
-            return session_result  # Returns NO_ACTIVE_SESSION error with guidance
+            return session_result  # Returns NO_GROUPS/NO_ACTIVE_SESSION with guidance
         session_id = session_result
 
     # Resolve mentions (slugs) to UUIDs using shared cache

roboco/mcp/schemas/__init__.py

@@ -258,3 +258,35 @@ class SessionLinkTaskInput(BaseModel):
         default="discussion",
         description="Type: discussion, planning, review, retrospective",
     )
+
+
+# =============================================================================
+# GROUP SCHEMAS (Main PM Only)
+# =============================================================================
+
+
+class GroupCreateInput(BaseModel):
+    """Input for creating a group in a channel (Main PM only).
+
+    Groups organize work into feature/initiative scopes within channels.
+    - Main PM creates Groups for features/initiatives
+    - Cell PMs create Sessions within Groups for work items
+    - Developers communicate within Sessions
+    """
+
+    channel_slug: str = Field(
+        ...,
+        description="Channel slug where group will be created (e.g., 'backend-cell')",
+    )
+    name: str = Field(
+        ...,
+        min_length=1,
+        max_length=100,
+        description="Group name (e.g., 'User Preferences Feature')",
+    )
+    hierarchy_level: int = Field(
+        default=4,
+        ge=0,
+        le=4,
+        description="Access level: 0=CEO, 1=Board, 2=Main PM, 3=Cell PM, 4=Members",
+    )

roboco/mcp/task_server.py

@@ -34,6 +34,7 @@ from typing import Any
 from mcp.server.fastmcp import FastMCP
 
 from roboco.mcp.schemas import (
+    GroupCreateInput,
     SessionCreateForTasksInput,
     SessionLinkTaskInput,
     TaskAssignInput,
@@ -45,6 +46,7 @@ from roboco.mcp.schemas import (
 from roboco.mcp.tasks.handlers import (
     handle_agent_idle,
     handle_docs_complete,
+    handle_group_create,
     handle_session_create_for_tasks,
     handle_session_get_for_task,
     handle_session_link_task,
@@ -656,6 +658,29 @@ def _register_session_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> N
         """
         return await handle_session_get_for_task(client, task_id, agent_id)
 
+    @mcp.tool()
+    async def roboco_group_create(data: GroupCreateInput) -> dict[str, Any]:
+        """
+        Create a group in a channel (Main PM only).
+
+        Groups organize work into feature/initiative scopes within channels.
+        The typical workflow is:
+        1. Main PM creates a Group for a feature/initiative
+        2. Cell PM creates Sessions within the Group for work items
+        3. Developers communicate within Sessions
+
+        ENFORCEMENT:
+        - Only Main PM, CEO, or Auditor can create groups
+        - Cell PMs should escalate if they need a group created
+
+        Args:
+            data: GroupCreateInput with channel_slug, name, hierarchy_level
+
+        Returns:
+            Created group with guidance
+        """
+        return await handle_group_create(client, data, agent_id)
+
 
 def create_task_mcp_server(agent_id: str) -> FastMCP:
     """

roboco/mcp/tasks/handlers/__init__.py

@@ -30,6 +30,7 @@ from roboco.mcp.tasks.handlers.review import (
 )
 from roboco.mcp.tasks.handlers.scan import handle_task_get, handle_task_scan
 from roboco.mcp.tasks.handlers.sessions import (
+    handle_group_create,
     handle_session_create_for_tasks,
     handle_session_get_for_task,
     handle_session_link_task,
@@ -44,6 +45,7 @@ from roboco.mcp.tasks.handlers.work import (
 __all__ = [
     "handle_agent_idle",
     "handle_docs_complete",
+    "handle_group_create",
     "handle_session_create_for_tasks",
     "handle_session_get_for_task",
     "handle_session_link_task",

roboco/mcp/tasks/handlers/sessions.py

@@ -258,3 +258,86 @@ async def handle_session_get_for_task(
         "primary_session_id": primary.get("session_id") if primary else None,
         "guidance": guidance,
     }
+
+
+# =============================================================================
+# GROUP HANDLERS (Main PM Only)
+# =============================================================================
+
+
+def _validate_main_pm_permissions(agent_id: str) -> dict[str, Any] | None:
+    """Validate agent has Main PM permissions (for group creation).
+
+    Only Main PM, CEO, and Auditor can create groups.
+    Cell PMs should escalate to Main PM for group creation.
+    """
+    role = get_agent_role(agent_id)
+    allowed_roles = {"main_pm", "ceo", "auditor"}
+    if role not in allowed_roles:
+        return format_error_response(
+            "PERMISSION_DENIED",
+            "Only Main PM can create groups. Cell PMs should escalate.",
+            {
+                "role": role,
+                "guidance": "Use roboco_task_escalate to request group creation.",
+            },
+        )
+    return None
+
+
+async def handle_group_create(
+    client: ApiClient,
+    input_data: Any,  # GroupCreateInput from roboco.mcp.schemas
+    agent_id: str,
+) -> dict[str, Any]:
+    """Handle group creation (Main PM only).
+
+    Groups organize work into feature/initiative scopes within channels.
+    Cell PMs then create sessions within groups for work items.
+    """
+    if error := _validate_main_pm_permissions(agent_id):
+        return error
+
+    payload = {
+        "channel_slug": input_data.channel_slug,
+        "name": input_data.name,
+        "hierarchy_level": input_data.hierarchy_level,
+    }
+
+    try:
+        resp = await client.post("/groups", 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_403_FORBIDDEN):
+        return format_error_response(
+            "PERMISSION_DENIED",
+            "API rejected group creation",
+            {"detail": resp.text},
+        )
+
+    if resp.is_status(status.HTTP_404_NOT_FOUND):
+        return format_error_response(
+            "NOT_FOUND",
+            f"Channel not found: {input_data.channel_slug}",
+        )
+
+    if not resp.is_status(status.HTTP_201_CREATED):
+        return format_error_response(
+            "CREATE_FAILED",
+            "Failed to create group",
+            {"status_code": resp.status_code, "detail": resp.text},
+        )
+
+    group_data = resp.json()
+    return {
+        "status": "created",
+        "group": group_data,
+        "guidance": (
+            f"Group '{input_data.name}' created in #{input_data.channel_slug}. "
+            "Cell PMs can now create sessions within this group."
+        ),
+    }

roboco/models/agents.py

@@ -117,6 +117,7 @@ class TaskContext:
 
     task_id: UUID
     title: str
+    session_id: UUID | None = None  # Primary session for this task
     phase: DevTaskPhase = DevTaskPhase.CLAIM
     subtasks: list[dict[str, Any]] = field(default_factory=list)
     current_subtask: int = 0
@@ -170,6 +171,7 @@ class ReviewContext:
 
     task_id: UUID
     title: str
+    session_id: UUID | None = None  # Primary session for this task
     phase: QATaskPhase = QATaskPhase.RECEIVE
     test_cases: list[TestCase] = field(default_factory=list)
     current_test: int = 0
@@ -286,6 +288,7 @@ class DocContext:
 
     task_id: UUID
     title: str
+    session_id: UUID | None = None  # Primary session for this task
     phase: DocTaskPhase = DocTaskPhase.RECEIVE
     # Gathered materials
     dev_notes: str | None = None

roboco/runtime/orchestrator.py

@@ -47,8 +47,39 @@ AgentState = OrchestratorAgentState
 AgentConfig = OrchestratorAgentConfig
 
 # Docker configuration
-AGENT_IMAGE = "roboco-agent"
 AGENT_NETWORK = "roboco_default"
+AGENT_BASE_IMAGE = "roboco-agent-base"
+
+# Role -> Image mapping
+# Specialized images extend the base with role-specific tools
+AGENT_IMAGES: dict[str, str] = {
+    # Backend
+    "be-dev-1": "roboco-agent-dev-be",
+    "be-dev-2": "roboco-agent-dev-be",
+    "be-qa": "roboco-agent-qa-be",
+    "be-pm": "roboco-agent-pm",
+    "be-doc": "roboco-agent-doc",
+    # Frontend
+    "fe-dev-1": "roboco-agent-dev-fe",
+    "fe-dev-2": "roboco-agent-dev-fe",
+    "fe-qa": "roboco-agent-qa-fe",
+    "fe-pm": "roboco-agent-pm",
+    "fe-doc": "roboco-agent-doc",
+    # UX/UI
+    "ux-dev": "roboco-agent-ux",
+    "ux-qa": "roboco-agent-ux",  # Uses same as dev for now
+    "ux-pm": "roboco-agent-pm",
+    "ux-doc": "roboco-agent-doc",
+    # Board
+    "main-pm": "roboco-agent-pm",
+    "product-owner": "roboco-agent-pm",
+    "head-marketing": "roboco-agent-pm",
+    "auditor": "roboco-agent-pm",
+}
+
+def get_agent_image(agent_id: str) -> str:
+    """Get the Docker image for an agent."""
+    return AGENT_IMAGES.get(agent_id, AGENT_BASE_IMAGE)
 
 # When running in a container, we need host paths for volume mounts.
 # These can be overridden via environment variables.
@@ -141,37 +172,70 @@ class AgentOrchestrator:
 
         logger.info("Orchestrator stopped")
 
-    async def _ensure_agent_image(self) -> None:
+    async def _ensure_agent_image(self, agent_id: str | None = None) -> None:
-        """Ensure the agent Docker image is built."""
+        """Ensure the agent Docker images are built.
+
+        Builds base image first, then specialized image if agent_id provided.
+        """
+        # Determine build context
+        if PROJECT_HOST_PATH:
+            build_context = PROJECT_HOST_PATH
+            docker_dir = f"{PROJECT_HOST_PATH}/docker"
+        else:
+            build_context = str(self.project_root)
+            docker_dir = str(self.project_root / "docker")
+
+        # Always ensure base image exists
+        await self._build_image_if_missing(
+            AGENT_BASE_IMAGE,
+            f"{docker_dir}/agent-base.Dockerfile",
+            build_context,
+        )
+
+        # Build specialized image if agent specified
+        if agent_id:
+            image = get_agent_image(agent_id)
+            if image != AGENT_BASE_IMAGE:
+                # Map image name to dockerfile
+                dockerfile_map = {
+                    "roboco-agent-pm": "agent-pm.Dockerfile",
+                    "roboco-agent-dev-be": "agent-dev-be.Dockerfile",
+                    "roboco-agent-dev-fe": "agent-dev-fe.Dockerfile",
+                    "roboco-agent-qa-be": "agent-qa-be.Dockerfile",
+                    "roboco-agent-qa-fe": "agent-qa-fe.Dockerfile",
+                    "roboco-agent-doc": "agent-doc.Dockerfile",
+                    "roboco-agent-ux": "agent-ux.Dockerfile",
+                }
+                dockerfile = dockerfile_map.get(image)
+                if dockerfile:
+                    await self._build_image_if_missing(
+                        image,
+                        f"{docker_dir}/{dockerfile}",
+                        build_context,
+                    )
+
+    async def _build_image_if_missing(
+        self, image_name: str, dockerfile_path: str, build_context: str
+    ) -> None:
+        """Build a Docker image if it doesn't exist."""
         # Check if image exists
         proc = await asyncio.create_subprocess_exec(
             "docker",
             "image",
             "inspect",
-            AGENT_IMAGE,
+            image_name,
             stdout=asyncio.subprocess.DEVNULL,
             stderr=asyncio.subprocess.DEVNULL,
         )
         await proc.wait()
 
         if proc.returncode != 0:
-            logger.info("Building agent Docker image...")
+            logger.info("Building Docker image...", image=image_name)
-
-            # Determine paths - use host paths when running in container
-            if PROJECT_HOST_PATH:
-                # Running in container - use host paths for Docker build
-                dockerfile_path = f"{PROJECT_HOST_PATH}/docker/agent.Dockerfile"
-                build_context = PROJECT_HOST_PATH
-            else:
-                # Running on host
-                dockerfile_path = str(self.project_root / "docker" / "agent.Dockerfile")
-                build_context = str(self.project_root)
-
             proc = await asyncio.create_subprocess_exec(
                 "docker",
                 "build",
                 "-t",
-                AGENT_IMAGE,
+                image_name,
                 "-f",
                 dockerfile_path,
                 build_context,
@@ -180,8 +244,10 @@ class AgentOrchestrator:
             )
             _, stderr = await proc.communicate()
             if proc.returncode != 0:
-                raise RuntimeError(f"Failed to build agent image: {stderr.decode()}")
+                raise RuntimeError(
-            logger.info("Agent Docker image built successfully")
+                    f"Failed to build image {image_name}: {stderr.decode()}"
+                )
+            logger.info("Docker image built successfully", image=image_name)
 
     def _ensure_agent_claude_settings(self) -> None:
         """
@@ -291,6 +357,9 @@ class AgentOrchestrator:
             # Ensure agent Claude settings have MCP tools allowed
             self._ensure_agent_claude_settings()
 
+            # Ensure agent-specific Docker image is built
+            await self._ensure_agent_image(agent_id)
+
             # Generate MCP config
             mcp_config_path = await self._generate_mcp_config(agent_id)
 
@@ -402,8 +471,8 @@ class AgentOrchestrator:
             f"ROBOCO_AGENT_ID={config.agent_id}",
             "-e",
             "ROBOCO_API_URL=http://roboco-orchestrator:8000",
-            # The image
+            # The image (role-specific)
-            AGENT_IMAGE,
+            get_agent_image(config.agent_id),
             # Claude Code arguments
             "--model",
             MODEL_MAP.get(config.model, config.model),
@@ -458,16 +527,16 @@ class AgentOrchestrator:
         await proc.wait()
 
     async def _generate_mcp_config(self, agent_id: str) -> Path:
-        """Generate MCP config for an agent."""
+        """Generate role-aware MCP config for an agent.
+
+        Different roles get different MCP server access:
+        - All agents: task, message, journal
+        - PMs only: notify (for sending notifications)
+        """
         # MCP servers run inside agent containers, need to connect via Docker network
-        # When orchestrator is in a container: use container hostname
-        # When orchestrator is on host: use localhost
-        # NOTE: internal_api_url adds /api/v1, so we just need the base URL here
         if PROJECT_HOST_PATH:
-            # Running in container - agents connect via Docker network
             api_url = "http://roboco-orchestrator:8000"
         else:
-            # Running on host - agents connect to localhost
             api_url = f"http://127.0.0.1:{settings.port}"
 
         mcp_env = {
@@ -475,49 +544,53 @@ class AgentOrchestrator:
             "ROBOCO_AGENT_ID": agent_id,
         }
 
-        config = {
+        # Base MCP servers - all agents get these
-            "mcpServers": {
+        mcp_servers: dict[str, dict[str, Any]] = {
-                "roboco-task": {
+            "roboco-task": {
-                    "command": "uv",
+                "command": "uv",
-                    "args": ["run", "python", "-m", "roboco.mcp.task_server", agent_id],
+                "args": ["run", "python", "-m", "roboco.mcp.task_server", agent_id],
-                    "env": mcp_env,
+                "env": mcp_env,
-                },
+            },
-                "roboco-message": {
+            "roboco-message": {
-                    "command": "uv",
+                "command": "uv",
-                    "args": [
+                "args": [
-                        "run",
+                    "run",
-                        "python",
+                    "python",
-                        "-m",
+                    "-m",
-                        "roboco.mcp.message_server",
+                    "roboco.mcp.message_server",
-                        agent_id,
+                    agent_id,
-                    ],
+                ],
-                    "env": mcp_env,
+                "env": mcp_env,
-                },
+            },
-                "roboco-notify": {
+            "roboco-journal": {
-                    "command": "uv",
+                "command": "uv",
-                    "args": [
+                "args": [
-                        "run",
+                    "run",
-                        "python",
+                    "python",
-                        "-m",
+                    "-m",
-                        "roboco.mcp.notify_server",
+                    "roboco.mcp.journal_server",
-                        agent_id,
+                    agent_id,
-                    ],
+                ],
-                    "env": mcp_env,
+                "env": mcp_env,
-                },
+            },
-                "roboco-journal": {
-                    "command": "uv",
-                    "args": [
-                        "run",
-                        "python",
-                        "-m",
-                        "roboco.mcp.journal_server",
-                        agent_id,
-                    ],
-                    "env": mcp_env,
-                },
-            }
         }
 
+        # Notify server - everyone can READ notifications, only PMs can SEND
+        # (permission check happens at handler level)
+        mcp_servers["roboco-notify"] = {
+            "command": "uv",
+            "args": [
+                "run",
+                "python",
+                "-m",
+                "roboco.mcp.notify_server",
+                agent_id,
+            ],
+            "env": mcp_env,
+        }
+
+        config: dict[str, Any] = {"mcpServers": mcp_servers}
+
         # Write to shared config directory (mounted in both orchestrator and agents)
         # When running in container: /app/mcp-configs -> host's ./data/mcp-configs
         # When running on host: use temp directory