General fixes mainly around git integration into task lifecycle

CLAUDE.md

@@ -2,11 +2,14 @@
 
 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
 
-## IMPORTANT NOTES
+## IGNORING THESE WILL FORCE A COMPLETE SHUTDOWN OF CLAUDE CODE
 
 **IGNORING != FIXING**
 **`# noqa` & `# type: ignore` != FIXING**
 **`uv run mypy ... --ignore-missing-imports` | ANY IGNORING AT ALL != GOOD PRACTICES**
+**`http://192.168.50.111:8000/docs` IS THE API DOCS**
+** You need `X-Agent-Id` and `X-Agent-Role` headers to be set as 'ceo' for all API calls **
+**`ssh renzof@renzof-nas.local` SSH TO THE SERVER**
 
 ## Project Overview
 
@@ -223,7 +226,7 @@ Certain transitions require specific roles:
 ### Git Integration Requirements
 
 For tasks with `requires_git=True`:
-1. **claimed -> in_progress**: Must have `branch_name` set (PM creates branch first)
+1. **claimed -> in_progress**: `branch_name` is auto-set on claim (hierarchical branches)
 2. **awaiting_documentation -> awaiting_pm_review**: Requires BOTH `docs_complete=True` AND `pr_created=True`
 3. **awaiting_pm_review -> awaiting_ceo_approval**: Must have `pr_number` set
 
@@ -259,7 +262,7 @@ Major tasks are escalated to CEO for final approval:
 task_type: TaskType      # code, documentation, research, planning, design, administrative
 requires_git: bool       # Whether git workflow applies
 project_id: UUID         # Project this task works on
-branch_name: str         # Branch created for this task (set by PM)
+branch_name: str         # Branch for this task (auto-created on claim)
 work_session_id: UUID    # Active work session
 
 # PR tracking (parallel execution in awaiting_documentation)

agents/prompts/roles/board.md

@@ -88,10 +88,11 @@ roboco_task_cancel(task_id)    # If no longer needed
 - `roboco_git_log(project_slug, limit)` - Recent commits
 - `roboco_git_branch_list(project_slug)` - List branches
 - `roboco_git_diff(project_slug, staged)` - View code changes
-- `roboco_git_create_branch(project_slug, task_id, branch_type, parent_branch)` - Create branches
 - `roboco_git_checkout(project_slug, branch)` - Switch branches
 - `roboco_git_merge_pr(project_slug, pr_number, task_id, merge_method)` - Merge PRs
 
+**Note:** Branches are auto-created when tasks are claimed. No manual creation needed.
+
 **Session Management:**
 - `roboco_session_create_for_tasks`, `roboco_session_link_task`
 - `roboco_session_unlink_task`, `roboco_session_get_for_task`

agents/prompts/roles/cell_pm.md

@@ -17,7 +17,7 @@ For communication structure: `roboco_kb_search("communication hierarchy")`
 ## Workflow
 
 ```
-SCAN → CLAIM → PLAN → SESSION → SUBTASKS → CREATE_BRANCH → ACTIVATE → NOTIFY → PAUSE → MONITOR → REVIEW_PR → COMPLETE
+SCAN → CLAIM → PLAN → SESSION → SUBTASKS → ACTIVATE → NOTIFY → PAUSE → MONITOR → REVIEW_PR → COMPLETE
 ```
 
 ### 1. SCAN
@@ -56,54 +56,19 @@ roboco_task_create(
 - Completion tracking breaks
 - Your task can't complete
 
-### 5. CREATE BRANCH (Git Tasks)
+### 5. ACTIVATE
-**For tasks with `requires_git=True`:**
-```
-roboco_git_create_branch(project_slug, task_id, branch_type, parent_branch)
-```
-
-- **branch_type**: `feature`, `bug`, `chore`, `docs`, `hotfix`
-- **parent_branch**: Parent task's branch (or project's default branch for top-level)
-- Branch naming: `{type}/{team}/{task_id}` (auto-generated)
-
-**Example:**
-```
-roboco_git_create_branch("roboco", "abc123", "feature")
-# Creates: feature/backend/abc123
-```
-
-**Branch Hierarchy (when using hierarchical branching):**
-Branches are typically created top-down:
-1. Main PM creates root branch from default branch
-2. You create subtask branch from root branch
-3. Dev works on subsubtask branch from your branch
-
-**When creating a subtask branch:**
-- Check if parent task has a branch set
-- If so, use parent's branch as `parent_branch` parameter
-- If branching from default branch directly, that works too
-
-**Branch hierarchy example:**
-```
-default (main/master/etc)
-  └─ feature/backend/ROOT123     ← Main PM's branch
-      └─ feature/backend/ROOT123/SUB456  ← Your branch
-          └─ feature/backend/ROOT123/SUB456/DEV789  ← Dev branch
-```
-
-### 6. ACTIVATE
 `roboco_task_activate()` moves backlog → pending. Now visible to devs.
 
-### 7. NOTIFY
+### 6. NOTIFY
 `roboco_notify_send()` to each assignee. REQUIRED.
 
-### 8. PAUSE + IDLE
+### 7. PAUSE + IDLE
 `roboco_task_pause()` with checkpoint, then `roboco_agent_idle()`.
 
-### 9. MONITOR
+### 8. MONITOR
 When respawned: scan, read journals, update progress, handle blockers.
 
-### 10. REVIEW PR (Git Tasks)
+### 9. REVIEW PR (Git Tasks)
 When subtasks reach `awaiting_pm_review`:
 1. Review the PR: `roboco_git_diff(project_slug)` to see changes
 2. Check QA notes and documentation
@@ -113,7 +78,7 @@ When subtasks reach `awaiting_pm_review`:
 
 **Merge methods:** `squash` (default), `merge`, `rebase`
 
-### 11. COMPLETE
+### 10. COMPLETE
 When ALL subtasks done: reflect + complete your task.
 
 ## Your Tools
@@ -134,10 +99,11 @@ When ALL subtasks done: reflect + complete your task.
 - `roboco_git_diff(project_slug, staged)` - View changes
 
 **Git (PM Branch Management):**
-- `roboco_git_create_branch(project_slug, task_id, branch_type, parent_branch)` - Create task branch
 - `roboco_git_checkout(project_slug, branch)` - Switch branches
 - `roboco_git_merge_pr(project_slug, pr_number, task_id, merge_method)` - Merge PR (subtask→parent)
 
+**Note:** Branches are auto-created when tasks are claimed. No manual branch creation needed.
+
 **Git (Developer Tools - You Have These Too):**
 - `roboco_git_commit`, `roboco_git_push`, `roboco_git_create_pr`
 

agents/prompts/roles/developer.md

@@ -29,7 +29,7 @@ Use `roboco_task_claim()`. Status: pending → claimed.
 
 ### 4. CHECKOUT (Git Tasks)
 **For tasks with `requires_git=True`:**
-- Branch already created by PM
+- Branch auto-created when you claimed the task
 - **Auto-checkout happens on `roboco_task_start()`** - no manual checkout needed
 - System blocks if you have uncommitted changes
 

agents/prompts/roles/main_pm.md

@@ -18,7 +18,7 @@ For communication structure: `roboco_kb_search("communication hierarchy")`
 ## Workflow
 
 ```
-SCAN → CLAIM → PLAN → CREATE_PARENT_BRANCH → CREATE GROUP → CREATE CELL TASKS → ACTIVATE → NOTIFY → PAUSE → MONITOR → REVIEW_PR → COMPLETE
+SCAN → CLAIM → PLAN → CREATE GROUP → CREATE CELL TASKS → ACTIVATE → NOTIFY → PAUSE → MONITOR → REVIEW_PR → COMPLETE
 ```
 
 ### 1. SCAN
@@ -27,30 +27,10 @@ Use `roboco_task_scan()` for tasks assigned to you from Board/CEO.
 ### 2. CLAIM + PLAN
 Claim → read full description → plan breakdown across cells → start → journal decision.
 
-### 3. CREATE PARENT BRANCH (Git Tasks)
+### 3. CREATE GROUP
-**For tasks with `requires_git=True`:**
-```
-roboco_git_create_branch(project_slug, task_id, branch_type)
-```
-
-- Creates parent branch from the project's default branch (e.g., `main`, `master`)
-- Cell PM subtask branches will fork from this
-- Example: `feature/cross/abc123` for cross-cell work
-
-**Branch Hierarchy for Git Tasks:**
-- For hierarchical branching: default branch → Your branch → Cell PM branch → Dev branch
-- If using hierarchical branches, create root branch before Cell PMs create theirs
-- Cell PM branches fork from your branch (set as `parent_branch`)
-
-**Typical order for hierarchical branching:**
-1. Create your root branch from project's default branch
-2. Create cell tasks, Cell PMs create branches from your branch
-3. Devs create branches from Cell PM branches
-
-### 4. CREATE GROUP
 Use `roboco_group_create()` in each relevant cell channel. Cell PMs need groups to create sessions.
 
-### 5. CREATE CELL TASKS
+### 4. CREATE CELL TASKS
 
 **CRITICAL: Always set `parent_task_id` to YOUR task ID.** Without this, you create orphan tasks, not subtasks.
 
