Added tasks sequence

CLAUDE.md

@@ -70,8 +70,8 @@ pnpm test
 | Cache/Queue | Redis |
 | Container Runtime | Docker + Docker Compose |
 | Cloud LLM | Claude API (claude-opus-4-5-20251101) |
-| Local LLM | Ollama (qwen3:8b for HyDE/RAG) |
-| Embeddings | BAAI/bge-base-en-v1.5 (768 dim) |
+| Local LLM | Ollama (gemma3:4b for HyDE/RAG) |
+| Embeddings | embeddinggemma:300m (768 dim) |
 | Frontend | React / Next.js (future) |
 
 ## Multi-Agent Workspace Structure
@@ -327,11 +327,67 @@ ROBOCO_RAG_USE_HYBRID_SEARCH=true
 
 # AI/LLM
 ROBOCO_DEFAULT_LLM_MODEL=claude-opus-4-5-20251101
-ROBOCO_DEFAULT_EMBEDDING_MODEL=BAAI/bge-base-en-v1.5
-ROBOCO_LOCAL_LLM_MODEL=qwen3:8b
-ROBOCO_LOCAL_LLM_BASE_URL=http://192.168.50.111:11434/v1
+ROBOCO_DEFAULT_EMBEDDING_MODEL=embeddinggemma:300m
+ROBOCO_LOCAL_LLM_MODEL=gemma3:4b
+ROBOCO_LOCAL_LLM_BASE_URL=http://roboco-ollama:11434/v1
+ROBOCO_OLLAMA_BASE_URL=http://roboco-ollama:11434
 ```
 
+## Docker Deployment
+
+### Container Architecture
+
+The system runs as Docker Compose services:
+
+| Service | Purpose | Healthcheck |
+|---------|---------|-------------|
+| `postgres` | PostgreSQL + pgvector | `pg_isready` |
+| `redis` | Cache, sessions, event bus | `redis-cli ping` |
+| `ollama` | Local LLM + embeddings | `ollama list` |
+| `ollama-init` | Pulls models on startup | One-shot |
+| `orchestrator` | API + agent spawner | Depends on all above |
+
+### Startup Sequence
+
+The startup order is critical due to dependencies:
+
+```
+postgres ──┐
+redis ─────┼──> ollama ──> ollama-init ──> orchestrator
+           │        │            │
+           │        │            └── Pulls embeddinggemma:300m, gemma3:4b
+           │        └── Healthcheck: ollama list
+           └── Healthcheck: pg_isready, redis-cli ping
+```
+
+**Important timing notes:**
+1. `ollama-init` pulls models (~30s for embedding model, ~2min for LLM)
+2. Orchestrator waits for models before starting
+3. FastAPI lifespan indexes documents using Ollama (~30-60s)
+4. Orchestrator polls `/health` until API is ready before starting dispatcher
+
+### Ollama Configuration
+
+Ollama provides two APIs:
+- `/v1/*` - OpenAI-compatible API (for LLM chat/completion)
+- `/api/*` - Native Ollama API (for embeddings, model management)
+
+The embedder uses `/api/embed` endpoint with the `embeddinggemma:300m` model.
+
+**Environment variables for Docker:**
+```bash
+ROBOCO_LOCAL_LLM_BASE_URL=http://roboco-ollama:11434/v1    # OpenAI-compat
+ROBOCO_OLLAMA_BASE_URL=http://roboco-ollama:11434          # Native API
+```
+
+### Common Issues
+
+| Symptom | Cause | Fix |
+|---------|-------|-----|
+| `404 /api/embed` | Model not pulled | Check `docker logs roboco-ollama-init` |
+| `All connection attempts failed` | API not ready | Orchestrator starts before FastAPI lifespan completes |
+| Healthcheck failing | Wrong endpoint | Use `ollama list` not `curl` |
+
 ## Blueprint Reference
 
 The complete system design is documented in `HOMELAB_TEAM_V0.md`, which contains:

README.md

@@ -90,8 +90,8 @@ ROBOCO_WORKSPACES_ROOT=/data/workspaces
 ROBOCO_WORKSPACE_AUTO_CLONE=true
 
 # RAG/LLM
-ROBOCO_LOCAL_LLM_BASE_URL=http://localhost:11434/v1
-ROBOCO_LOCAL_LLM_MODEL=qwen3:8b
+ROBOCO_LOCAL_LLM_BASE_URL=http://roboco-ollama:11434/v1
+ROBOCO_LOCAL_LLM_MODEL=gemma3:4b
 ```
 
 ## Multi-Agent Workspace Structure
@@ -176,8 +176,8 @@ uv run mypy roboco/
 | Vector Store | pgvector (via piragi) |
 | Cache/Queue | Redis |
 | RAG Library | piragi |
-| Embeddings | BAAI/bge-base-en-v1.5 (sentence-transformers) |
-| Local LLM | Ollama (qwen3:8b) |
+| Embeddings | embeddinggemma:300m (sentence-transformers) |
+| Local LLM | Ollama (gemma3:4b) |
 | Cloud LLM | Claude API (Anthropic) |
 | Package Manager | uv |
 

alembic/versions/001_initial_schema.py

@@ -323,6 +323,8 @@ def upgrade() -> None:
             postgresql.ARRAY(postgresql.UUID(as_uuid=True)),
             server_default="{}",
         ),
+        # Ordering for sibling tasks (lower = first)
+        sa.Column("sequence", sa.Integer(), nullable=False, server_default="0"),
         sa.Column(
             "created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()
         ),

roboco/api/routes/git.py

@@ -51,6 +51,7 @@ from roboco.api.schemas.git import (
 from roboco.config import settings
 from roboco.services.project import get_project_service
 from roboco.services.workspace import WorkspaceError, get_workspace_service
+from roboco.utils.converters import require_uuid
 
 router = APIRouter()
 
@@ -433,16 +434,18 @@ async def create_commit(
         task_service = get_task_service(db)
         await task_service.add_commit(
             task_id=task_uuid,
-            commit_hash=commit_hash,
+            hash=commit_hash,
             message=data.message,  # Store original message without prefix
-            author_id=agent.agent_id,
+            agent_id=agent.agent_id,
         )
 
         # If task has a work session, add commit there too
         task = await task_service.get(task_uuid)
         if task and task.work_session_id:
             work_session_service = get_work_session_service(db)
-            await work_session_service.add_commit(task.work_session_id, commit_hash)
+            await work_session_service.add_commit(
+                require_uuid(task.work_session_id), commit_hash
+            )
 
         await db.commit()
     except Exception:

roboco/api/routes/notifications.py

@@ -28,6 +28,7 @@ from roboco.enforcement import (
     validate_notification_permission,
 )
 from roboco.services.notification_delivery import get_notification_delivery_service
+from roboco.utils.converters import require_uuid
 
 router = APIRouter()
 
@@ -161,7 +162,7 @@ async def send_notification(
 
     # Deliver notification via Redis Streams for real-time push
     delivery_service = get_notification_delivery_service(db)
-    await delivery_service.deliver(notification.id)
+    await delivery_service.deliver(require_uuid(notification.id))
 
     return notification_to_response(notification, agent_id)
 

roboco/api/routes/tasks.py

@@ -106,6 +106,8 @@ async def create_task(
         target_date=data.target_date,
         estimated_complexity=data.estimated_complexity,
         status=data.status,
+        sequence=data.sequence,  # Task ordering within siblings
+        dependency_ids=data.dependency_ids,  # Dependencies for claim filtering
     )
     task = await service.create(req)
     await db.commit()
@@ -678,7 +680,7 @@ async def soft_block_task(
             )
 
             delivery_service = get_notification_delivery_service(db)
-            await delivery_service.deliver_notification(notification)
+            await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -742,7 +744,7 @@ async def unblock_task(
         )
 
         delivery_service = get_notification_delivery_service(db)
-        await delivery_service.deliver(notification.id)
+        await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -1066,7 +1068,7 @@ async def docs_complete(
             )
 
             delivery_service = get_notification_delivery_service(db)
-            await delivery_service.deliver(notification.id)
+            await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -1152,7 +1154,7 @@ async def submit_for_pm_review(
             )
 
             delivery_service = get_notification_delivery_service(db)
-            await delivery_service.deliver(notification.id)
+            await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -1377,7 +1379,7 @@ async def escalate_to_ceo(
         )
 
         delivery_service = get_notification_delivery_service(db)
-        await delivery_service.deliver(notification.id)
+        await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -1476,7 +1478,7 @@ async def ceo_reject_task(
         )
 
         delivery_service = get_notification_delivery_service(db)
-        await delivery_service.deliver(notification.id)
+        await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
     return task_to_response(task)
@@ -1582,7 +1584,7 @@ async def escalate_task(
     from roboco.services.notification_delivery import get_notification_delivery_service
 
     delivery_service = get_notification_delivery_service(db)
-    await delivery_service.deliver(notification.id)
+    await delivery_service.deliver(require_uuid(notification.id))
 
     await db.commit()
 

roboco/api/schemas/project.py

@@ -49,7 +49,7 @@ class ProjectResponse(BaseModel):
     created_by: UUID
     is_active: bool
     created_at: datetime
-    updated_at: datetime
+    updated_at: datetime | None = None
 
     class Config:
         """Pydantic config."""

roboco/api/schemas/tasks.py

@@ -224,6 +224,7 @@ class TaskResponse(BaseModel):
     # Status
     status: TaskStatus
     priority: int
+    sequence: int  # Order number within siblings
 
     # Ownership
     team: Team
@@ -528,6 +529,7 @@ def task_to_response(task: "TaskTable") -> TaskResponse:
         acceptance_criteria=task.acceptance_criteria or [],
         status=task.status,
         priority=task.priority,
+        sequence=task.sequence,
         team=task.team,
         created_by=require_uuid(task.created_by),
         assigned_to=to_python_uuid(task.assigned_to),

roboco/api/schemas/work_session.py

@@ -54,7 +54,7 @@ class WorkSessionResponse(BaseModel):
 
     # Timestamps
     created_at: datetime
-    updated_at: datetime
+    updated_at: datetime | None = None
 
     class Config:
         """Pydantic config."""

roboco/db/tables.py

@@ -187,6 +187,11 @@ class TaskTable(Base):
         ARRAY(UUID(as_uuid=True)), default=list
     )
 
+    # Ordering (for sibling tasks under the same parent)
+    sequence: Mapped[int] = mapped_column(
+        Integer, default=0, nullable=False, index=True
+    )
+
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False

roboco/mcp/schemas/__init__.py

@@ -163,7 +163,12 @@ class SendNotificationInput(BaseModel):
 
 
 class TaskCreateInput(BaseModel):
-    """Input for creating a task (PM only)."""
+    """Input for creating a task (PM only).
+
+    ORDERING: Use sequence and dependency_ids to control task execution order.
+    - sequence: Lower numbers execute first (1, 2, 3...)
+    - dependency_ids: Tasks that must complete before this one can be claimed
+    """
 
     title: str = Field(..., description="Task title")
     description: str = Field(..., description="Task description")
@@ -183,6 +188,15 @@ class TaskCreateInput(BaseModel):
         default="backlog",
         description="Status: 'backlog' (default) or 'pending' (ready for work)",
     )
+    # Task ordering - IMPORTANT for subtask sequencing
+    sequence: int = Field(
+        default=0,
+        description="Execution order within siblings (lower = first). E.g., 1, 2, 3",
+    )
+    dependency_ids: list[str] = Field(
+        default_factory=list,
+        description="Task IDs that must complete before this task can be claimed",
+    )
 
 
 class TaskAssignInput(BaseModel):

roboco/mcp/task_server.py

@@ -634,10 +634,22 @@ def _register_pm_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
         - Only PMs and management can create tasks
         - Cell PMs can only create tasks for their own team
 
+        ORDERING SUBTASKS:
+        When creating multiple subtasks, use sequence and dependency_ids:
+        - sequence: Lower numbers execute first (1, 2, 3...)
+        - dependency_ids: Task IDs that must complete first
+
+        Example for 3 ordered subtasks:
+        1. "Fix bug" (sequence=1, no deps)
+        2. "Add feature" (sequence=2, depends on #1)
+        3. "Write tests" (sequence=3, depends on #1 and #2)
+
         Args:
-            data: TaskCreateInput with title, description, acceptance_criteria,
-                  team, and optional parent_task_id, assigned_to, priority, status.
-                  Use status="backlog" for subtasks needing session setup.
+            data: TaskCreateInput with:
+                - title, description, acceptance_criteria, team (required)
+                - parent_task_id, assigned_to, priority, status (optional)
+                - sequence: Order within siblings (0 = default)
+                - dependency_ids: Task IDs that must complete first
 
         Returns:
             Created task with next step guidance

roboco/mcp/tasks/handlers/_helpers.py

@@ -145,12 +145,18 @@ async def validate_task_claimable(
         "qa": ["pending", "awaiting_qa"],
         # Documenters: pending (direct docs tasks) or awaiting_documentation (workflow)
         "documenter": ["pending", "awaiting_documentation"],
+        # Developers: pending or needs_revision (after QA/CEO rejection)
+        "developer": ["pending", "needs_revision"],
+        # PMs: pending or awaiting_pm_review
+        "cell_pm": ["pending", "awaiting_pm_review"],
+        "main_pm": ["pending", "awaiting_pm_review"],
     }
-    allowed = claimable_statuses.get(agent_role, ["pending"])
+    # Default for unlisted roles is pending + needs_revision (developer-like behavior)
+    allowed = claimable_statuses.get(agent_role, ["pending", "needs_revision"])
 
-    # Special case: agent can claim pending tasks already assigned to them
-    # This handles PM directly assigning tasks to QA/docs agents
-    if task_status == "pending":
+    # Special case: agent can claim tasks already assigned to them
+    # This handles PM directly assigning tasks or needs_revision tasks
+    if task_status in ("pending", "needs_revision"):
         assigned_to = task.get("assigned_to")
         if assigned_to:
             agent_uuid = await resolve_agent_uuid_cached(agent_id, client)

roboco/mcp/tasks/handlers/claim.py

@@ -114,13 +114,78 @@ async def handle_task_claim(
     if claimed_task.get("project_id"):
         project = await get_project_context(client, claimed_task["project_id"])
 
+    # Context-aware guidance based on task state
+    guidance = _build_claim_guidance(claimed_task, task)
+    next_step = "REVIEW" if claimed_task.get("plan") else "PLAN"
+
     return format_task_response(
         claimed_task,
-        "PLAN",
+        next_step,
+        guidance,
+        project=project,
+    )
+
+
+def _build_claim_guidance(claimed_task: dict, original_task: dict) -> str:
+    """Build context-aware guidance based on task's previous state."""
+    original_status = original_task.get("status", "pending")
+    has_plan = claimed_task.get("plan")
+    qa_notes = claimed_task.get("qa_notes")
+    dev_notes = claimed_task.get("dev_notes")
+    checkpoints = claimed_task.get("checkpoints", [])
+    progress_updates = claimed_task.get("progress_updates", [])
+
+    # NEEDS_REVISION: Task was rejected by QA or CEO - READ FEEDBACK FIRST
+    if original_status == "needs_revision":
+        parts = [
+            "⚠️ REVISION REQUIRED - READ EXISTING CONTEXT FIRST!\n",
+            "This task was REJECTED and needs fixes. Before doing anything:\n",
+        ]
+        if qa_notes:
+            parts.append(f"1. READ QA FEEDBACK: {qa_notes[:200]}...\n")
+        if has_plan:
+            plan_steps = has_plan.get("steps", [])
+            completed = sum(1 for s in plan_steps if s.get("completed"))
+            parts.append(
+                f"2. REVIEW EXISTING PLAN: {completed}/{len(plan_steps)} "
+                "steps completed\n"
+            )
+        if dev_notes:
+            parts.append("3. CHECK DEV NOTES for previous work context\n")
+        parts.append(
+            "\nFix the specific issues mentioned, don't restart from scratch.\n"
+            "Call roboco_task_start() to resume work."
+        )
+        return "".join(parts)
+
+    # Task with existing plan (resumed/paused/etc)
+    if has_plan:
+        plan_steps = has_plan.get("steps", [])
+        completed = sum(1 for s in plan_steps if s.get("completed"))
+        parts = [
+            "📋 EXISTING PLAN FOUND - REVIEW BEFORE CONTINUING!\n",
+            f"Plan progress: {completed}/{len(plan_steps)} steps completed\n",
+        ]
+        if checkpoints:
+            latest_cp = checkpoints[-1]
+            parts.append(f"Last checkpoint: {latest_cp.get('state_summary', 'N/A')}\n")
+        if progress_updates:
+            latest_prog = progress_updates[-1]
+            parts.append(
+                f"Last progress: {latest_prog.get('percentage', 0)}% - "
+                f"{latest_prog.get('message', 'N/A')}\n"
+            )
+        parts.append(
+            "\nREVIEW the plan and continue from where work stopped.\n"
+            "Call roboco_task_start() to resume work."
+        )
+        return "".join(parts)
+
+    # Fresh task - no plan yet
+    return (
         "Task claimed. NEXT: Call roboco_task_plan() before you can start.\n"
         "1. Read the description and acceptance criteria\n"
         "2. Ask questions if anything is unclear\n"
         "3. Call roboco_task_plan(task_id, approach, steps)\n"
-        "4. Then call roboco_task_start(task_id)",
-        project=project,
+        "4. Then call roboco_task_start(task_id)"
     )

roboco/mcp/tasks/handlers/management.py

@@ -258,9 +258,12 @@ def _build_task_payload(input_data: TaskCreateInput) -> dict[str, Any]:
         "priority": input_data.priority,
         "estimated_complexity": input_data.complexity,
         "status": input_data.status,  # Always included, defaults to "backlog"
+        "sequence": input_data.sequence,  # Task ordering (lower = first)
     }
     if input_data.parent_task_id:
         payload["parent_task_id"] = input_data.parent_task_id
+    if input_data.dependency_ids:
+        payload["dependency_ids"] = input_data.dependency_ids
     return payload
 
 

roboco/models/events.py

@@ -34,6 +34,7 @@ class EventType(str, Enum):
     TASK_QA_PASSED = "task.qa_passed"
     TASK_QA_FAILED = "task.qa_failed"
     TASK_AWAITING_DOCS = "task.awaiting_docs"
+    TASK_ESCALATED_TO_MAIN_PM = "task.escalated_to_main_pm"  # Cell PM → Main PM
     TASK_AWAITING_CEO_APPROVAL = "task.awaiting_ceo_approval"  # Escalated to CEO
     TASK_CEO_APPROVED = "task.ceo_approved"  # CEO approved
     TASK_CEO_REJECTED = "task.ceo_rejected"  # CEO rejected, needs revision
@@ -158,8 +159,16 @@ class OrchestratorAccessProtocol(Protocol):
 
     def get_waiting_agents(self) -> dict[str, "WaitingRecord"]: ...
 
+    def get_running_agents(self) -> set[str]: ...
+
     async def resolve_wait(self, agent_id: str, resolution: dict[str, Any]) -> Any: ...
 
+    async def spawn_agent(
+        self,
+        agent_id: str,
+        initial_prompt: str | None = None,
+    ) -> Any: ...
+
 
 # =============================================================================
 # EVENT CONTEXT (dependency container)

roboco/models/optimal.py

@@ -73,12 +73,15 @@ class IndexConversationParams:
 
 @dataclass
 class IndexJournalEntryParams:
-    """Parameters for indexing a journal entry."""
+    """Parameters for indexing a journal entry.
+
+    Note: entry_id and agent_id can be None for system events (e.g., lifecycle events).
+    """
 
-    entry_id: UUID
-    agent_id: UUID
     content: str
     entry_type: str
+    entry_id: UUID | None = None
+    agent_id: UUID | None = None
     task_id: UUID | None = None
     tags: list[str] | None = None
 
@@ -171,6 +174,7 @@ class IndexStandardParams:
     title: str
     content: str
     language: str | None = None
+    scope: str | None = None
     severity: str = "recommended"
     tags: list[str] | None = None
     source_file: str | None = None

roboco/models/task.py

@@ -279,6 +279,15 @@ class TaskCreate(RobocoBase):
     estimated_complexity: Complexity = Complexity.MEDIUM
     status: TaskStatus | None = None  # PM can set 'backlog' for subtasks needing setup
 
+    # Ordering and dependencies
+    sequence: int = Field(
+        default=0, description="Order within siblings (lower = first)"
+    )
+    dependency_ids: list[UUID] = Field(
+        default_factory=list,
+        description="Task IDs that must complete before this task can be claimed",
+    )
+
     # Git configuration
     task_type: TaskType = TaskType.CODE
     requires_git: bool = True
@@ -333,6 +342,10 @@ class TaskCreateRequest:
     estimated_complexity: Complexity = field(default=Complexity.MEDIUM)
     status: TaskStatus | None = None  # PM can set BACKLOG for subtasks
 
+    # Ordering and dependencies
+    sequence: int = 0  # Order within siblings (lower = first)
+    dependency_ids: list[UUID] = field(default_factory=list)
+
     # Git configuration
     task_type: TaskType = field(default=TaskType.CODE)
     requires_git: bool = True

roboco/runtime/orchestrator.py

@@ -178,6 +178,10 @@ class AgentOrchestrator:
 
         logger.info("Orchestrator stopped")
 
+    def get_running_agents(self) -> set[str]:
+        """Get set of currently running agent IDs."""
+        return set(self._instances.keys())
+
     async def _ensure_agent_image(self, agent_id: str | None = None) -> None:
         """Ensure the agent Docker images are built.
 

roboco/services/a2a.py

@@ -7,7 +7,7 @@ Provides business logic for A2A protocol operations including:
 - Message handling and routing
 """
 
-from typing import Any
+from typing import Any, cast
 from uuid import UUID
 
 import structlog
@@ -517,7 +517,7 @@ class A2AService:
         # Check for explicit target
         target = metadata.get("target_agent")
         if target and target in ALL_AGENTS:
-            return target
+            return cast("str", target)
 
         # Check for skill-based routing
         skill = metadata.get("skill")
@@ -549,7 +549,7 @@ class A2AService:
             return
 
         # Assign task to target agent
-        task.assigned_to = UUID(target_uuid)
+        task.assigned_to = cast("Any", UUID(target_uuid))
         await self.session.flush()
 
         # Publish A2A request event for routing

roboco/services/messaging.py

@@ -46,6 +46,7 @@ from roboco.models.session import (
     SessionTaskRelationshipType,
 )
 from roboco.services.base import BaseService, ConflictError, NotFoundError
+from roboco.utils.converters import require_uuid, to_python_uuid
 
 # =============================================================================
 # MESSAGING SERVICE
@@ -811,7 +812,7 @@ class MessagingService(BaseService):
             await self.session.flush()
 
             # Deliver via Redis Streams -> WebSocket
-            await delivery_service.deliver(notification.id)
+            await delivery_service.deliver(require_uuid(notification.id))
 
             self.log.debug(
                 "Mention notification sent",
@@ -829,10 +830,10 @@ class MessagingService(BaseService):
             await optimal.index_conversation(
                 IndexConversationParams(
                     content=message.content,
-                    channel_id=message.channel_id,
-                    session_id=message.session_id,
-                    agent_id=message.agent_id,
-                    task_id=message.task_id,
+                    channel_id=require_uuid(message.channel_id),
+                    session_id=require_uuid(message.session_id),
+                    agent_id=require_uuid(message.agent_id),
+                    task_id=to_python_uuid(message.task_id),
                     message_type=message.type.value if message.type else None,
                 )
             )

roboco/services/notification.py

@@ -12,6 +12,7 @@ from roboco.db.base import get_db_context
 from roboco.db.tables import NotificationTable
 from roboco.models import NotificationPriority, NotificationType
 from roboco.models.notification import CreateNotificationParams
+from roboco.utils.converters import require_uuid
 
 logger = structlog.get_logger()
 
@@ -191,7 +192,7 @@ class NotificationService:
             )
 
             delivery_service = get_notification_delivery_service(db)
-            await delivery_service.deliver(notification.id)
+            await delivery_service.deliver(require_uuid(notification.id))
 
             await db.commit()
 

roboco/services/optimal.py

@@ -396,8 +396,8 @@ class OptimalService:
             title=f"Message in {params.channel_id or 'channel'}",
             preview=params.content[:500] if params.content else None,
             metadata={
-                "channel_id": params.channel_id,
-                "session_id": params.session_id,
+                "channel_id": str(params.channel_id) if params.channel_id else None,
+                "session_id": str(params.session_id) if params.session_id else None,
                 "agent_id": str(params.agent_id) if params.agent_id else None,
             },
         )
@@ -425,7 +425,7 @@ class OptimalService:
             title=f"Journal: {params.entry_type or 'entry'}",
             preview=params.content[:500] if params.content else None,
             metadata={
-                "entry_id": params.entry_id,
+                "entry_id": str(params.entry_id) if params.entry_id else None,
                 "agent_id": str(params.agent_id) if params.agent_id else None,
                 "entry_type": params.entry_type,
                 "tags": params.tags,

roboco/services/task.py

@@ -26,10 +26,11 @@ from roboco.enforcement import (
     validate_task_transition,
 )
 from roboco.events import Event, EventType, get_event_bus
-from roboco.models.base import TaskStatus, Team
+from roboco.models.base import AgentRole, TaskStatus, Team
 from roboco.models.task import TaskCreateRequest
 from roboco.models.work_session import WorkSessionStatus
 from roboco.services.base import BaseService
+from roboco.utils.converters import require_uuid, to_python_uuid
 
 # UUID format constants for validation
 _UUID_LENGTH = 36  # Standard UUID string length
@@ -207,6 +208,8 @@ class TaskService(BaseService):
             target_date=req.target_date,
             estimated_complexity=req.estimated_complexity,
             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
         )
         self.session.add(task)
         await self.session.flush()
@@ -822,12 +825,12 @@ class TaskService(BaseService):
             for content, ltype in learnings:
                 await learning_svc.record_learning(
                     RecordLearningParams(
-                        agent_id=assigned_to or agent_id or UUID(int=0),
+                        agent_id=to_python_uuid(assigned_to) or agent_id or UUID(int=0),
                         agent_role="developer",
                         content=content,
                         learning_type=ltype,
                         scope=scope,
-                        task_id=task_id,
+                        task_id=to_python_uuid(task_id),
                         tags=["auto-extracted", task_team or "general"],
                     )
                 )
@@ -1159,7 +1162,8 @@ class TaskService(BaseService):
             task_team: Team for categorization
             details: Additional event details
         """
-        from roboco.services.optimal import IndexType, get_optimal_service
+        from roboco.models.optimal import IndexJournalEntryParams
+        from roboco.services.optimal import get_optimal_service
 
         try:
             optimal = await get_optimal_service()
@@ -1168,18 +1172,18 @@ class TaskService(BaseService):
             # Build content for indexing
             content = f"[{event_type.upper()}] {task_title}"
             if details:
-                content += f"\n{details}"
+                content += f"\nDetails: {details}"
 
             # Index to journals for lifecycle tracking
-            await optimal.ingest(
-                index_type=IndexType.JOURNALS,
-                content=content,
-                doc_id=f"lifecycle-{task_id}-{event_type}",
-                task_id=task_id,
-                project=task_team.value if task_team else "default",
-                entry_type="lifecycle",
-                event_type=event_type,
-                **details,
+            await optimal.index_journal_entry(
+                IndexJournalEntryParams(
+                    content=content,
+                    entry_id=None,  # Will be auto-generated
+                    agent_id=None,  # System event, no specific agent
+                    entry_type=f"lifecycle_{event_type}",
+                    task_id=task_id,
+                    tags=[event_type, task_team.value if task_team else "default"],
+                )
             )
             self.log.debug(
                 "Indexed lifecycle event",
@@ -1361,7 +1365,9 @@ class TaskService(BaseService):
             "what_needed": what_needed,
         }
         bg_task = asyncio.create_task(
-            self._index_blocker_background(task.id, task.team, blocker_info)
+            self._index_blocker_background(
+                require_uuid(task.id), task.team, blocker_info
+            )
         )
         self._background_tasks.add(bg_task)
         bg_task.add_done_callback(self._background_tasks.discard)
@@ -1539,11 +1545,11 @@ class TaskService(BaseService):
         # Index positive QA review (fire-and-forget)
         bg_task = asyncio.create_task(
             self._index_qa_review_background(
-                task.id,
+                require_uuid(task.id),
                 task.quick_context,
                 True,
                 notes or "Passed QA review",
-                qa_agent_id,
+                to_python_uuid(qa_agent_id),
             )
         )
         self._background_tasks.add(bg_task)
@@ -1605,11 +1611,11 @@ class TaskService(BaseService):
         # Index negative QA review (fire-and-forget)
         review_task = asyncio.create_task(
             self._index_qa_review_background(
-                task.id,
+                require_uuid(task.id),
                 task.quick_context,
                 False,
                 notes,
-                qa_agent_id,
+                to_python_uuid(qa_agent_id),
             )
         )
         self._background_tasks.add(review_task)
@@ -1617,7 +1623,9 @@ class TaskService(BaseService):
 
         # Index issues as error patterns (fire-and-forget)
         error_task = asyncio.create_task(
-            self._index_qa_errors_background(task.id, task.title, task.team, notes)
+            self._index_qa_errors_background(
+                require_uuid(task.id), task.title, task.team, notes
+            )
         )
         self._background_tasks.add(error_task)
         error_task.add_done_callback(self._background_tasks.discard)
@@ -1738,7 +1746,7 @@ class TaskService(BaseService):
         # Index documentation artifacts (fire-and-forget)
         if task.documents:
             bg_task = asyncio.create_task(
-                self._index_docs_background(task.id, task.documents)
+                self._index_docs_background(require_uuid(task.id), task.documents)
             )
             self._background_tasks.add(bg_task)
             bg_task.add_done_callback(self._background_tasks.discard)
@@ -1899,7 +1907,7 @@ class TaskService(BaseService):
         )
         return task
 
-    async def complete(
+    async def complete(  # noqa: PLR0911, PLR0912, PLR0915
         self,
         task_id: UUID,
         agent_id: UUID | None = None,
@@ -1909,9 +1917,10 @@ class TaskService(BaseService):
         """
         Mark task as completed (PM only).
 
-        Two completion paths:
-        1. Developer work: task must be in AWAITING_PM_REVIEW (went through QA/Docs)
-        2. PM's own task: task can be IN_PROGRESS if assigned to the completing PM
+        Approval hierarchy:
+        1. Cell PM reviews → reassigns to Main PM (same awaiting_pm_review state)
+        2. Main PM reviews leaf task → completes
+        3. Main PM reviews parent task (all descendants terminal) → escalates to CEO
 
         PM Override for cancelled subtasks:
         Use force_with_cancelled=True with justification to complete despite
@@ -1931,6 +1940,20 @@ class TaskService(BaseService):
         if not task:
             return None
 
+        # Get the completing agent's role
+        completing_agent_role = None
+        if agent_id:
+            agent_result = await self.session.execute(
+                select(AgentTable).where(AgentTable.id == agent_id)
+            )
+            completing_agent = agent_result.scalar_one_or_none()
+            if completing_agent and completing_agent.role:
+                completing_agent_role = (
+                    completing_agent.role.value
+                    if hasattr(completing_agent.role, "value")
+                    else str(completing_agent.role)
+                )
+
         # Check if PM is completing their own task (assigned to them)
         is_own_task = agent_id and task.assigned_to == agent_id
 
@@ -1968,6 +1991,50 @@ class TaskService(BaseService):
             )
             return None
 
+        # APPROVAL HIERARCHY: Cell PM → Main PM → CEO
+        if task.status == TaskStatus.AWAITING_PM_REVIEW:
+            # Cell PM reviewed - escalate to Main PM
+            if completing_agent_role == "cell_pm":
+                main_pm_result = await self.session.execute(
+                    select(AgentTable).where(AgentTable.role == AgentRole.MAIN_PM)
+                )
+                main_pm = main_pm_result.scalar_one_or_none()
+                if main_pm:
+                    task.assigned_to = cast("Any", main_pm.id)
+                    await self.session.flush()
+
+                    # Emit event for Main PM notification
+                    await self._emit_task_event(
+                        EventType.TASK_ESCALATED_TO_MAIN_PM,
+                        task_id,
+                        {
+                            "main_pm_id": str(main_pm.id),
+                            "cell_pm_id": str(agent_id) if agent_id else None,
+                        },
+                    )
+
+                    self.log.info(
+                        "Cell PM approved - escalating to Main PM",
+                        task_id=str(task_id),
+                        main_pm_id=str(main_pm.id),
+                    )
+                    return task  # Return without completing - Main PM needs to review
+                else:
+                    self.log.warning(
+                        "No Main PM found - proceeding with completion",
+                        task_id=str(task_id),
+                    )
+
+            # Main PM reviewed - check if parent task needs CEO approval
+            if completing_agent_role == "main_pm" and all_descendants:
+                # Parent task with descendants - needs CEO approval
+                self.log.info(
+                    "Main PM approved parent task - escalating to CEO",
+                    task_id=str(task_id),
+                    descendant_count=len(all_descendants),
+                )
+                return await self.escalate_to_ceo(task_id, "main_pm")
+
         # Check for cancelled descendants (only matters if force override requested)
         cancelled_descendants = [
             st for st in all_descendants if st.status == TaskStatus.CANCELLED
@@ -2015,7 +2082,7 @@ class TaskService(BaseService):
         if task.commits:
             code_task = asyncio.create_task(
                 self._index_code_changes_background(
-                    task.id,
+                    require_uuid(task.id),
                     task.commits,
                     task.team.value if task.team else "default",
                 )
@@ -2027,7 +2094,11 @@ class TaskService(BaseService):
         if task.dev_notes:
             decision_task = asyncio.create_task(
                 self._index_decisions_background(
-                    task.id, task.title, task.team, task.dev_notes, task.assigned_to
+                    require_uuid(task.id),
+                    task.title,
+                    task.team,
+                    task.dev_notes,
+                    to_python_uuid(task.assigned_to),
                 )
             )
             self._background_tasks.add(decision_task)
@@ -2446,13 +2517,17 @@ class TaskService(BaseService):
         status: TaskStatus | None = None,
         limit: int = 100,
     ) -> list[TaskTable]:
-        """List tasks for a specific team."""
+        """List tasks for a team, ordered by priority, sequence, created_at."""
         query = select(TaskTable).where(TaskTable.team == team)
 
         if status:
             query = query.where(TaskTable.status == status)
 
-        query = query.order_by(TaskTable.priority, TaskTable.created_at.desc())
+        query = query.order_by(
+            TaskTable.priority,
+            TaskTable.sequence,
+            TaskTable.created_at,
+        )
         query = query.limit(limit)
 
         result = await self.session.execute(query)
@@ -2513,20 +2588,61 @@ class TaskService(BaseService):
         status: TaskStatus,
         team: Team | None = None,
     ) -> list[TaskTable]:
-        """List tasks with a specific status."""
+        """List tasks by status, ordered by priority, sequence, created_at."""
         query = select(TaskTable).where(TaskTable.status == status)
 
         if team:
             query = query.where(TaskTable.team == team)
 
-        query = query.order_by(TaskTable.priority, TaskTable.created_at.desc())
+        # Order by priority first, then sequence (for sibling order), then created_at
+        query = query.order_by(
+            TaskTable.priority,
+            TaskTable.sequence,
+            TaskTable.created_at,
+        )
 
         result = await self.session.execute(query)
         return list(result.scalars().all())
 
-    async def list_pending(self, team: Team | None = None) -> list[TaskTable]:
-        """List pending tasks (available to claim)."""
-        return await self.list_by_status(TaskStatus.PENDING, team)
+    async def list_pending(
+        self,
+        team: Team | None = None,
+        filter_by_dependencies: bool = True,
+    ) -> list[TaskTable]:
+        """
+        List pending tasks (available to claim).
+
+        Args:
+            team: Filter by team
+            filter_by_dependencies: If True, exclude tasks with incomplete dependencies
+
+        Returns:
+            List of pending tasks, ordered by priority, sequence, then created_at
+        """
+        tasks = await self.list_by_status(TaskStatus.PENDING, team)
+
+        if not filter_by_dependencies:
+            return tasks
+
+        # Filter out tasks whose dependencies aren't complete
+        available_tasks = []
+        for task in tasks:
+            if not task.dependency_ids:
+                available_tasks.append(task)
+                continue
+
+            # Check if all dependencies are complete
+            deps_result = await self.session.execute(
+                select(TaskTable.status).where(TaskTable.id.in_(task.dependency_ids))
+            )
+            dep_statuses = deps_result.scalars().all()
+
+            # All dependencies must be COMPLETED or CANCELLED
+            terminal_statuses = {TaskStatus.COMPLETED, TaskStatus.CANCELLED}
+            if all(s in terminal_statuses for s in dep_statuses):
+                available_tasks.append(task)
+
+        return available_tasks
 
     async def list_blocked(self, team: Team | None = None) -> list[TaskTable]:
         """List blocked tasks."""
@@ -2660,7 +2776,7 @@ async def resolve_pm_for_substitute(
         select(AgentTable).where(AgentTable.slug == target_pm_slug)
     )
     pm_agent = pm_result.scalar_one_or_none()
-    return target_pm_slug, pm_agent.id if pm_agent else None
+    return target_pm_slug, to_python_uuid(pm_agent.id) if pm_agent else None
 
 
 async def notify_pm_for_substitute(
@@ -2703,7 +2819,7 @@ async def notify_pm_for_substitute(
     await db.flush()
 
     delivery_service = get_notification_delivery_service(db)
-    await delivery_service.deliver(notification.id)
+    await delivery_service.deliver(require_uuid(notification.id))
 
 
 # =============================================================================