@@ -76,19 +56,18 @@ roboco_task_create(
 - Completion tracking breaks
 - Your task can't complete
 
-- Set `project_id` and `branch_name` for git tasks
+- Set `project_id` for git tasks (branches are auto-created on claim)
-- Cell PMs will create subtask branches from your parent branch
 
-### 6. ACTIVATE + NOTIFY
+### 5. ACTIVATE + NOTIFY
 `roboco_task_activate()` each task, then `roboco_notify_send()` to each Cell PM. REQUIRED.
 
-### 7. PAUSE + IDLE
+### 6. PAUSE + IDLE
 `roboco_task_pause()` with checkpoint, then `roboco_agent_idle()`.
 
-### 8. MONITOR
+### 7. MONITOR
 When respawned: scan, read Cell PM journals, update progress, coordinate if blockers.
 
-### 9. REVIEW PR (Git Tasks)
+### 8. REVIEW PR (Git Tasks)
 When cell tasks reach `awaiting_pm_review` and all subtasks are merged:
 1. Review the parent PR (all subtask work combined)
 2. Coordinate with Cell PM - **BOTH must approve**
@@ -98,7 +77,7 @@ When cell tasks reach `awaiting_pm_review` and all subtasks are merged:
 
 **CEO merges the final PR to main.**
 
-### 10. COMPLETE
+### 9. COMPLETE
 When CEO approves and PR is merged: reflect + complete your task.
 
 ## Your Tools
@@ -121,10 +100,11 @@ When CEO approves and PR is merged: reflect + complete your task.
 - `roboco_git_diff(project_slug, staged)` - View changes
 
 **Git (PM Branch Management):**
-- `roboco_git_create_branch(project_slug, task_id, branch_type, "main")` - Create parent branch
 - `roboco_git_checkout(project_slug, branch)` - Switch branches
 - `roboco_git_merge_pr(project_slug, pr_number, task_id, merge_method)` - Merge PR
 
+**Note:** Branches are auto-created when tasks are claimed. No manual branch creation needed.
+
 **Group Management (Main PM ONLY):**
 - `roboco_group_create` - Create groups in channels
 

docs/rag/roles/cell-pm.md

@@ -14,7 +14,7 @@
 3. Assign work to cell members
 4. Complete tasks after full workflow
 5. Handle escalations from cell
-6. Create branches for git tasks
+6. Review and merge PRs for git tasks
 
 ## What You CAN Do
 
@@ -26,7 +26,7 @@
 - Unblock blocked tasks
 - Send notifications
 - Index code and documentation
-- Create branches: `roboco_git_create_branch()`
+- Merge PRs: `roboco_git_merge_pr()`
 
 ## What You CANNOT Do
 
@@ -62,17 +62,13 @@ roboco_notify_send({
 
 For tasks with `requires_git=True`:
 
-```python
+**Branches are auto-created when tasks are claimed:**
-# Create branch BEFORE developer can start
+- When you claim your task: `feature/team/MAIN_PM_ID/YOUR_ID`
-roboco_git_create_branch(
+- When devs claim their subtasks: `feature/team/MAIN_PM_ID/YOUR_ID/DEV_ID`
-    project_slug="roboco",
-    task_id=task_id,
-    branch_type="feature"
-)
-# Creates: feature/backend/a1b2c3d4
-```
 
-Developer cannot start (`claimed` → `in_progress`) until branch exists.
+**No manual branch creation needed.** Just claim the task and the hierarchical branch is auto-created.
+
+PRs merge bottom-up: dev branch → your branch → main PM branch → main.
 
 ## Completing Tasks
 
@@ -120,7 +116,7 @@ See: `roboco_kb_search("tool permissions")`
 | `roboco_task_complete` | Finish task |
 | `roboco_task_cancel` | Cancel task |
 | `roboco_task_unblock` | Unblock blocked task |
-| `roboco_git_create_branch` | Create task branch |
+| `roboco_git_merge_pr` | Merge developer PRs |
 | `roboco_notify_send` | Send notification |
 | `roboco_project_update` | Update own cell's projects |
 | `roboco_workspace_list` | List own cell's workspaces |

docs/rag/roles/developer.md

@@ -46,11 +46,12 @@ pending → claim → plan → start → work → submit_verification → submit
 | State | Meaning |
 |-------|---------|
 | `NEEDS_PLAN` | Must call `roboco_task_plan()` first |
-| `WAITING_FOR_BRANCH` | PM must create branch (git tasks) |
 | `READY_TO_START` | Call `roboco_task_start()` |
 | `EXECUTING` | Work in progress |
 | `REVISION_REQUIRED` | Fix QA/PM feedback |
 
+Note: Git branches are auto-created when you claim the task. No waiting needed.
+
 ## Tool Restrictions
 
 Use `roboco_*` MCP tools, not native Claude tools:

docs/rag/tools/git-tools.md

@@ -27,19 +27,18 @@ log = roboco_git_log(
 
 ## Branch Operations
 
+**Branches are auto-created when tasks are claimed:**
+- Root task claim → `feature/team/ROOT_ID`
+- Subtask claim → `feature/team/ROOT_ID/SUB_ID`
+- Sub-subtask claim → `feature/team/ROOT_ID/SUB_ID/SUBSUB_ID`
+
+No manual branch creation needed.
+
 ```python
 # List branches
 branches = roboco_git_branch_list(project_slug="roboco")
 
-# Create branch (PM only)
+# Checkout branch (if needed)
-roboco_git_create_branch(
-    project_slug="roboco",
-    task_id=task_id,
-    branch_type="feature"  # feature, bug, chore, docs, hotfix
-)
-# Creates: feature/backend/a1b2c3d4
-
-# Checkout branch
 roboco_git_checkout(
     project_slug="roboco",
     branch="feature/backend/a1b2c3d4"

docs/rag/troubleshooting/blocked-tools.md

@@ -44,13 +44,14 @@
 
 See: `roboco_kb_search("task planning workflow")`
 
-## WAITING_FOR_BRANCH
+## Parent Branch Required
 
-**Symptom:** Can't start git task, state is WAITING_FOR_BRANCH
+**Symptom:** Can't claim subtask, error "Parent task must be claimed first"
 
-**Cause:** PM hasn't created the branch yet
+**Cause:** Parent task hasn't been claimed yet, so it has no branch
 
 **Solution:**
-1. Message PM to create branch
+1. Parent task must be claimed first (branch auto-creates on claim)
-2. Or escalate: `roboco_task_escalate(task_id, "Need branch")`
+2. Then subtask can be claimed (its branch forks from parent's)
-3. Wait for branch_name to be set on task
+
+Note: Branches are auto-created hierarchically. No manual creation needed.

docs/rag/troubleshooting/task-errors.md

@@ -27,14 +27,14 @@
 
 **Causes**:
 1. Task not claimed by you
-2. For git tasks: branch not created yet
+2. Task in wrong status
-3. Task in wrong status
 
 **Solutions**:
 - Claim first: `roboco_task_claim(task_id)`
-- Wait for PM to create branch (git tasks)
 - Check current status
 
+Note: Git branches are auto-created on claim, no waiting needed.
+
 ## Cannot Submit for QA
 
 **Error**: "Invalid transition from current status"
@@ -74,16 +74,15 @@ parent_id = task.parent_task_id
 roboco_task_escalate_to_ceo(parent_id, notes="...")
 ```
 
-## Git Task: No Branch
+## Git Task: Parent Branch Required
 
-**Error**: "Branch name required for git tasks"
+**Error**: "Parent task must be claimed first to create its branch"
 
-**Cause**: PM hasn't created branch yet
+**Cause**: Trying to claim a subtask when parent task hasn't been claimed yet
 
-**Solution**: Wait for PM or ask PM to create branch:
+**Solution**: Parent task must be claimed first. Branches are auto-created hierarchically:
-```python
+1. Parent is claimed → parent branch created
-roboco_git_create_branch(project_slug, task_id, "feature")
+2. Then subtask can be claimed → subtask branch created (forked from parent)
-```
 
 ## Task Has Incomplete Subtasks
 

docs/rag/workflows/git-branch-naming.md

@@ -11,4 +11,4 @@ Branches follow: `{type}/{team}/{root-uuid}[/{subtask-uuid}[/{sub-sub-uuid}]]`
 - Subtask: `feature/backend/550e8400.../6ba7b810...`
 - Sub-sub: `feature/backend/550e8400.../6ba7b810.../f47ac10b...`
 
-PM creates branches via `roboco_git_create_branch()`.
+**Branches are auto-created when tasks are claimed.** No manual creation needed.

docs/rag/workflows/task-planning.md

@@ -12,7 +12,6 @@ CLAIM → PLAN → START → EXECUTE
 | State | Meaning | Next Step |
 |-------|---------|-----------|
 | `NEEDS_PLAN` | Task claimed, no plan yet | Call `roboco_task_plan()` |
-| `WAITING_FOR_BRANCH` | Plan approved, git task needs branch | PM creates branch |
 | `READY_TO_START` | Plan approved, ready to work | Call `roboco_task_start()` |
 | `EXECUTING` | Work in progress | Continue development |
 | `REVISION_REQUIRED` | QA/PM requested changes | Reclaim and fix |
@@ -38,12 +37,12 @@ Calling `roboco_task_start()` without a plan returns:
 - Message: "Cannot start without a plan"
 - Hint: Submit plan first
 
-## Git Tasks Need Branch
+## Git Tasks
 
 For tasks with `requires_git=True`:
-1. Submit plan
+- **Branches are auto-created when you claim the task**
-2. PM creates branch: `roboco_git_create_branch()`
+- Root tasks: branch created from default branch (main/master)
-3. Task gets `branch_name` field set
+- Subtasks: branch forked from parent's branch
-4. Then you can call `roboco_task_start()`
+- No manual `roboco_git_create_branch()` needed
 
-If no branch: workflow state = `WAITING_FOR_BRANCH`
+Hierarchical branch naming: `feature/team/ROOT_ID/SUB_ID/SUBSUB_ID`

roboco/api/routes/git.py

@@ -48,8 +48,10 @@ from roboco.api.schemas.git import (
     GitStatusResponse,
 )
 from roboco.exceptions import GitCommandError, GitTimeoutError
+from roboco.models.base import TaskStatus
 from roboco.services.base import NotFoundError, ServiceError, ValidationError
 from roboco.services.git import get_git_service
+from roboco.services.project import get_project_service
 from roboco.services.task import get_task_service
 from roboco.services.work_session import get_work_session_service
 from roboco.utils.converters import require_uuid
@@ -356,17 +358,55 @@ async def create_branch(
     """
     git_service = get_git_service(db)
     task_service = get_task_service(db)
+    project_service = get_project_service(db)
 
     try:
         workspace = await git_service.get_workspace(data.project_slug, agent.agent_id)
 
+        task_uuid = UUID(data.task_id)
+        task = await task_service.get(task_uuid)
+        project = await project_service.get_by_slug(data.project_slug)
+
+        project_cell: str | None = None
+        if project and project.assigned_cell:
+            project_cell = (
+                project.assigned_cell.value
+                if hasattr(project.assigned_cell, "value")
+                else str(project.assigned_cell)
+            )
+
+        task_team: str | None = None
+        if task and task.team:
+            task_team = (
+                task.team.value if hasattr(task.team, "value") else str(task.team)
+            )
+
+        if project_cell == "fullstack":
+            effective_team = task_team or "cross"
+            team_for_branch = f"{data.project_slug}/{effective_team}"
+        elif project_cell:
+            team_for_branch = project_cell
+        elif task_team:
+            team_for_branch = task_team
+        else:
+            team_for_branch = "cross"
+
         branch_name, created_from = await git_service.create_branch(
-            workspace, agent.team or "unknown", data
+            workspace, team_for_branch, data
         )
 
-        # Store branch name on task
-        task_uuid = UUID(data.task_id)
         await task_service.update(task_uuid, branch_name=branch_name)
+
+        children = await task_service.get_subtasks(task_uuid)
+        for child in children:
+            if (
+                child.status == TaskStatus.BACKLOG
+                and child.requires_git
+                and not child.branch_name
+            ):
+                child_uuid = UUID(str(child.id))
+                await task_service.update(child_uuid, branch_name=branch_name)
+
         await db.commit()
     except ServiceError as e:
         raise _translate_error(e) from e

roboco/api/routes/sessions.py

@@ -431,6 +431,7 @@ async def create_session_for_tasks(
     req = SessionForTasksCreate(
         task_ids=data.task_ids,
         channel_slug=data.channel_slug,
+        group_id=data.group_id,
         scope=data.scope,
         relationship_type=rel_type,
     )

roboco/api/routes/tasks.py

@@ -1638,11 +1638,28 @@ async def escalate_task(
     delivery_service = get_notification_delivery_service(db)
     await delivery_service.deliver(require_uuid(notification.id))
 
+    # CRITICAL FIX: Set task to BLOCKED to stop orchestrator from respawning dev
+    # Previously used PENDING which could still cause respawn loops.
+    # BLOCKED ensures task is truly paused until PM unblocks it.
+    task.assigned_to = target_agent.id
+    task.status = TaskStatus.BLOCKED  # Blocked until PM addresses it
+
+    # Add escalation note for context
+    existing_notes = task.dev_notes or ""
+    escalation_note = (
+        f"\n\n[ESCALATED] From {agent_record.slug} to {target_slug}\n"
+        f"Reason: {data.reason}"
+    )
+    task.dev_notes = existing_notes + escalation_note
+
+    await db.flush()
+
     await db.commit()
 
     msg = (
-        f"Task escalated to {target_slug}. "
+        f"Task escalated to {target_slug} and set to BLOCKED. "
-        "They will be notified and can reassign or provide guidance."
+        f"PM will receive notification and must call roboco_task_unblock() "
+        "to provide guidance or reassign."
     )
     return EscalateResponse(
         status="escalated",

roboco/api/schemas/sessions.py

@@ -73,6 +73,10 @@ class SessionForTasksCreateRequest(BaseModel):
 
     task_ids: list[UUID] = Field(..., min_length=1)
     channel_slug: str
+    group_id: UUID | None = Field(
+        default=None,
+        description="Optional group ID to place session under",
+    )
     scope: SessionScope = Field(
         default=SessionScope.CELL,
         description="Scope level: initiative (Main PM), cell (Cell PM), task (dev)",

roboco/db/tables.py

@@ -143,7 +143,7 @@ class TaskTable(Base):
     )
     requires_git: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
 
-    # Project & Branch (set by PM during setup)
+    # Project & Branch (branch auto-created on claim)
     project_id: Mapped[UUID | None] = mapped_column(
         UUID(as_uuid=True),
         ForeignKey("projects.id", ondelete="SET NULL"),

roboco/enforcement/__init__.py

@@ -40,12 +40,15 @@ from roboco.enforcement.notification_perms import (
 from roboco.enforcement.task_lifecycle import (
     ROLE_RESTRICTED_TRANSITIONS,
     VALID_TRANSITIONS,
+    GitContext,
+    GitRequirementError,
     TaskLifecycleError,
     can_agent_transition,
     get_valid_transitions,
     is_active_state,
     is_terminal_state,
     is_waiting_state,
+    validate_git_requirements,
     validate_task_transition,
 )
 from roboco.enforcement.task_ownership import (
@@ -61,6 +64,8 @@ __all__ = [
     "ROLE_RESTRICTED_TRANSITIONS",
     "VALID_TRANSITIONS",
     "ChannelAccessDeniedError",
+    "GitContext",
+    "GitRequirementError",
     "JournalAccessDeniedError",
     "NotificationPermissionError",
     "TaskClaimContext",
@@ -77,6 +82,7 @@ __all__ = [
     "is_terminal_state",
     "is_waiting_state",
     "validate_channel_access",
+    "validate_git_requirements",
     "validate_journal_access",
     "validate_notification_permission",
     "validate_task_claim",

roboco/enforcement/task_lifecycle.py

@@ -297,7 +297,7 @@ def validate_git_requirements(
         Requires pr_number to be set (PR exists)
 
     - claimed → in_progress (git tasks):
-        Should have branch_name set (PM created the branch)
+        Should have branch_name set (auto-created on claim)
 
     Args:
         current_status: Current task status
@@ -361,8 +361,8 @@ def validate_git_requirements(
             requirement="branch_name",
             message=(
                 "Cannot start work: no branch assigned to this task. "
-                "For git tasks, PM must create a branch first using "
+                "Branches are auto-created on claim. If missing, either "
-                "roboco_git_create_branch(project_slug, task_id, branch_type)."
+                "re-claim the task or check if parent task needs claiming first."
             ),
         )
 

roboco/mcp/git/git_server.py

@@ -316,7 +316,7 @@ def create_git_mcp_server(agent_id: str) -> FastMCP:
     Tools are registered based on role:
     - All agents: read-only tools (status, log, branch list, diff)
     - Developers: commit, push, create PR
-    - PMs: create branch, checkout, merge PR
+    - PMs: checkout, merge PR (branches auto-created on claim)
 
     Args:
         agent_id: The agent identifier (e.g., "be-dev-1")

roboco/mcp/schemas/__init__.py

@@ -174,8 +174,10 @@ class TaskCreateInput(BaseModel):
     - Use roboco_project_list() to see available projects
     """
 
-    title: str = Field(..., description="Task title")
+    title: str = Field(..., min_length=1, max_length=200, description="Task title")
-    description: str = Field(..., description="Task description")
+    description: str = Field(
+        ..., min_length=10, description="Task description (min 10 chars)"
+    )
     acceptance_criteria: list[str] = Field(
         ..., min_length=1, description="At least one acceptance criterion"
     )
@@ -192,6 +194,13 @@ class TaskCreateInput(BaseModel):
         default=True,
         description="Whether task requires git. If True, project_slug is required.",
     )
+    task_type: str = Field(
+        default="code",
+        description=(
+            "Task type: code (git work), documentation, research, planning, "
+            "design, administrative. For subtasks, inherits from parent if not set."
+        ),
+    )
     parent_task_id: str | None = Field(
         default=None, description="Parent task for subtasks"
     )
@@ -269,6 +278,10 @@ class SessionCreateForTasksInput(BaseModel):
         ..., min_length=1, description="Task IDs to link to the session"
     )
     channel_slug: str = Field(..., description="Channel where session is created")
+    group_id: str | None = Field(
+        default=None,
+        description="Group ID to place session under (from roboco_group_create)",
+    )
     scope: str = Field(
         default="cell",
         description="Scope level: initiative (Main PM), cell (Cell PM), task (dev)",

roboco/mcp/tasks/handlers/_helpers.py

@@ -139,6 +139,20 @@ async def validate_task_claimable(
     Special case: If an agent is already assigned to a pending task (PM assigned
     it directly to them), they can claim it to transition to 'claimed' status.
     """
+    # ENFORCEMENT: Main PM must delegate code tasks, not execute them
+    task_type = task.get("task_type", "code")
+    if agent_role == "main_pm" and task_type == "code":
+        return format_error_response(
+            "PM_CANNOT_EXECUTE_CODE",
+            "Main PM cannot claim code tasks. You coordinate, not execute.",
+            {"task_type": task_type, "your_role": agent_role},
+            hint=(
+                "Create a subtask for the appropriate Cell PM (be-pm, fe-pm, ux-pm) "
+                "using roboco_task_create(parent_task_id=this_task_id, team='backend', "
+                "assigned_to='be-pm'). Then activate it with roboco_task_activate()."
+            ),
+        )
+
     task_status = task.get("status")
     claimable_statuses = {
         # QA: pending (direct QA tasks from PM) or awaiting_qa (normal workflow)

roboco/mcp/tasks/handlers/claim.py

@@ -6,8 +6,6 @@ Handler for claiming tasks.
 
 from typing import Any
 
-from starlette import status
-
 from roboco.agents_config import get_agent_role
 from roboco.mcp.tasks import format_task_response
 from roboco.mcp.tasks.handlers._helpers import (
@@ -58,51 +56,6 @@ async def _is_pre_assigned_to_agent(
     return agent_uuid is not None and str(assigned_to) == agent_uuid
 
 
-async def _validate_branch_hierarchy(
-    client: ApiClient, task: dict[str, Any]
-) -> dict[str, Any] | None:
-    """Walk up task hierarchy ensuring all git-enabled ancestors have branches.
-
-    Returns error dict if hierarchy is incomplete, None if valid.
-    Branch hierarchy must be created from root down:
-    - Main PM creates root branch
-    - Cell PM creates subtask branch
-    - Developer works on subsubtask branch
-    """
-    current_parent_id = task.get("parent_task_id")
-    ancestors_checked: list[str] = []
-
-    while current_parent_id:
-        parent_resp = await client.get(f"/tasks/{current_parent_id}")
-        if parent_resp.status_code != status.HTTP_200_OK:
-            # Can't fetch parent - may have been deleted, allow to proceed
-            break
-
-        parent = parent_resp.json()
-        parent_title = parent.get("title", str(current_parent_id)[:8])
-        ancestors_checked.append(parent_title)
-
-        # Check if parent is a git task missing its branch
-        if parent.get("requires_git") and not parent.get("branch_name"):
-            return format_error_response(
-                "BRANCH_HIERARCHY_INCOMPLETE",
-                f"Parent task '{parent_title}' has no branch yet.",
-                {
-                    "missing_branch_task": str(current_parent_id),
-                    "ancestors_checked": ancestors_checked,
-                },
-                hint=(
-                    "Branch hierarchy must be created from root down. "
-                    "Main PM creates root branch, Cell PM creates subtask branch. "
-                    "Ask your PM to create the parent branch first."
-                ),
-            )
-
-        current_parent_id = parent.get("parent_task_id")
-
-    return None
-
-
 async def _execute_claim(
     client: ApiClient, task_id: str, agent_id: str
 ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
@@ -122,45 +75,105 @@ async def _execute_claim(
 async def _validate_git_requirements(
     client: ApiClient, task: dict[str, Any], task_id: str
 ) -> dict[str, Any] | None:
-    """Validate git-related requirements. Returns error or None."""
+    """Validate git requirements for hierarchical branching.
+
+    Rules:
+    - Must have project_id (always)
+    - Root tasks: branch auto-created from default branch
+    - Subtasks: need parent to have branch (for forking)
+    """
     if not task.get("requires_git"):
         return None
 
-    # Branch must exist for git tasks
+    # Must have project
-    if not task.get("branch_name"):
+    if not task.get("project_id"):
         return format_error_response(
-            "BRANCH_REQUIRED",
+            "PROJECT_REQUIRED",
-            "Cannot claim git task - PM must create branch first.",
+            "Git tasks require a project for branch creation.",
-            {"task_id": task_id, "requires_git": True},
+            {"task_id": task_id},
-            hint="PM should call roboco_git_create_branch() before developer can claim",
+            hint="Assign a project to this task.",
         )
 
-    # Validate full branch hierarchy for git tasks with parent
+    # If already has branch, good to go
-    if task.get("parent_task_id"):
+    if task.get("branch_name"):
-        return await _validate_branch_hierarchy(client, task)
+        return None
+
+    # For subtasks, parent must have branch (so we can fork from it)
+    parent_id = task.get("parent_task_id")
+    if parent_id:
+        parent_resp = await client.get(f"/tasks/{parent_id}")
+        if parent_resp.ok:
+            parent = parent_resp.json()
+            if parent.get("requires_git") and not parent.get("branch_name"):
+                return format_error_response(
+                    "PARENT_BRANCH_REQUIRED",
+                    "Parent task must be claimed first to create its branch.",
+                    {
+                        "task_id": task_id,
+                        "parent_id": parent_id,
+                        "parent_title": parent.get("title", "")[:50],
+                    },
+                    hint="Parent task must be claimed first. Branch is auto-created.",
+                )
+
+    # Root task or parent has branch - will auto-create on claim
+    return None
+
+
+async def _validate_sibling_sequence(
+    client: ApiClient, task: dict[str, Any]
+) -> dict[str, Any] | None:
+    """Ensure earlier sequence siblings are complete before claiming.
+
+    If task has sequence=2, then sequence=1 siblings must be completed first.
+    """
+    parent_id = task.get("parent_task_id")
+    if not parent_id:
+        return None  # Root task - no siblings
+
+    my_sequence = task.get("sequence", 0)
+    if my_sequence == 0:
+        return None  # First in sequence, no prior siblings
+
+    # Get all sibling tasks (same parent)
+    siblings_resp = await client.get(f"/tasks?parent_task_id={parent_id}")
+    if not siblings_resp.ok:
+        return None
+
+    siblings = siblings_resp.json()
+
+    # Check if any earlier sequence siblings are NOT complete
+    terminal_statuses = ["completed", "cancelled"]
+    for sibling in siblings:
+        sibling_seq = sibling.get("sequence", 0)
+        sibling_status = sibling.get("status")
+
+        # Skip self
+        if sibling.get("id") == task.get("id"):
+            continue
+
+        # If sibling has lower sequence and isn't complete, block
+        if sibling_seq < my_sequence and sibling_status not in terminal_statuses:
+            return format_error_response(
+                "SEQUENCE_ORDER_VIOLATION",
+                f"Cannot claim - task with sequence {sibling_seq} must complete first.",
+                {
+                    "your_sequence": my_sequence,
+                    "blocking_task_id": sibling.get("id"),
+                    "blocking_sequence": sibling_seq,
+                    "blocking_status": sibling_status,
+                    "blocking_title": sibling.get("title", "")[:50],
+                },
+                hint=f"Wait for '{sibling.get('title', '')[:30]}' to complete.",
+            )
 
     return None
 
 
-async def handle_task_claim(
+async def _run_claim_validations(
-    client: ApiClient, task_id: str, agent_id: str
+    client: ApiClient, task: dict[str, Any], task_id: str, agent_id: str
-) -> dict[str, Any]:
+) -> dict[str, Any] | None:
-    """Handle task claiming.
+    """Run all claim validations. Returns error dict or None if all pass."""
-
-    Flow:
-    1. Fetch the task first
-    2. Check if it's pre-assigned to this agent (PM assigned directly)
-    3. If pre-assigned, skip blocking check for THIS task
-    4. Otherwise, run full blocking check
-    5. Validate task is claimable for this role
-    6. Execute claim
-    """
-    # Fetch task first - we need to check if it's pre-assigned
-    task, error = await fetch_task_or_error(client, task_id)
-    if error:
-        return error
-    assert task is not None
-
     # Check if this task is pre-assigned to the agent
     is_pre_assigned = await _is_pre_assigned_to_agent(task, agent_id, client)
 
@@ -174,10 +187,38 @@ async def handle_task_claim(
     if error := await validate_task_claimable(task, agent_role, agent_id, client):
         return error
 
-    # Validate git requirements (branch exists, hierarchy valid)
+    # Validate git requirements (project exists, parent has branch if subtask)
     if error := await _validate_git_requirements(client, task, task_id):
         return error
 
+    # TODO: Re-enable when sequence workflow is refined
+    # Validate sibling sequence order (earlier sequence must complete first)
+    # if error := await _validate_sibling_sequence(client, task):
+    #     return error
+
+    return None
+
+
+async def handle_task_claim(
+    client: ApiClient, task_id: str, agent_id: str
+) -> dict[str, Any]:
+    """Handle task claiming.
+
+    Flow:
+    1. Fetch the task first
+    2. Run validations (blocking tasks, role, git, sequence)
+    3. Execute claim
+    """
+    # Fetch task first
+    task, error = await fetch_task_or_error(client, task_id)
+    if error:
+        return error
+    assert task is not None
+
+    # Run all validations
+    if error := await _run_claim_validations(client, task, task_id, agent_id):
+        return error
+
     # Execute the claim
     claimed_task, error = await _execute_claim(client, task_id, agent_id)
     if error:

roboco/mcp/tasks/handlers/management.py

@@ -311,9 +311,29 @@ async def _fetch_and_validate_project(
 
 
 def _build_task_payload(
-    input_data: TaskCreateInput, project_id: str | None = None
+    input_data: TaskCreateInput,
+    project_id: str | None = None,
+    parent_task: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Build task creation payload from input data."""
+    """Build task creation payload from input data.
+
+    For subtasks, inherits task_type from parent if not explicitly set to non-default.
+    If assigning to a PM, task_type defaults to 'planning' (PMs coordinate, not code).
+    """
+    # Determine task_type
+    task_type = input_data.task_type
+
+    # If assigning to a PM, default to 'planning' (PMs don't code)
+    if input_data.assigned_to and task_type == "code":
+        assignee_role = get_agent_role(input_data.assigned_to)
+        if assignee_role in ("cell_pm", "main_pm"):
+            task_type = "planning"
+
+    # For subtasks, inherit from parent if still at default
+    if parent_task and task_type == "code":
+        parent_type = parent_task.get("task_type", "code")
+        task_type = parent_type
+
     payload: dict[str, Any] = {
         "title": input_data.title,
         "description": input_data.description,
@@ -325,9 +345,12 @@ def _build_task_payload(
         "status": input_data.status,  # Always included, defaults to "backlog"
         "sequence": input_data.sequence,  # Task ordering (lower = first)
         "requires_git": input_data.requires_git,
+        "task_type": task_type,
     }
     if input_data.parent_task_id:
         payload["parent_task_id"] = input_data.parent_task_id
+        # NOTE: Branch is auto-created on claim, not inherited at creation time.
+        # Each task in the hierarchy gets its own branch forked from parent's branch.
     if input_data.dependency_ids:
         payload["dependency_ids"] = input_data.dependency_ids
     if project_id:
@@ -365,6 +388,40 @@ async def _validate_task_create_inputs(
     if error := _validate_cell_pm_team(agent_id, input_data.team):
         return None, error
 
+    # ENFORCEMENT: All tasks require meaningful descriptions
+    # Main PM delegating to Cell PM needs to explain the work clearly
+    description = input_data.description.strip()
+    min_root_description_len = 30
+    min_subtask_description_len = 50
+
+    is_subtask = input_data.parent_task_id is not None
+    min_len = min_subtask_description_len if is_subtask else min_root_description_len
+
+    if is_subtask and len(description) < min_len:
+        # Subtasks need more detail (Cell PM → Dev)
+        return None, format_error_response(
+            "SUBTASK_DESCRIPTION_REQUIRED",
+            f"Subtasks MUST have detailed descriptions "
+            f"(min {min_len} chars, got {len(description)}). "
+            "Explain what to do, why, and expected outcome.",
+            {
+                "parent_task_id": input_data.parent_task_id,
+                "description_length": len(description),
+            },
+        )
+    elif not is_subtask and len(description) < min_len:
+        # Root tasks also need context (Main PM → Cell PM)
+        return None, format_error_response(
+            "TASK_DESCRIPTION_REQUIRED",
+            f"Tasks MUST have meaningful descriptions "
+            f"(min {min_len} chars, got {len(description)}). "
+            "Main PM must explain what Cell PM should accomplish.",
+            {
+                "description_length": len(description),
+                "guidance": "Include: goal, context, and acceptance criteria.",
+            },
+        )
+
     # Validate assignee BEFORE creating task
     assignee = input_data.assigned_to
     if assignee:
@@ -384,7 +441,41 @@ async def handle_task_create(
     if error:
         return error
 
-    payload = _build_task_payload(input_data, project_id)
+    # Fetch parent task for inheritance (task_type, etc.)
+    parent_task = None
+    if input_data.parent_task_id:
+        parent_resp = await client.get(f"/tasks/{input_data.parent_task_id}")
+        if parent_resp.ok:
+            parent_task = parent_resp.json()
+
+    # GUARDRAIL: If parent requires git, child must also require git (hierarchy)
+    if parent_task and parent_task.get("requires_git") and not input_data.requires_git:
+        return format_error_response(
+            "GIT_INHERITANCE_REQUIRED",
+            "Subtasks of git tasks must also require git.",
+            {
+                "parent_task_id": parent_task["id"],
+                "parent_requires_git": True,
+                "child_requires_git": input_data.requires_git,
+            },
+            hint="Set requires_git=True (or omit it, defaults to True).",
+        )
+
+    # GUARDRAIL: Git subtasks need parent to have branch (for forking)
+    if parent_task and input_data.requires_git:
+        parent_branch = parent_task.get("branch_name")
+        if not parent_branch:
+            return format_error_response(
+                "PARENT_BRANCH_REQUIRED",
+                "Parent task must have a branch before creating git subtasks.",
+                {
+                    "parent_task_id": parent_task["id"],
+                    "parent_status": parent_task.get("status"),
+                },
+                hint="Claim the parent task first. Branch is auto-created on claim.",
+            )
+
+    payload = _build_task_payload(input_data, project_id, parent_task)
 
     try:
         create_resp = await client.post("/tasks", json=payload)
@@ -445,6 +536,57 @@ async def handle_task_assign(
     if validation_error:
         return validation_error
 
+    # ENFORCEMENT: Block Cell PM from directly assigning devs on complex tasks
+    # Cell PM must create subtasks first for medium+ complexity work
+    assignee_role = get_agent_role(input_data.assignee)
+    if role == "cell_pm" and assignee_role == "developer":
+        complexity = task.get("estimated_complexity", "low")
+        if complexity in ("medium", "high", "critical"):
+            # Check if task already has subtasks
+            try:
+                subtasks_resp = await client.get(
+                    f"/tasks/{input_data.task_id}/subtasks"
+                )
+                subtasks = subtasks_resp.json() if subtasks_resp.ok else []
+            except Exception:
+                subtasks = []
+
+            # Also allow if this task IS a subtask (has parent)
+            is_subtask = task.get("parent_task_id") is not None
+
+            if not subtasks and not is_subtask:
+                return format_error_response(
+                    "SUBTASK_REQUIRED",
+                    f"Cannot assign {complexity} complexity task directly to dev. "
+                    "Cell PM must break down the work into subtasks first.",
+                    {
+                        "task_id": task.get("id"),
+                        "complexity": complexity,
+                        "guidance": (
+                            f"Create subtasks with: roboco_task_create("
+                            f"parent_task_id='{task.get('id')}', ...) "
+                            "Then assign each subtask to developers."
+                        ),
+                    },
+                )
+
+    # GUARDRAIL: Git tasks need branch before assigning to developers
+    if task.get("requires_git") and not task.get("branch_name"):
+        if assignee_role == "developer":
+            return format_error_response(
+                "NO_BRANCH_FOR_GIT_TASK",
+                "Git task must have a branch before assigning to developer.",
+                {
+                    "task_id": task.get("id"),
+                    "requires_git": True,
+                    "has_branch": False,
+                },
+                hint=(
+                    "Either claim the task first (creates branch), "
+                    "or create subtasks for developers."
+                ),
+            )
+
     assigned_task, assign_error = await assign_task_to_agent(
         client, input_data.task_id, input_data.assignee
     )
@@ -565,6 +707,57 @@ async def handle_task_escalate(
         )
 
 
+async def _validate_activation_sequence(
+    client: ApiClient, task_id: str
+) -> dict[str, Any] | None:
+    """Ensure parent is in_progress before activating subtask."""
+    task_resp = await client.get(f"/tasks/{task_id}")
+    if not task_resp.ok:
+        return None
+
+    task = task_resp.json()
+    parent_id = task.get("parent_task_id")
+    if not parent_id:
+        return None  # Root task
+
+    parent_resp = await client.get(f"/tasks/{parent_id}")
+    if not parent_resp.ok:
+        return None
+
+    parent = parent_resp.json()
+    parent_status = parent.get("status")
+
+    # Parent must be in_progress or paused (PM has started work)
+    if parent_status not in ["in_progress", "paused"]:
+        return format_error_response(
+            "PARENT_NOT_STARTED",
+            f"Cannot activate - parent task is '{parent_status}'.",
+            {"parent_id": parent_id, "parent_status": parent_status},
+            hint="Call roboco_task_start() on parent task first.",
+        )
+
+    return None
+
+
+def _handle_activate_response(resp: Any, task_id: str) -> dict[str, Any] | None:
+    """Handle API response for activation. Returns error dict or None if success."""
+    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},
+        )
+
+    return None
+
+
 async def handle_task_activate(
     client: ApiClient, task_id: str, agent_id: str
 ) -> dict[str, Any]:
@@ -584,6 +777,11 @@ async def handle_task_activate(
             {"role": get_agent_role(agent_id)},
         )
 
+    # TODO: Re-enable when sequence workflow is refined
+    # Validate parent is started before activating subtask
+    # if error := await _validate_activation_sequence(client, task_id):
+    #     return error
+
     try:
         resp = await client.post(f"/tasks/{task_id}/activate")
     except Exception as e:
@@ -592,19 +790,8 @@ async def handle_task_activate(
             f"Failed to connect to API: {type(e).__name__}",
         )
 
-    if resp.is_status(status.HTTP_404_NOT_FOUND):
+    if error := _handle_activate_response(resp, task_id):
-        return format_error_response("NOT_FOUND", f"Task {task_id} not found")
+        return error
-
-    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 = (

roboco/mcp/tasks/handlers/sessions.py

@@ -97,6 +97,8 @@ async def handle_session_create_for_tasks(
         "scope": input_data.scope,
         "relationship_type": input_data.relationship_type,
     }
+    if input_data.group_id:
+        payload["group_id"] = input_data.group_id
 
     try:
         resp = await client.post("/sessions/for-tasks", json=payload)

roboco/mcp/tasks/handlers/work.py

@@ -120,7 +120,7 @@ async def _safe_checkout(
             "CHECKOUT_FAILED",
             f"Failed to checkout branch '{branch_name}'",
             {"status_code": checkout_resp.status_code, "detail": checkout_resp.text},
-            hint="Branch may not exist yet. Ask PM to create it.",
+            hint="Branch may not exist. Re-claim or check if parent needs claiming.",
         )
 
     # Success - no error

roboco/models/README.md

@@ -169,7 +169,7 @@ class Task:
     task_type: TaskType              # code, documentation, research, planning, design, administrative
     requires_git: bool               # Whether git workflow applies
 
-    # Project & Branch (set by PM during setup)
+    # Project & Branch (branch auto-created on claim)
     project_id: UUID | None
     branch_name: str | None
     work_session_id: UUID | None

roboco/models/base.py

@@ -68,7 +68,8 @@ class Team(str, Enum):
     BACKEND = "backend"
     FRONTEND = "frontend"
     UX_UI = "ux_ui"
-    MAIN_PM = "main_pm"  # Main PM level - cross-cell coordination
+    FULLSTACK = "fullstack"
+    MAIN_PM = "main_pm"
     BOARD = "board"
     MARKETING = "marketing"
 

roboco/models/session.py

@@ -198,6 +198,9 @@ class SessionForTasksCreate(RobocoBase):
 
     task_ids: list[UUID] = Field(..., min_length=1, description="Tasks to link")
     channel_slug: str = Field(..., description="Channel where session is created")
+    group_id: UUID | None = Field(
+        default=None, description="Optional group ID to place session under"
+    )
     scope: SessionScope = Field(
         default=SessionScope.CELL,
         description="Session scope level for context loading strategy",

roboco/models/task.py

@@ -176,7 +176,7 @@ class Task(TimestampMixin):
         default=True, description="Whether this task requires git workflow"
     )
 
-    # Project & Branch (set by PM during setup)
+    # Project & Branch (branch auto-created on claim)
     project_id: UUID | None = Field(
         default=None, description="Project this task works on"
     )

roboco/runtime/orchestrator.py

@@ -18,7 +18,7 @@ import json
 import os
 import shutil
 import tempfile
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any, ClassVar
 
@@ -910,7 +910,7 @@ class AgentOrchestrator:
         # Role-based permissions enforced at handler level:
         # - All agents: read-only (status, log, diff, branch list)
         # - Developers: commit, push, create PR
-        # - PMs: create branch, checkout, merge PR
+        # - PMs: checkout, merge PR (branches auto-created on claim)
         mcp_servers["roboco-git"] = {
             "command": "uv",
             "args": [
@@ -1449,6 +1449,113 @@ Start by:
             return False
         return self._instances[agent_id].state == AgentState.ACTIVE
 
+    async def _validate_task_for_spawn(
+        self,
+        client: httpx.AsyncClient,
+        task: dict,
+        agent_slug: str,
+    ) -> str | None:
+        """
+        Validate task is ready for agent spawn.
+
+        Returns None if valid, or error message if task cannot proceed.
+        This prevents spawning agents on tasks that are missing prerequisites.
+        """
+        task_id = task.get("id")
+        if not task_id:
+            return "Task missing ID"
+        min_description_len = 10
+
+        # VALIDATION 1: Check description is not empty/trivial
+        description = (task.get("description") or "").strip()
+        if len(description) < min_description_len:
+            return (
+                f"Task {task_id} has inadequate description ({len(description)} chars)"
+            )
+
+        # VALIDATION 2: For git tasks, check project + parent branch requirements
+        requires_git = task.get("requires_git", False)
+        if requires_git:
+            # Must have project
+            if not task.get("project_id"):
+                await self._auto_block_task(
+                    client, task_id, "Git task needs project_id"
+                )
+                return f"Task {task_id} needs project"
+
+            # For subtasks, parent must have branch (for forking)
+            parent_id = task.get("parent_task_id")
+            if parent_id:
+                parent_resp = await client.get(f"{self._api_url}/tasks/{parent_id}")
+                if parent_resp.is_success:
+                    parent = parent_resp.json()
+                    if parent.get("requires_git") and not parent.get("branch_name"):
+                        await self._auto_block_task(
+                            client,
+                            task_id,
+                            "Parent task must be claimed first to create its branch",
+                        )
+                        return f"Task {task_id} waiting for parent branch"
+
+            # Root task or parent has branch - branch will auto-create on claim
+            logger.info(
+                "Git task ready for hierarchical branch creation", task_id=task_id
+            )
+
+        # VALIDATION 3: Check complexity vs subtasks for devs
+        from roboco.agents_config import get_agent_role
+
+        agent_role = get_agent_role(agent_slug)
+        if agent_role == "developer":
+            complexity = task.get("estimated_complexity", "low")
+            parent_task_id = task.get("parent_task_id")
+
+            if complexity in ("medium", "high", "critical") and not parent_task_id:
+                # Check if this task has subtasks
+                try:
+                    resp = await client.get(f"{self._api_url}/tasks/{task_id}/subtasks")
+                    subtasks = resp.json() if resp.is_success else []
+                except Exception:
+                    subtasks = []
+
+                if not subtasks:
+                    await self._auto_block_task(
+                        client,
+                        task_id,
+                        f"Task complexity is {complexity} but no subtasks. "
+                        "Cell PM must break down work first.",
+                    )
+                    return (
+                        f"Task {task_id} is {complexity} complexity "
+                        "without subtasks - Cell PM must break it down"
+                    )
+
+        return None  # All validations passed
+
+    async def _auto_block_task(
+        self, client: httpx.AsyncClient, task_id: str, reason: str
+    ) -> None:
+        """Auto-block a task that cannot proceed due to missing prerequisites."""
+        try:
+            await client.patch(
+                f"{self._api_url}/tasks/{task_id}",
+                json={
+                    "status": "blocked",
+                    "dev_notes": f"[AUTO-BLOCKED] {reason}",
+                },
+            )
+            logger.info(
+                "Auto-blocked task with missing prerequisites",
+                task_id=task_id,
+                reason=reason,
+            )
+        except Exception as e:
+            logger.error(
+                "Failed to auto-block task",
+                task_id=task_id,
+                error=str(e),
+            )
+
     def _select_agent_for_cell(self, cell: str, role: str) -> str | None:
         """
         Select the best available agent for a cell and role.
@@ -1546,7 +1653,7 @@ Start by:
     ) -> list[dict[str, Any]]:
         """Fetch notifications by type."""
         params: dict[str, Any] = {
-            "type": notification_type,
+            "type_filter": notification_type,
             "pending_ack_only": str(unacknowledged).lower(),
         }
         try:
@@ -1638,39 +1745,49 @@ Start by:
 
     def _classify_task_routing(self, task: dict[str, Any]) -> str:
         """
-        Classify a task for routing based on team, complexity, and keywords.
+        Classify a task for routing based on task_type, team, complexity, and keywords.
 
         Returns one of: "board", "main_pm", "cell_pm", "dev", "marketing"
         """
         team = task.get("team")
+        task_type = task.get("task_type", "code")
+        cell_teams = ("backend", "frontend", "ux_ui")
+        result: str | None = None
 
-        # Explicit team assignment takes precedence
+        # Task type takes precedence for non-code work
-        if team in self._TEAM_ROUTING_MAP:
+        if task_type in ("planning", "research", "administrative"):
-            return self._TEAM_ROUTING_MAP[team]
+            result = "cell_pm" if team in cell_teams else "main_pm"
+        elif task_type == "design" and team not in ("backend", "frontend"):
+            result = "cell_pm"
+        elif task_type == "documentation" and not task.get("requires_git"):
+            result = "cell_pm" if team in cell_teams else "main_pm"
+        elif team in self._TEAM_ROUTING_MAP:
+            result = self._TEAM_ROUTING_MAP[team]
 
-        # For cell teams, use keyword/complexity analysis
+        if result:
+            return result
+
+        # Keyword/complexity analysis for code tasks without explicit routing
         title = (task.get("title") or "").lower()
         description = (task.get("description") or "").lower()
         text = f"{title} {description}"
         complexity = task.get("estimated_complexity", "medium").lower()
 
-        # Board-level keywords → Board
         if self._has_board_keywords(text):
             return "board"
 
-        # Cross-cell keywords (e.g., "all teams") → Main PM (regardless of complexity)
+        needs_main_pm = (
-        if self._has_cross_cell_keywords(text):
+            self._has_cross_cell_keywords(text)
+            or complexity in ("high", "critical")
+            or not team
+            or team == "all"
+        )
+        if needs_main_pm:
             return "main_pm"
 
-        # High complexity or cross-team → Main PM
-        if complexity in ("high", "critical") or not team or team == "all":
-            return "main_pm"
-
-        # PM keywords or medium complexity → Cell PM
         if self._has_pm_keywords(text) or complexity == "medium":
             return "cell_pm"
 
-        # Low complexity, single team → Direct to dev
         return "dev"
 
     # Team to PM mapping for routing
@@ -1958,6 +2075,9 @@ Start now: roboco_task_get("{task_id}")
             # Scheduled dispatchers
             await self._dispatch_audit_work(client)
 
+            # Proactive enforcement - detect and block stuck tasks
+            await self._detect_stuck_tasks(client)
+
     # =========================================================================
     # SMART DISPATCHER - TASK-BASED DISPATCHERS
     # =========================================================================
@@ -2254,6 +2374,21 @@ Begin with step 1: roboco_task_get("{task_id}")
             # For pending tasks that ARE already assigned (by PM),
             # spawn the assigned agent with the appropriate prompt
             if agent_slug and not self._is_agent_active(agent_slug):
+                # PRE-SPAWN VALIDATION: Check task readiness before spawning
+                # This prevents spawning agents on tasks that can't proceed
+                validation_issue = await self._validate_task_for_spawn(
+                    client, task, agent_slug
+                )
+                if validation_issue:
+                    # Log and skip - don't spawn on invalid tasks
+                    logger.warning(
+                        "Skipping spawn due to validation failure",
+                        task_id=task["id"],
+                        agent=agent_slug,
+                        reason=validation_issue,
+                    )
+                    continue
+
                 await self.spawn_agent(
                     agent_id=agent_slug,
                     task_id=task["id"],
@@ -2561,6 +2696,98 @@ Begin with step 1: roboco_task_get("{task_id}")
         # TODO: Add scheduled periodic audits
         # Check last audit time, spawn if overdue
 
+    async def _detect_stuck_tasks(self, client: httpx.AsyncClient) -> None:
+        """
+        Detect and auto-block tasks that are stuck.
+
+        This is a proactive enforcement mechanism that finds tasks which
+        have been pending without progress and have prerequisite issues.
+        Runs every dispatcher cycle but only takes action on truly stuck tasks.
+
+        CEO-approved timeout: 10 minutes
+        """
+        STUCK_THRESHOLD_MINUTES = 10  # CEO-approved threshold
+
+        tasks = await self._fetch_tasks(client, "pending")
+
+        for task in tasks:
+            age = self._get_task_age(task)
+            if age is None or age < timedelta(minutes=STUCK_THRESHOLD_MINUTES):
+                continue
+
+            issues = self._check_stuck_conditions(task)
+            issues.extend(await self._check_dev_subtask_issue(client, task))
+
+            if issues:
+                task_id = task.get("id")
+                if not task_id:
+                    continue
+                age_mins = int(age.total_seconds() // 60)
+                reason = f"Task stuck for {age_mins} minutes: " + ", ".join(issues)
+                await self._auto_block_task(client, task_id, reason)
+                logger.warning(
+                    "Auto-blocked stuck task",
+                    task_id=task_id,
+                    age_minutes=age_mins,
+                    issues=issues,
+                )
+
+    def _get_task_age(self, task: dict[str, Any]) -> timedelta | None:
+        """Parse task created_at and return age, or None if unparseable."""
+        created_at_str = task.get("created_at")
+        if not created_at_str:
+            return None
+        try:
+            if created_at_str.endswith("Z"):
+                created_at_str = created_at_str[:-1] + "+00:00"
+            created_at = datetime.fromisoformat(created_at_str)
+            if created_at.tzinfo is None:
+                created_at = created_at.replace(tzinfo=UTC)
+            return datetime.now(UTC) - created_at
+        except (ValueError, TypeError):
+            return None
+
+    _MIN_DESCRIPTION_LEN = 10
+
+    def _check_stuck_conditions(self, task: dict[str, Any]) -> list[str]:
+        """Check for common stuck conditions (git, description)."""
+        issues: list[str] = []
+        if task.get("requires_git") and not task.get("branch_name"):
+            issues.append("Git task missing branch_name")
+        description = (task.get("description") or "").strip()
+        if len(description) < self._MIN_DESCRIPTION_LEN:
+            issues.append("Empty or inadequate description")
+        return issues
+
+    async def _check_dev_subtask_issue(
+        self, client: httpx.AsyncClient, task: dict[str, Any]
+    ) -> list[str]:
+        """Check if complex dev task is missing subtasks."""
+        from roboco.agents_config import get_agent_role
+
+        assigned_to = task.get("assigned_to")
+        if not assigned_to:
+            return []
+
+        agent_slug = self._resolve_agent_slug(assigned_to)
+        if not agent_slug or get_agent_role(agent_slug) != "developer":
+            return []
+
+        complexity = task.get("estimated_complexity", "low")
+        is_low_complexity = complexity not in ("medium", "high", "critical")
+        if is_low_complexity or task.get("parent_task_id"):
+            return []
+
+        try:
+            resp = await client.get(f"{self._api_url}/tasks/{task.get('id')}/subtasks")
+            subtasks = resp.json() if resp.is_success else []
+        except Exception:
+            subtasks = []
+
+        if not subtasks:
+            return [f"{complexity} complexity task without subtasks"]
+        return []
+
     async def _dispatch_a2a_work(self, client: httpx.AsyncClient) -> None:
         """
         Dispatch A2A (Agent-to-Agent) requests to target agents.
@@ -2597,16 +2824,12 @@ Begin with step 1: roboco_task_get("{task_id}")
         self,
         status: str,
         has_plan: bool,
-        requires_git: bool,
-        branch_name: str | None,
     ) -> str:
         """Determine developer workflow state from task attributes.
 
         Args:
             status: Task status (claimed, in_progress, needs_revision, etc.)
             has_plan: Whether task has a plan submitted
-            requires_git: Whether task requires git workflow
-            branch_name: Branch name if git task (PM creates this)
 
         Returns:
             Workflow state string (NEEDS_PLAN, READY_TO_START, EXECUTING, etc.)
@@ -2625,8 +2848,6 @@ Begin with step 1: roboco_task_get("{task_id}")
         if status == "claimed":
             if not has_plan:
                 return "NEEDS_PLAN"
-            if requires_git and not branch_name:
-                return "WAITING_FOR_BRANCH"
             return "READY_TO_START"
 
         return status.upper()
@@ -2657,20 +2878,6 @@ Call roboco_task_plan("{task_id}", {{
 }})
 
 You CANNOT call roboco_task_start() until plan is submitted.
-""",
-            "WAITING_FOR_BRANCH": """## BLOCKED: Waiting for Branch
-
-Your plan is approved, but this is a git task and no branch has been created yet.
-
-The PM must create a branch for you using:
-`roboco_git_create_branch(project_slug, task_id, branch_type)`
-
-**What to do:**
-1. Send a message to your PM requesting branch creation
-2. Or escalate: `roboco_task_escalate(task_id, "Need branch created for git task")`
-3. Wait for notification that branch is ready
-
-You CANNOT call roboco_task_start() until branch_name is set on the task.
 """,
             "READY_TO_START": f"""## NEXT STEP: Start Work
 
@@ -2716,11 +2923,7 @@ Run quality checks and verify against acceptance criteria:
 
         # Determine workflow state based on task attributes
         has_plan = bool(task.get("plan"))
-        requires_git = task.get("requires_git", False)
+        workflow_state = self._get_workflow_state(status, has_plan)
-        branch_name = task.get("branch_name")
-        workflow_state = self._get_workflow_state(
-            status, has_plan, requires_git, branch_name
-        )
         instructions = self._get_workflow_instructions(workflow_state, task_id)
 
         return f"""You have been assigned a development task.

roboco/services/git.py

@@ -367,8 +367,7 @@ class GitService(BaseService):
             if not result.stdout.strip():
                 raise ValidationError(
                     f"Parent branch '{base_branch}' does not exist on remote. "
-                    f"The parent task's branch must be created and pushed first. "
+                    f"The parent task must be claimed first (claim creates branch)."
-                    f"Ensure Main PM/Cell PM created their branches before this task."
                 )
 
         # Create and push branch

roboco/services/messaging.py

@@ -665,11 +665,19 @@ class MessagingService(BaseService):
         if not channel:
             raise NotFoundError(f"Channel '{req.channel_slug}' not found")
 
-        # Get first group in channel (or create default)
+        # Use provided group_id or fall back to first group in channel
-        groups = await self.list_groups_in_channel(cast("UUID", channel.id))
+        if req.group_id:
-        if not groups:
+            group_result = await self.session.execute(
-            raise ValueError(f"No groups found in channel '{req.channel_slug}'")
+                select(GroupTable).where(GroupTable.id == req.group_id)
-        group = groups[0]
+            )
+            group = group_result.scalar_one_or_none()
+            if not group:
+                raise NotFoundError(f"Group '{req.group_id}' not found")
+        else:
+            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(

roboco/services/task.py

@@ -21,7 +21,9 @@ from roboco.db.tables import (
     WorkSessionTable,
 )
 from roboco.enforcement import (
+    GitContext,
     TaskOwnershipError,
+    validate_git_requirements,
     validate_task_ownership,
     validate_task_transition,
 )
@@ -157,7 +159,8 @@ class TaskService(BaseService):
         Validate and set task status with lifecycle enforcement.
 
         This is the single point of truth for status changes. All transitions
-        are validated against VALID_TRANSITIONS and ROLE_RESTRICTED_TRANSITIONS.
+        are validated against VALID_TRANSITIONS, ROLE_RESTRICTED_TRANSITIONS,
+        and git requirements.
 
         Args:
             task: The task to update
@@ -166,6 +169,7 @@ class TaskService(BaseService):
 
         Raises:
             TaskLifecycleError: If transition is invalid or role not permitted
+            GitRequirementError: If git requirements not met
         """
         current = (
             task.status.value if isinstance(task.status, TaskStatus) else task.status
@@ -175,6 +179,17 @@ class TaskService(BaseService):
         # Validate the transition (raises TaskLifecycleError if invalid)
         validate_task_transition(current, target, agent_role)
 
+        # Validate git requirements (raises GitRequirementError if not met)
+        if task.requires_git:
+            git_ctx = GitContext(
+                requires_git=True,
+                docs_complete=bool(task.docs_complete),
+                pr_created=bool(task.pr_created),
+                pr_number=task.pr_number,
+                branch_name=str(task.branch_name) if task.branch_name else None,
+            )
+            validate_git_requirements(current, target, git_ctx)
+
         # Apply the status change
         task.status = new_status
         self.log.info(
@@ -211,6 +226,10 @@ class TaskService(BaseService):
             status=req.status if req.status else TaskStatus.PENDING,
             sequence=req.sequence,  # Task ordering within siblings
             dependency_ids=req.dependency_ids,  # Task IDs that must complete first
+            # Git configuration - CRITICAL: These must be passed through
+            task_type=req.task_type,
+            requires_git=req.requires_git,
+            project_id=req.project_id,
         )
         self.session.add(task)
         await self.session.flush()
@@ -325,6 +344,16 @@ class TaskService(BaseService):
                 "before activating."
             )
 
+        # ENFORCEMENT: Git tasks require project_id before activation
+        if task.requires_git and not task.project_id:
+            raise ValueError(
+                f"Cannot activate task '{task.title}' - requires git but no project. "
+                "Fix: (1) Re-create task with project_slug, OR "
+                "(2) Set requires_git=False if git not needed."
+            )
+
+        # NOTE: Git branch is auto-created on claim, not required at activation
+
         # Transition to PENDING
         task.status = TaskStatus.PENDING
         await self.session.flush()
@@ -336,6 +365,120 @@ class TaskService(BaseService):
         )
         return task
 
+    async def _ensure_branch_for_git_task(
+        self,
+        task: TaskTable,
+        agent_id: UUID,
+    ) -> str:
+        """Auto-create hierarchical branch for git tasks. Raises on failure.
+
+        Strategy:
+        - If branch exists: return it
+        - If no project: raise error
+        - Create NEW branch (hierarchical name built by build_branch_name)
+        - Branch created from parent's branch (or default if root)
+
+        Raises:
+            ValueError: If branch cannot be created (mandatory for git tasks)
+        """
+        if not task.requires_git:
+            raise ValueError("Task does not require git")
+
+        if task.branch_name:
+            return str(task.branch_name)
+
+        if not task.project_id:
+            raise ValueError(
+                "Git task requires project_id to create branch. "
+                "Assign a project before claiming."
+            )
+
+        return await self._auto_create_branch(task, agent_id)
+
+    async def _auto_create_branch(
+        self,
+        task: TaskTable,
+        agent_id: UUID,
+    ) -> str:
+        """Create hierarchical branch for git task. Raises on failure.
+
+        Branch naming (via build_branch_name):
+        - Root: feature/team/ROOT_ID
+        - Subtask: feature/team/ROOT_ID/SUB_ID
+        - Sub-subtask: feature/team/ROOT_ID/SUB_ID/SUBSUB_ID
+
+        Parent branch resolution:
+        - Subtask: uses parent task's branch_name
+        - Root: uses project's default branch (main/master)
+
+        Raises:
+            ValueError: If branch cannot be created
+        """
+        from roboco.api.schemas.git import GitCreateBranchRequest
+        from roboco.services.git import get_git_service
+        from roboco.services.project import get_project_service
+
+        git_service = get_git_service(self.session)
+        project_service = get_project_service(self.session)
+
+        project = await project_service.get(UUID(str(task.project_id)))
+        if not project:
+            raise ValueError(f"Project {task.project_id} not found")
+
+        parent_branch: str | None = None
+        if task.parent_task_id:
+            parent_task = await self.get(UUID(str(task.parent_task_id)))
+            if parent_task and parent_task.branch_name:
+                parent_branch = str(parent_task.branch_name)
+            else:
+                raise ValueError(
+                    "Parent task must be claimed first. "
+                    "Subtasks fork from parent branch (auto-created on claim)."
+                )
+
+        workspace = await git_service.get_workspace(project.slug, agent_id)
+
+        project_cell = (
+            project.assigned_cell.value
+            if project.assigned_cell and hasattr(project.assigned_cell, "value")
+            else str(project.assigned_cell)
+            if project.assigned_cell
+            else None
+        )
+        task_team = (
+            task.team.value if task.team and hasattr(task.team, "value") else None
+        )
+
+        if project_cell == "fullstack":
+            team = f"{project.slug}/{task_team or 'cross'}"
+        elif project_cell:
+            team = project_cell
+        elif task_team:
+            team = task_team
+        else:
+            team = "cross"
+
+        request = GitCreateBranchRequest(
+            task_id=str(task.id),
+            project_slug=project.slug,
+            branch_type="feature",
+            agent_id=str(agent_id),
+            parent_branch=parent_branch,
+        )
+
+        branch_name, _ = await git_service.create_branch(workspace, team, request)
+
+        task.branch_name = branch_name
+        await self.session.flush()
+
+        self.log.info(
+            "Auto-created hierarchical branch",
+            task_id=str(task.id),
+            branch_name=branch_name,
+            parent_branch=parent_branch or "default",
+        )
+        return branch_name
+
     async def get(self, task_id: UUID) -> TaskTable | None:
         """Get a task by ID."""
         result = await self.session.execute(
@@ -544,7 +687,14 @@ class TaskService(BaseService):
 
         await self.session.flush()
 
+        # Auto-create hierarchical branch for git tasks (mandatory - raises on failure)
+        # Must happen BEFORE work session creation (work session needs branch_name)
+        if task.requires_git and not task.branch_name:
+            await self._ensure_branch_for_git_task(task, agent_id)
+            await self.session.refresh(task)
+
         # Create work session for git-enabled tasks claimed by developers
+        # (now branch exists, so work session can be created)
         await self._create_work_session_if_needed(task, agent_id, agent_role)
 
         # Trigger proactive knowledge injection (fire-and-forget)
@@ -624,7 +774,7 @@ class TaskService(BaseService):
         Only creates a session if:
         - Task requires git (requires_git=True)
         - Task has a project_id set
-        - Task has a branch_name set (PM created the branch)
+        - Task has a branch_name set (auto-created on claim)
         - Agent is a developer (not QA/Documenter claiming for review)
 
         Args:
@@ -2126,6 +2276,16 @@ class TaskService(BaseService):
             )
             return None
 
+        # ENFORCEMENT: Git tasks must have PR created before CEO approval
+        if task.requires_git and not task.pr_number:
+            self.log.warning(
+                "Cannot escalate to CEO - git task has no PR",
+                task_id=str(task_id),
+                requires_git=task.requires_git,
+                pr_created=task.pr_created,
+            )
+            return None
+
         # Store escalation notes
         if notes:
             existing_context = task.quick_context or ""