+ Radon + Xenon + Pip-Audit + Bandit + Safety

.safety-project.ini

@@ -0,0 +1,5 @@
+[project]
+id = roboco
+url = /codebases/roboco/findings
+name = roboco
+

roboco/agents/board.py

@@ -681,19 +681,10 @@ efficiency,warning,Unclear handoff process,3 tasks delayed,Document handoff step
         except Exception as e:
             self.log.error("Failed to send CEO report", error=str(e))
 
-    async def _perform_audit(self, audit_type: str) -> str | None:
+    async def _audit_code_quality(self, tasks: list[dict]) -> str | None:
-        """Perform a specific type of audit."""
+        """Audit code quality from completed tasks."""
-        try:
+        if not tasks:
-            # Query relevant data based on audit type
+            return None
-            if audit_type == "code_quality":
-                result = await self._api_call(
-                    "GET",
-                    "/tasks",
-                    params={"status": "completed", "limit": 10},
-                )
-                tasks = result.get("items", [])
-                # Analyze completed tasks for quality issues
-                if tasks:
         prompt = f"""
 Analyze these completed tasks for code quality patterns:
 
@@ -709,34 +700,42 @@ Report findings or None if all looks good.
 """
         return await self.think(prompt)
 
-            elif audit_type == "documentation":
+    async def _audit_documentation(self, tasks: list[dict]) -> str | None:
-                result = await self._api_call(
+        """Audit documentation completeness."""
-                    "GET",
-                    "/tasks",
-                    params={"status": "completed", "limit": 10},
-                )
-                tasks = result.get("items", [])
         missing_docs = [t for t in tasks if not t.get("documentation_complete")]
         if missing_docs:
-                    count = len(missing_docs)
+            return f"Found {len(missing_docs)} tasks with incomplete documentation"
-                    return f"Found {count} tasks with incomplete documentation"
+        return None
 
-            elif audit_type == "process_compliance":
+    async def _audit_process_compliance(self, tasks: list[dict]) -> str | None:
-                # Check for process violations
+        """Audit process compliance."""
+        violations = [
+            f"{t.get('title')} - no QA" for t in tasks if not t.get("qa_passed")
+        ]
+        if violations:
+            return f"Process violations: {', '.join(violations)}"
+        return None
+
+    async def _perform_audit(self, audit_type: str) -> str | None:
+        """Perform a specific type of audit."""
+        audit_handlers = {
+            "code_quality": self._audit_code_quality,
+            "documentation": self._audit_documentation,
+            "process_compliance": self._audit_process_compliance,
+        }
+
+        handler = audit_handlers.get(audit_type)
+        if not handler:
+            return None
+
+        try:
             result = await self._api_call(
                 "GET",
                 "/tasks",
                 params={"status": "completed", "limit": 10},
             )
             tasks = result.get("items", [])
-                violations = []
+            return await handler(tasks)
-                for task in tasks:
-                    if not task.get("qa_passed"):
-                        violations.append(f"{task.get('title')} - no QA")
-                if violations:
-                    return f"Process violations: {', '.join(violations)}"
-
-            return None
         except Exception as e:
             self.log.warning("Failed to perform audit", error=str(e))
             return None

roboco/agents/documenter.py

@@ -144,6 +144,51 @@ class DocumenterAgent(Agent):
 
         return None
 
+    async def _run_phase(self, ctx: DocContext) -> bool | None:
+        """
+        Run the current phase and return transition info.
+
+        Returns:
+            True if task is complete
+            False if phase didn't complete (stay in current phase)
+            None if phase completed (advance to next phase)
+        """
+        phase_transitions: dict[DocTaskPhase, tuple[DocTaskPhase | None, bool]] = {
+            DocTaskPhase.RECEIVE: (DocTaskPhase.GATHER, True),
+            DocTaskPhase.GATHER: (DocTaskPhase.SYNTHESIZE, True),
+            DocTaskPhase.SYNTHESIZE: (DocTaskPhase.WRITE, True),
+            DocTaskPhase.REVIEW: (DocTaskPhase.PUBLISH, True),
+            DocTaskPhase.PUBLISH: (None, True),  # Terminal phase
+        }
+
+        phase_handlers = {
+            DocTaskPhase.RECEIVE: self._phase_receive,
+            DocTaskPhase.GATHER: self._phase_gather,
+            DocTaskPhase.SYNTHESIZE: self._phase_synthesize,
+            DocTaskPhase.REVIEW: self._phase_review,
+            DocTaskPhase.PUBLISH: self._phase_publish,
+        }
+
+        if ctx.phase == DocTaskPhase.WRITE:
+            completed = await self._phase_write(ctx)
+            if completed:
+                ctx.phase = DocTaskPhase.REVIEW
+            return None
+
+        handler = phase_handlers.get(ctx.phase)
+        if handler:
+            await handler(ctx)
+
+        transition = phase_transitions.get(ctx.phase)
+        if transition:
+            next_phase, _ = transition
+            if next_phase is None:
+                self._doc_context = None
+                return True
+            ctx.phase = next_phase
+
+        return None
+
     async def execute_task(self, task_id: UUID) -> bool:
         """
         Execute documentation through lifecycle phases.
@@ -156,40 +201,15 @@ class DocumenterAgent(Agent):
                 title=await self._get_task_title(task_id),
             )
 
-        ctx = self._doc_context
-
         try:
-            match ctx.phase:
+            result = await self._run_phase(self._doc_context)
-                case DocTaskPhase.RECEIVE:
+            return result is True
-                    await self._phase_receive(ctx)
-                    ctx.phase = DocTaskPhase.GATHER
-
-                case DocTaskPhase.GATHER:
-                    await self._phase_gather(ctx)
-                    ctx.phase = DocTaskPhase.SYNTHESIZE
-
-                case DocTaskPhase.SYNTHESIZE:
-                    await self._phase_synthesize(ctx)
-                    ctx.phase = DocTaskPhase.WRITE
-
-                case DocTaskPhase.WRITE:
-                    completed = await self._phase_write(ctx)
-                    if completed:
-                        ctx.phase = DocTaskPhase.REVIEW
-
-                case DocTaskPhase.REVIEW:
-                    await self._phase_review(ctx)
-                    ctx.phase = DocTaskPhase.PUBLISH
-
-                case DocTaskPhase.PUBLISH:
-                    await self._phase_publish(ctx)
-                    self._doc_context = None
-                    return True
-
-            return False
-
         except Exception as e:
-            self.log.error("Error in doc phase", phase=ctx.phase.value, error=str(e))
+            self.log.error(
+                "Error in doc phase",
+                phase=self._doc_context.phase.value,
+                error=str(e),
+            )
             return False
 
     # =========================================================================

roboco/agents/qa.py

@@ -135,53 +135,71 @@ class QAAgent(Agent):
 
         return None
 
+    async def _run_phase(self, ctx: ReviewContext) -> bool | None:
+        """
+        Run the current phase and return transition info.
+
+        Returns:
+            True if review is complete
+            None if phase completed (advance to next phase)
+        """
+        phase_transitions: dict[QATaskPhase, QATaskPhase | None] = {
+            QATaskPhase.RECEIVE: QATaskPhase.UNDERSTAND,
+            QATaskPhase.UNDERSTAND: QATaskPhase.TEST,
+            QATaskPhase.VERDICT: QATaskPhase.DOCUMENT,
+            QATaskPhase.DOCUMENT: QATaskPhase.RETURN,
+            QATaskPhase.RETURN: None,  # Terminal phase
+        }
+
+        phase_handlers = {
+            QATaskPhase.RECEIVE: self._phase_receive,
+            QATaskPhase.UNDERSTAND: self._phase_understand,
+            QATaskPhase.VERDICT: self._phase_verdict,
+            QATaskPhase.DOCUMENT: self._phase_document,
+        }
+
+        if ctx.phase == QATaskPhase.TEST:
+            completed = await self._phase_test(ctx)
+            if completed:
+                ctx.phase = QATaskPhase.VERDICT
+            return None
+
+        if ctx.phase == QATaskPhase.RETURN:
+            self._review_context = None
+            return True
+
+        handler = phase_handlers.get(ctx.phase)
+        if handler:
+            await handler(ctx)
+
+        next_phase = phase_transitions.get(ctx.phase)
+        if next_phase:
+            ctx.phase = next_phase
+
+        return None
+
     async def execute_task(self, task_id: UUID) -> bool:
         """
         Execute review through QA lifecycle phases.
 
         Returns True when review is complete.
         """
-        # Initialize or restore review context
         if self._review_context is None or self._review_context.task_id != task_id:
             self._review_context = ReviewContext(
                 task_id=task_id,
                 title=await self._get_task_title(task_id),
             )
 
-        ctx = self._review_context
-
         try:
-            match ctx.phase:
+            result = await self._run_phase(self._review_context)
-                case QATaskPhase.RECEIVE:
+            return result is True
-                    await self._phase_receive(ctx)
-                    ctx.phase = QATaskPhase.UNDERSTAND
-
-                case QATaskPhase.UNDERSTAND:
-                    await self._phase_understand(ctx)
-                    ctx.phase = QATaskPhase.TEST
-
-                case QATaskPhase.TEST:
-                    completed = await self._phase_test(ctx)
-                    if completed:
-                        ctx.phase = QATaskPhase.VERDICT
-
-                case QATaskPhase.VERDICT:
-                    await self._phase_verdict(ctx)
-                    ctx.phase = QATaskPhase.DOCUMENT
-
-                case QATaskPhase.DOCUMENT:
-                    await self._phase_document(ctx)
-                    ctx.phase = QATaskPhase.RETURN
-
-                case QATaskPhase.RETURN:
-                    self._review_context = None
-                    return True
-
-            return False
-
         except Exception as e:
-            self.log.error("Error in review phase", phase=ctx.phase.value, error=str(e))
+            self.log.error(
-            ctx.findings.append(f"Error during review: {e}")
+                "Error in review phase",
+                phase=self._review_context.phase.value,
+                error=str(e),
+            )
+            self._review_context.findings.append(f"Error during review: {e}")
             return False
 
     # =========================================================================

roboco/api/routes/dashboard.py

@@ -411,6 +411,94 @@ async def send_auditor_report(report_id: UUID) -> dict[str, str]:
 # =============================================================================
 
 
+async def _get_team_health_list(metrics_service: Any) -> list[TeamHealth]:
+    """Get health status for all teams."""
+    teams = [Team.BACKEND, Team.FRONTEND, Team.UX_UI, Team.BOARD]
+    health_list = []
+    for team in teams:
+        health = await metrics_service.get_health_status(team)
+        health_list.append(
+            TeamHealth(
+                team=team.value,
+                status=health["status"],
+                active_tasks=health["active_tasks"],
+                blocked_tasks=health["blocked_tasks"],
+                blocked_ratio=health["blocked_ratio"],
+                completed_this_week=health["completed_this_week"],
+            )
+        )
+    return health_list
+
+
+async def _get_key_metrics(metrics_service: Any) -> dict[str, Any]:
+    """Get key organization metrics."""
+    velocity = await metrics_service.get_velocity(7)
+    team_metrics = await metrics_service.get_all_team_metrics()
+    blockers = await metrics_service.get_blocker_metrics()
+
+    total_docs = sum(tm.documentation_coverage for tm in team_metrics)
+    avg_doc_coverage = total_docs / len(team_metrics) if team_metrics else 0
+
+    return {
+        "velocity_weekly": velocity.tasks_completed,
+        "completion_rate": velocity.completion_rate,
+        "documentation_coverage": round(avg_doc_coverage, 2),
+        "active_blockers": blockers.active_blockers,
+    }
+
+
+def _count_unresolved_flags(severity: str) -> int:
+    """Count unresolved flags of a given severity."""
+    return sum(
+        1
+        for f in _flags.values()
+        if f.get("severity") == severity and not f.get("resolved_at")
+    )
+
+
+def _get_last_report_time() -> str | None:
+    """Get the timestamp of the most recent report."""
+    recent_reports = [r for r in _reports.values() if r.get("sent_at")]
+    if not recent_reports:
+        return None
+    last_time = max(r["sent_at"] for r in recent_reports)
+    return last_time.isoformat() if last_time else None
+
+
+def _get_auditor_alerts() -> dict[str, Any]:
+    """Get auditor alerts summary."""
+    return {
+        "urgent_count": _count_unresolved_flags("urgent"),
+        "warning_count": _count_unresolved_flags("warning"),
+        "last_report_at": _get_last_report_time(),
+    }
+
+
+async def _get_roadmap_progress(db: DbSession) -> dict[str, Any]:
+    """Get roadmap progress from high-priority tasks."""
+    total_result = await db.execute(
+        select(func.count(TaskTable.id)).where(TaskTable.priority <= 1)
+    )
+    total_priority = total_result.scalar() or 0
+
+    completed_result = await db.execute(
+        select(func.count(TaskTable.id)).where(
+            and_(
+                TaskTable.priority <= 1,
+                TaskTable.status == TaskStatus.COMPLETED,
+            )
+        )
+    )
+    completed_priority = completed_result.scalar() or 0
+    progress = completed_priority / total_priority if total_priority > 0 else 0
+
+    return {
+        "current_quarter_progress": round(progress, 2),
+        "high_priority_total": total_priority,
+        "high_priority_completed": completed_priority,
+    }
+
+
 @router.get("/ceo", response_model=CEOOverview)
 async def get_ceo_overview(
     db: DbSession,
@@ -426,94 +514,11 @@ async def get_ceo_overview(
     """
     metrics_service = get_metrics_service(db)
 
-    # Get health status for each team
-    health_status = []
-    for team in [Team.BACKEND, Team.FRONTEND, Team.UX_UI, Team.BOARD]:
-        health = await metrics_service.get_health_status(team)
-        health_status.append(
-            TeamHealth(
-                team=team.value,
-                status=health["status"],
-                active_tasks=health["active_tasks"],
-                blocked_tasks=health["blocked_tasks"],
-                blocked_ratio=health["blocked_ratio"],
-                completed_this_week=health["completed_this_week"],
-            )
-        )
-
-    # Get key metrics
-    velocity = await metrics_service.get_velocity(7)
-    team_metrics = await metrics_service.get_all_team_metrics()
-
-    # Calculate documentation coverage
-    total_docs = sum(tm.documentation_coverage for tm in team_metrics)
-    avg_doc_coverage = total_docs / len(team_metrics) if team_metrics else 0
-
-    # Get blocker info
-    blockers = await metrics_service.get_blocker_metrics()
-
-    key_metrics = {
-        "velocity_weekly": velocity.tasks_completed,
-        "completion_rate": velocity.completion_rate,
-        "documentation_coverage": round(avg_doc_coverage, 2),
-        "active_blockers": blockers.active_blockers,
-    }
-
-    # Auditor alerts summary
-    urgent_flags = sum(
-        1
-        for f in _flags.values()
-        if f.get("severity") == "urgent" and not f.get("resolved_at")
-    )
-    warning_flags = sum(
-        1
-        for f in _flags.values()
-        if f.get("severity") == "warning" and not f.get("resolved_at")
-    )
-
-    recent_reports = [r for r in _reports.values() if r.get("sent_at")]
-    last_report_at = (
-        max((r["sent_at"] for r in recent_reports), default=None)
-        if recent_reports
-        else None
-    )
-
-    auditor_alerts = {
-        "urgent_count": urgent_flags,
-        "warning_count": warning_flags,
-        "last_report_at": last_report_at.isoformat() if last_report_at else None,
-    }
-
-    # Roadmap progress (simplified - would query epics/milestones)
-    # For now, calculate from task completion rates
-    total_result = await db.execute(
-        select(func.count(TaskTable.id)).where(TaskTable.priority <= 1)
-    )
-    total_priority = total_result.scalar() or 0
-
-    completed_result = await db.execute(
-        select(func.count(TaskTable.id)).where(
-            and_(
-                TaskTable.priority <= 1,
-                TaskTable.status == TaskStatus.COMPLETED,
-            )
-        )
-    )
-    completed_priority = completed_result.scalar() or 0
-
-    progress = completed_priority / total_priority if total_priority > 0 else 0
-
-    roadmap_progress = {
-        "current_quarter_progress": round(progress, 2),
-        "high_priority_total": total_priority,
-        "high_priority_completed": completed_priority,
-    }
-
     return CEOOverview(
-        health_status=health_status,
+        health_status=await _get_team_health_list(metrics_service),
-        key_metrics=key_metrics,
+        key_metrics=await _get_key_metrics(metrics_service),
-        auditor_alerts=auditor_alerts,
+        auditor_alerts=_get_auditor_alerts(),
-        roadmap_progress=roadmap_progress,
+        roadmap_progress=await _get_roadmap_progress(db),
     )
 
 

roboco/api/routes/messages.py

@@ -5,7 +5,7 @@ CRUD operations for messages within sessions.
 """
 
 from datetime import UTC, datetime
-from typing import Annotated
+from typing import Annotated, cast
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, status
@@ -212,6 +212,77 @@ async def get_message(
     )
 
 
+async def _get_session_with_group(
+    db: DbSession,
+    session_id: UUID,
+) -> SessionTable:
+    """Get session with group loaded, or raise 404."""
+    result = await db.execute(
+        select(SessionTable)
+        .where(SessionTable.id == session_id)
+        .options(selectinload(SessionTable.group))
+    )
+    session = result.scalar_one_or_none()
+    if not session:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Session not found",
+        )
+    if session.status != SessionStatus.ACTIVE:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Session is not active",
+        )
+    return session
+
+
+async def _get_channel_with_access(
+    db: DbSession,
+    channel_id: UUID,
+    agent_id: UUID,
+) -> ChannelTable:
+    """Get channel and verify write access, or raise 403."""
+    result = await db.execute(select(ChannelTable).where(ChannelTable.id == channel_id))
+    channel = result.scalar_one_or_none()
+    if not channel or agent_id not in channel.writers:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You don't have write access to this channel",
+        )
+    return channel
+
+
+async def _validate_reply_target(
+    db: DbSession,
+    reply_to: UUID,
+    session_id: UUID,
+) -> None:
+    """Validate reply target exists in session."""
+    result = await db.execute(
+        select(MessageTable).where(
+            MessageTable.id == reply_to,
+            MessageTable.session_id == session_id,
+        )
+    )
+    if not result.scalar_one_or_none():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Reply target message not found in this session",
+        )
+
+
+def _check_session_boundaries(session: SessionTable) -> bool:
+    """Check if session should be closed based on boundaries."""
+    msg_limit_exceeded = (
+        session.max_message_count and session.message_count >= session.max_message_count
+    )
+    content_limit_exceeded = (
+        session.max_content_length
+        and session.total_content_length >= session.max_content_length
+    )
+    return bool(msg_limit_exceeded or content_limit_exceeded)
+
+
 @router.post(
     "",
     response_model=MessageResponse,
@@ -225,54 +296,15 @@ async def send_message(
     data: MessageCreateRequest,
 ) -> MessageResponse:
     """Send a message to a session."""
-    # Get session with group and channel
+    session = await _get_session_with_group(db, data.session_id)
-    session_result = await db.execute(
-        select(SessionTable)
-        .where(SessionTable.id == data.session_id)
-        .options(selectinload(SessionTable.group))
-    )
-    session = session_result.scalar_one_or_none()
-
-    if not session:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="Session not found",
-        )
-
-    if session.status != SessionStatus.ACTIVE:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Session is not active",
-        )
-
-    # Get channel for access check
     group = session.group
-    channel_result = await db.execute(
+    channel = await _get_channel_with_access(
-        select(ChannelTable).where(ChannelTable.id == group.channel_id)
+        db, cast("UUID", group.channel_id), agent_id
-    )
-    channel = channel_result.scalar_one_or_none()
-
-    if not channel or agent_id not in channel.writers:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="You don't have write access to this channel",
     )
 
-    # Verify reply_to if provided
     if data.reply_to:
-        reply_result = await db.execute(
+        await _validate_reply_target(db, data.reply_to, data.session_id)
-            select(MessageTable).where(
-                MessageTable.id == data.reply_to,
-                MessageTable.session_id == data.session_id,
-            )
-        )
-        if not reply_result.scalar_one_or_none():
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Reply target message not found in this session",
-            )
 
-    # Create message
     content_length = len(data.content)
     message = MessageTable(
         agent_id=agent_id,
@@ -288,35 +320,20 @@ async def send_message(
         task_id=data.task_id,
         commit_ref=data.commit_ref,
     )
-
     db.add(message)
 
-    # Update session stats
+    now = datetime.now(UTC)
     session.message_count += 1
     session.total_content_length += content_length
-    session.last_activity_at = datetime.now(UTC)
+    session.last_activity_at = now
-
-    # Update group stats
     group.total_messages += 1
-    group.last_activity = datetime.now(UTC)
+    group.last_activity = now
-
-    # Update channel stats
     channel.message_count += 1
-    channel.last_activity = datetime.now(UTC)
+    channel.last_activity = now
 
-    # Check if session should be closed
+    if _check_session_boundaries(session):
-    should_close = False
-    if session.max_message_count and session.message_count >= session.max_message_count:
-        should_close = True
-    if (
-        session.max_content_length
-        and session.total_content_length >= session.max_content_length
-    ):
-        should_close = True
-
-    if should_close:
         session.status = SessionStatus.CLOSED
-        session.closed_at = datetime.now(UTC)
+        session.closed_at = now
         group.active_session_id = None
 
     await db.flush()

roboco/api/routes/notifications.py

@@ -6,7 +6,7 @@ Enforces permission rules: only PMs, Board, and Auditor can send notifications.
 """
 
 from datetime import UTC, datetime
-from typing import Annotated
+from typing import Annotated, Any
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, status
@@ -90,48 +90,32 @@ class NotificationCreateRequest(BaseModel):
 # =============================================================================
 
 
-@router.get(
+def _build_notification_query(
-    "",
+    agent_id: UUID,
-    response_model=NotificationListResponse,
+    params: ListNotificationsParams,
-    summary="List notifications",
+) -> Any:
-    description="List notifications for the current agent.",
+    """Build the notification query with filters."""
-)
-async def list_notifications(
-    db: DbSession,
-    agent_id: CurrentAgentId,
-    params: Annotated[ListNotificationsParams, Depends()],
-) -> NotificationListResponse:
-    """List notifications for the agent."""
-    # Query notifications where agent is a recipient
     query = select(NotificationTable).where(
         NotificationTable.to_agents.contains([agent_id])
     )
-
     if params.unread_only:
         query = query.where(~NotificationTable.read_by.contains([agent_id]))
-
     if params.pending_ack_only:
         query = query.where(
             NotificationTable.requires_ack.is_(True),
             ~NotificationTable.acked_by.contains([agent_id]),
         )
-
     if params.type_filter:
         query = query.where(NotificationTable.type == params.type_filter)
+    return query.order_by(NotificationTable.timestamp.desc()).limit(params.limit)
 
-    query = query.order_by(NotificationTable.timestamp.desc()).limit(params.limit)
 
-    result = await db.execute(query)
+def _notification_to_response(
-    notifications = result.scalars().all()
+    n: NotificationTable,
-
+    agent_id: UUID,
-    # Count unread and pending ack
+) -> NotificationResponse:
-    unread_count = sum(1 for n in notifications if agent_id not in n.read_by)
+    """Convert a notification to response format."""
-    pending_ack_count = sum(
+    return NotificationResponse(
-        1 for n in notifications if n.requires_ack and agent_id not in n.acked_by
-    )
-
-    items = [
-        NotificationResponse(
         id=require_uuid(n.id),
         type=n.type,
         priority=n.priority,
@@ -147,8 +131,29 @@ async def list_notifications(
         timestamp=n.timestamp,
         expires_at=n.expires_at,
     )
-        for n in notifications
+
-    ]
+
+@router.get(
+    "",
+    response_model=NotificationListResponse,
+    summary="List notifications",
+    description="List notifications for the current agent.",
+)
+async def list_notifications(
+    db: DbSession,
+    agent_id: CurrentAgentId,
+    params: Annotated[ListNotificationsParams, Depends()],
+) -> NotificationListResponse:
+    """List notifications for the agent."""
+    query = _build_notification_query(agent_id, params)
+    result: Any = await db.execute(query)
+    notifications = result.scalars().all()
+
+    unread_count = sum(1 for n in notifications if agent_id not in n.read_by)
+    pending_ack_count = sum(
+        1 for n in notifications if n.requires_ack and agent_id not in n.acked_by
+    )
+    items = [_notification_to_response(n, agent_id) for n in notifications]
 
     return NotificationListResponse(
         items=items,

roboco/bootstrap.py

@@ -312,6 +312,65 @@ async def create_agents(session: AsyncSession) -> dict[str, str]:
     return agent_ids
 
 
+async def _get_channel(
+    session: AsyncSession,
+    channel_id: str,
+) -> ChannelTable | None:
+    """Fetch a channel by ID."""
+    result = await session.execute(
+        select(ChannelTable).where(ChannelTable.id == UUIDType(channel_id))
+    )
+    return result.scalar_one_or_none()
+
+
+def _build_member_uuids(
+    agent_slugs: list[str],
+    agent_ids: dict[str, str],
+) -> list[UUIDType]:
+    """Build a list of UUIDs from agent slugs."""
+    return [UUIDType(agent_ids[slug]) for slug in agent_slugs if slug in agent_ids]
+
+
+async def _configure_channel_members(
+    session: AsyncSession,
+    channel_ids: dict[str, str],
+    agent_ids: dict[str, str],
+) -> None:
+    """Configure members and writers for all channels."""
+    for channel_slug, members in CHANNEL_MEMBERSHIPS.items():
+        channel_id = channel_ids.get(channel_slug)
+        if not channel_id:
+            continue
+
+        channel = await _get_channel(session, channel_id)
+        if not channel:
+            continue
+
+        member_uuids = _build_member_uuids(members, agent_ids)
+        channel.members = member_uuids
+        channel.writers = member_uuids  # All members can write by default
+
+
+async def _add_auditor_silent_access(
+    session: AsyncSession,
+    channel_ids: dict[str, str],
+    auditor_uuid: UUIDType,
+) -> None:
+    """Add auditor as silent observer to specified channels."""
+    for channel_slug in AUDITOR_SILENT_ACCESS:
+        channel_id = channel_ids.get(channel_slug)
+        if not channel_id:
+            continue
+
+        channel = await _get_channel(session, channel_id)
+        if not channel:
+            continue
+
+        observers = channel.silent_observers or []
+        if auditor_uuid not in observers:
+            channel.silent_observers = [*observers, auditor_uuid]
+
+
 async def create_channel_memberships(
     session: AsyncSession,
     channel_ids: dict[str, str],
@@ -323,52 +382,11 @@ async def create_channel_memberships(
     Note: ChannelTable uses arrays for members/writers/silent_observers
     rather than a separate membership table.
     """
-    for channel_slug, members in CHANNEL_MEMBERSHIPS.items():
+    await _configure_channel_members(session, channel_ids, agent_ids)
-        channel_id = channel_ids.get(channel_slug)
-        if not channel_id:
-            continue
 
-        # Get the channel
-        result = await session.execute(
-            select(ChannelTable).where(ChannelTable.id == UUIDType(channel_id))
-        )
-        channel = result.scalar_one_or_none()
-        if not channel:
-            continue
-
-        # Build member and writer UUID lists
-        member_uuids = []
-        writer_uuids = []
-
-        for agent_slug in members:
-            db_agent_id = agent_ids.get(agent_slug)
-            if db_agent_id:
-                uuid = UUIDType(db_agent_id)
-                member_uuids.append(uuid)
-                writer_uuids.append(uuid)  # All members can write by default
-
-        # Update channel
-        channel.members = member_uuids
-        channel.writers = writer_uuids
-
-    # Add auditor silent access to specified channels
     auditor_db_id = agent_ids.get("auditor")
     if auditor_db_id:
-        auditor_uuid = UUIDType(auditor_db_id)
+        await _add_auditor_silent_access(session, channel_ids, UUIDType(auditor_db_id))
-
-        for channel_slug in AUDITOR_SILENT_ACCESS:
-            channel_id = channel_ids.get(channel_slug)
-            if not channel_id:
-                continue
-
-            result = await session.execute(
-                select(ChannelTable).where(ChannelTable.id == UUIDType(channel_id))
-            )
-            channel = result.scalar_one_or_none()
-            # Add auditor to silent_observers (read-only)
-            observers = channel.silent_observers or [] if channel else []
-            if channel and auditor_uuid not in observers:
-                channel.silent_observers = [*observers, auditor_uuid]
 
     logger.info("Channel memberships configured")
 

roboco/events/handlers.py

@@ -13,6 +13,76 @@ from roboco.events.bus import Event, EventType, get_event_bus
 logger = structlog.get_logger()
 
 
+async def _handle_task_blocked(
+    event: Event,
+    task_id: str,
+    notification_service: Any,
+) -> None:
+    """Handle blocked task notification."""
+    team = event.data.get("team")
+    if not team:
+        return
+    blocker_reason = event.data.get("reason", "Unknown blocker")
+    pm_id = f"{team[:2]}-pm"
+    await notification_service.send_blocker_notification(
+        task_id=task_id,
+        blocker_reason=blocker_reason,
+        from_agent=event.source_agent,
+        to_pm=pm_id,
+    )
+
+
+async def _handle_task_awaiting_qa(
+    event: Event,
+    task_id: str,
+    notification_service: Any,
+) -> None:
+    """Handle task awaiting QA notification."""
+    team = event.data.get("team")
+    if not team:
+        return
+    qa_id = f"{team[:2]}-qa"
+    await notification_service.send_qa_ready_notification(
+        task_id=task_id,
+        from_agent=event.source_agent,
+        to_qa=qa_id,
+    )
+
+
+async def _handle_task_qa_failed(
+    event: Event,
+    task_id: str,
+    notification_service: Any,
+) -> None:
+    """Handle QA failed notification."""
+    developer_id = event.data.get("assigned_to")
+    if not developer_id:
+        return
+    qa_notes = event.data.get("qa_notes", "See task for details")
+    await notification_service.send_qa_failed_notification(
+        task_id=task_id,
+        qa_notes=qa_notes,
+        to_developer=developer_id,
+    )
+
+
+async def _handle_task_awaiting_docs(
+    event: Event,
+    task_id: str,
+    notification_service: Any,
+) -> None:
+    """Handle task awaiting docs notification."""
+    team = event.data.get("team")
+    if not team:
+        return
+    doc_id = f"{team[:2]}-doc"
+    await notification_service.send_docs_ready_notification(
+        task_id=task_id,
+        from_agent=event.source_agent,
+        to_documenter=doc_id,
+    )
+
+
 async def handle_task_status_change(event: Event) -> None:
     """
     Handle task status change events.
@@ -25,68 +95,28 @@ async def handle_task_status_change(event: Event) -> None:
     """
     task_id_raw = event.data.get("task_id")
     task_id = str(task_id_raw) if task_id_raw else ""
-    agent_id = event.source_agent
-    event_type = event.type
 
     logger.info(
         "Task status changed",
         task_id=task_id,
-        event_type=event_type.value,
+        event_type=event.type.value,
-        agent=agent_id,
+        agent=event.source_agent,
     )
 
-    # Import here to avoid circular imports
     from roboco.services.notification import NotificationService  # noqa: PLC0415
 
     notification_service = NotificationService()
 
-    if event_type == EventType.TASK_BLOCKED:
+    handlers = {
-        # Notify the cell PM
+        EventType.TASK_BLOCKED: _handle_task_blocked,
-        blocker_reason = event.data.get("reason", "Unknown blocker")
+        EventType.TASK_AWAITING_QA: _handle_task_awaiting_qa,
-        team = event.data.get("team")
+        EventType.TASK_QA_FAILED: _handle_task_qa_failed,
+        EventType.TASK_AWAITING_DOCS: _handle_task_awaiting_docs,
+    }
 
-        if team:
+    handler = handlers.get(event.type)
-            pm_id = f"{team[:2]}-pm"  # e.g., "backend" -> "be-pm"
+    if handler:
-            await notification_service.send_blocker_notification(
+        await handler(event, task_id, notification_service)
-                task_id=task_id,
-                blocker_reason=blocker_reason,
-                from_agent=agent_id,
-                to_pm=pm_id,
-            )
-
-    elif event_type == EventType.TASK_AWAITING_QA:
-        # Notify the QA agent
-        team = event.data.get("team")
-        if team:
-            qa_id = f"{team[:2]}-qa"
-            await notification_service.send_qa_ready_notification(
-                task_id=task_id,
-                from_agent=agent_id,
-                to_qa=qa_id,
-            )
-
-    elif event_type == EventType.TASK_QA_FAILED:
-        # Notify the original developer
-        developer_id = event.data.get("assigned_to")
-        qa_notes = event.data.get("qa_notes", "See task for details")
-
-        if developer_id:
-            await notification_service.send_qa_failed_notification(
-                task_id=task_id,
-                qa_notes=qa_notes,
-                to_developer=developer_id,
-            )
-
-    elif event_type == EventType.TASK_AWAITING_DOCS:
-        # Notify the documenter
-        team = event.data.get("team")
-        if team:
-            doc_id = f"{team[:2]}-doc"
-            await notification_service.send_docs_ready_notification(
-                task_id=task_id,
-                from_agent=agent_id,
-                to_documenter=doc_id,
-            )
 
 
 async def handle_session_boundary(event: Event) -> None:
@@ -148,18 +178,35 @@ async def handle_handoff_created(event: Event) -> None:
         )
 
 
-async def handle_qa_result(event: Event) -> None:
+async def _try_resolve_agent_wait(
-    """
+    agent_id: str | None,
-    Handle QA result events.
+    waiting_for: str,
+    resolution: dict[str, Any],
+) -> None:
+    """Try to resolve a waiting agent if orchestrator is running."""
+    if not agent_id:
+        return
+    try:
+        from roboco.bootstrap import _BootstrapHolder  # noqa: PLC0415
 
-    Triggers:
+        orchestrator = _BootstrapHolder.orchestrator
-    - Resume developer agent if waiting on QA result
+        if not orchestrator:
-    - Notify appropriate parties
+            return
-    """
+        waiting = orchestrator.get_waiting_agents()
+        if agent_id not in waiting:
+            return
+        record = waiting[agent_id]
+        if record.waiting_for == waiting_for:
+            await orchestrator.resolve_wait(agent_id=agent_id, resolution=resolution)
+    except (ImportError, AttributeError):
+        pass
+
+
+async def handle_qa_result(event: Event) -> None:
+    """Handle QA result events."""
     task_id = event.data.get("task_id")
     passed = event.type == EventType.TASK_QA_PASSED
     developer_id = event.data.get("assigned_to")
-    qa_notes = event.data.get("qa_notes")
 
     logger.info(
         "QA result",
@@ -168,106 +215,41 @@ async def handle_qa_result(event: Event) -> None:
         developer=developer_id,
     )
 
-    # Check if developer agent is in WAITING_LONG state
+    await _try_resolve_agent_wait(
-
+        developer_id,
-    # Get orchestrator instance (if running)
+        "qa_result",
-    try:
+        {"passed": passed, "notes": event.data.get("qa_notes"), "task_id": task_id},
-        from roboco.bootstrap import _BootstrapHolder  # noqa: PLC0415
-
-        orchestrator = _BootstrapHolder.orchestrator
-        if orchestrator and developer_id:
-            waiting = orchestrator.get_waiting_agents()
-            if developer_id in waiting:
-                record = waiting[developer_id]
-                if record.waiting_for == "qa_result":
-                    # Resume the agent
-                    await orchestrator.resolve_wait(
-                        agent_id=developer_id,
-                        resolution={
-                            "passed": passed,
-                            "notes": qa_notes,
-                            "task_id": task_id,
-                        },
     )
-    except (ImportError, AttributeError):
-        pass  # Orchestrator not running
 
 
 async def handle_blocker_resolved(event: Event) -> None:
-    """
+    """Handle blocker resolution events."""
-    Handle blocker resolution events.
-
-    Triggers:
-    - Resume blocked agent
-    - Update task status
-    """
     task_id = event.data.get("task_id")
     agent_id = event.data.get("agent_id")
     resolution = event.data.get("resolution", "Resolved")
 
-    logger.info(
+    logger.info("Blocker resolved", task_id=task_id, agent=agent_id)
-        "Blocker resolved",
-        task_id=task_id,
-        agent=agent_id,
-    )
 
-    # Resume agent if waiting
+    await _try_resolve_agent_wait(
-    try:
+        agent_id,
-        from roboco.bootstrap import _BootstrapHolder  # noqa: PLC0415
+        "blocker_resolution",
-
+        {"details": resolution, "task_id": task_id},
-        orchestrator = _BootstrapHolder.orchestrator
-        if orchestrator and agent_id:
-            waiting = orchestrator.get_waiting_agents()
-            if agent_id in waiting:
-                record = waiting[agent_id]
-                if record.waiting_for == "blocker_resolution":
-                    await orchestrator.resolve_wait(
-                        agent_id=agent_id,
-                        resolution={
-                            "details": resolution,
-                            "task_id": task_id,
-                        },
     )
-    except (ImportError, AttributeError):
-        pass
 
 
 async def handle_question_answered(event: Event) -> None:
-    """
+    """Handle question answered events."""
-    Handle question answered events.
-
-    Triggers:
-    - Resume agent waiting for answer
-    """
     question_id = event.data.get("question_id")
     agent_id = event.data.get("asking_agent")
     answer = event.data.get("answer")
 
-    logger.info(
+    logger.info("Question answered", question_id=question_id, agent=agent_id)
-        "Question answered",
-        question_id=question_id,
-        agent=agent_id,
-    )
 
-    # Resume agent if waiting
+    await _try_resolve_agent_wait(
-    try:
+        agent_id,
-        from roboco.bootstrap import _BootstrapHolder  # noqa: PLC0415
+        "answer",
-
+        {"answer": answer, "question_id": question_id},
-        orchestrator = _BootstrapHolder.orchestrator
-        if orchestrator and agent_id:
-            waiting = orchestrator.get_waiting_agents()
-            if agent_id in waiting:
-                record = waiting[agent_id]
-                if record.waiting_for == "answer":
-                    await orchestrator.resolve_wait(
-                        agent_id=agent_id,
-                        resolution={
-                            "answer": answer,
-                            "question_id": question_id,
-                        },
     )
-    except (ImportError, AttributeError):
-        pass
 
 
 def register_default_handlers(bus: Any = None) -> None:

roboco/mcp/task_server.py

@@ -259,47 +259,35 @@ async def _handle_task_get(task_id: str) -> dict[str, Any]:
     return _format_task_response(task, next_step, guidance)
 
 
-async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]:
+def _check_blocking_tasks(active_tasks: list[dict]) -> dict[str, Any] | None:
-    """Handle task claiming."""
+    """Check for blocking active tasks. Returns error or None."""
-    async with httpx.AsyncClient() as client:
+    blocking_statuses = ["claimed", "in_progress", "verifying"]
-        # Check for existing active tasks
+    blocking = [t for t in active_tasks if t.get("status") in blocking_statuses]
-        active_resp = await client.get(
+    if blocking:
-            f"{_get_api_url()}/tasks",
-            params={"assigned_to": agent_id},
-        )
-        if active_resp.status_code == status.HTTP_200_OK:
-            active_tasks = active_resp.json()
-            # Check for non-waiting active tasks
-            blocking_tasks = [
-                t
-                for t in active_tasks
-                if t.get("status") in ["claimed", "in_progress", "verifying"]
-            ]
-            if blocking_tasks:
         return _format_error_response(
             "ALREADY_ACTIVE",
-                    f"You already have an active task: "
+            f"You already have an active task: {blocking[0]['id']}. "
-                    f"{blocking_tasks[0]['id']}. "
             "Complete or pause it before claiming a new task.",
-                    {"active_task_id": blocking_tasks[0]["id"]},
+            {"active_task_id": blocking[0]["id"]},
         )
+    return None
 
-            # Check for paused tasks
+
-            paused_tasks = [t for t in active_tasks if t.get("status") == "paused"]
+def _check_paused_tasks(active_tasks: list[dict]) -> dict[str, Any] | None:
-            if paused_tasks:
+    """Check for paused tasks. Returns error or None."""
+    paused = [t for t in active_tasks if t.get("status") == "paused"]
+    if paused:
         return _format_error_response(
             "PAUSED_TASKS_EXIST",
-                    f"You have {len(paused_tasks)} paused task(s). "
+            f"You have {len(paused)} paused task(s). "
             "Resume paused work before claiming new tasks.",
-                    {"paused_task_ids": [t["id"] for t in paused_tasks]},
+            {"paused_task_ids": [t["id"] for t in paused]},
         )
+    return None
 
-        # Get the task to check status
-        task_resp = await client.get(f"{_get_api_url()}/tasks/{task_id}")
-        if task_resp.status_code == status.HTTP_404_NOT_FOUND:
-            return _format_error_response("NOT_FOUND", f"Task {task_id} not found")
 
-        task = task_resp.json()
+def _validate_task_claimable(task: dict) -> dict[str, Any] | None:
+    """Validate task can be claimed. Returns error or None."""
     if task.get("status") != "pending":
         return _format_error_response(
             "INVALID_STATE",
@@ -307,13 +295,45 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]:
             "Only 'pending' tasks can be claimed.",
             {"current_status": task.get("status")},
         )
+    return None
+
+
+async def _get_project_context(project_id: str) -> dict[str, Any] | None:
+    """Fetch project context if available."""
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(f"{_get_api_url()}/projects/{project_id}")
+        if resp.status_code == status.HTTP_200_OK:
+            result: dict[str, Any] = resp.json()
+            return result
+    return None
+
+
+async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]:
+    """Handle task claiming."""
+    async with httpx.AsyncClient() as client:
+        active_resp = await client.get(
+            f"{_get_api_url()}/tasks",
+            params={"assigned_to": agent_id},
+        )
+        if active_resp.status_code == status.HTTP_200_OK:
+            active_tasks = active_resp.json()
+            if error := _check_blocking_tasks(active_tasks):
+                return error
+            if error := _check_paused_tasks(active_tasks):
+                return error
+
+        task_resp = await client.get(f"{_get_api_url()}/tasks/{task_id}")
+        if task_resp.status_code == status.HTTP_404_NOT_FOUND:
+            return _format_error_response("NOT_FOUND", f"Task {task_id} not found")
+
+        task = task_resp.json()
+        if error := _validate_task_claimable(task):
+            return error
 
-        # Claim the task
         claim_resp = await client.post(
             f"{_get_api_url()}/tasks/{task_id}/claim",
             json={"agent_id": agent_id},
         )
-
         if claim_resp.status_code != status.HTTP_200_OK:
             return _format_error_response(
                 "CLAIM_FAILED",
@@ -323,15 +343,9 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]:
 
         claimed_task = claim_resp.json()
 
-    # Get project context if available
     project = None
     if claimed_task.get("project_id"):
-        async with httpx.AsyncClient() as client:
+        project = await _get_project_context(claimed_task["project_id"])
-            proj_resp = await client.get(
-                f"{_get_api_url()}/projects/{claimed_task['project_id']}"
-            )
-            if proj_resp.status_code == status.HTTP_200_OK:
-                project = proj_resp.json()
 
     return _format_task_response(
         claimed_task,
@@ -344,70 +358,71 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]:
     )
 
 
-async def _handle_task_plan(
+def _validate_task_ownership(task: dict, agent_id: str) -> dict[str, Any] | None:
-    task_id: str,
+    """Validate agent owns the task. Returns error or None."""
-    plan_params: dict[str, Any],
-    agent_id: str,
-) -> dict[str, Any]:
-    """Handle task planning.
-
-    Args:
-        task_id: The task UUID
-        plan_params: Dict with 'approach', 'sub_tasks', 'risks',
-                     'open_questions'
-        agent_id: The agent ID
-    """
-    approach = plan_params["approach"]
-    sub_tasks = plan_params["sub_tasks"]
-    risks = plan_params.get("risks")
-    open_questions = plan_params.get("open_questions")
-
-    async with httpx.AsyncClient() as client:
-        # Verify task state and ownership
-        task_resp = await client.get(f"{_get_api_url()}/tasks/{task_id}")
-        if task_resp.status_code == status.HTTP_404_NOT_FOUND:
-            return _format_error_response("NOT_FOUND", f"Task {task_id} not found")
-
-        task = task_resp.json()
-
     if task.get("assigned_to") != agent_id:
         return _format_error_response(
             "NOT_OWNER",
             "You are not assigned to this task",
             {"assigned_to": task.get("assigned_to")},
         )
+    return None
 
+
+def _validate_task_status_claimed(task: dict) -> dict[str, Any] | None:
+    """Validate task is in claimed status. Returns error or None."""
     if task.get("status") != "claimed":
         return _format_error_response(
             "INVALID_STATE",
-                f"Cannot submit plan for task in "
+            f"Cannot submit plan for task in '{task.get('status')}' status. "
-                f"'{task.get('status')}' status. "
             "Task must be 'claimed'.",
             {"current_status": task.get("status")},
         )
+    return None
 
-        # Submit the plan
+
-        plan_data = {
+def _build_plan_data(plan_params: dict[str, Any]) -> dict[str, Any]:
-            "approach": approach,
+    """Build the plan data structure from params."""
+    return {
+        "approach": plan_params["approach"],
         "sub_tasks": [
             {
                 "title": st.get("title", ""),
                 "description": st.get("description", ""),
                 "order": i,
             }
-                for i, st in enumerate(sub_tasks)
+            for i, st in enumerate(plan_params["sub_tasks"])
         ],
-            "risks": [{"description": r} for r in (risks or [])],
+        "risks": [{"description": r} for r in (plan_params.get("risks") or [])],
         "open_questions": [
-                {"question": q, "answered": False} for q in (open_questions or [])
+            {"question": q, "answered": False}
+            for q in (plan_params.get("open_questions") or [])
         ],
     }
 
+
+async def _handle_task_plan(
+    task_id: str,
+    plan_params: dict[str, Any],
+    agent_id: str,
+) -> dict[str, Any]:
+    """Handle task planning."""
+    async with httpx.AsyncClient() as client:
+        task_resp = await client.get(f"{_get_api_url()}/tasks/{task_id}")
+        if task_resp.status_code == status.HTTP_404_NOT_FOUND:
+            return _format_error_response("NOT_FOUND", f"Task {task_id} not found")
+
+        task = task_resp.json()
+        if error := _validate_task_ownership(task, agent_id):
+            return error
+        if error := _validate_task_status_claimed(task):
+            return error
+
+        plan_data = _build_plan_data(plan_params)
         update_resp = await client.patch(
             f"{_get_api_url()}/tasks/{task_id}",
             json={"plan": plan_data},
         )
-
         if update_resp.status_code != status.HTTP_200_OK:
             return _format_error_response(
                 "UPDATE_FAILED",
@@ -417,13 +432,12 @@ async def _handle_task_plan(
 
         updated_task = update_resp.json()
 
-    # Check for blocking questions
+    open_questions = plan_params.get("open_questions")
     if open_questions:
         return _format_task_response(
             updated_task,
             "ASK_QUESTIONS",
-            f"Plan saved but you have {len(open_questions)} "
+            f"Plan saved but you have {len(open_questions)} open question(s). "
-            "open question(s). "
             "Ask these questions in your cell channel before starting. "
             "Do NOT proceed until questions are answered.",
         )

roboco/services/messaging.py

@@ -461,6 +461,54 @@ class MessagingService:
     # MESSAGE OPERATIONS (TASK-014)
     # =========================================================================
 
+    async def _get_message_context(
+        self,
+        session_id: UUID,
+    ) -> tuple[SessionTable, GroupTable, ChannelTable]:
+        """Get session, group, and channel for sending a message."""
+        session = await self.get_session(session_id)
+        if not session:
+            raise ValueError(f"Session {session_id} not found")
+        if session.status != SessionStatus.ACTIVE:
+            raise ValueError("Session is not active")
+
+        group = await self.get_group(cast("UUID", session.group_id))
+        if not group:
+            raise ValueError(f"Group {session.group_id} not found")
+
+        channel = await self.get_channel(cast("UUID", group.channel_id))
+        if not channel:
+            raise ValueError(f"Channel {group.channel_id} not found")
+
+        return session, group, channel
+
+    async def _validate_reply_target(
+        self,
+        reply_to: UUID,
+        session_id: UUID,
+    ) -> None:
+        """Validate reply target exists in session."""
+        reply_msg = await self.get_message(reply_to)
+        if not reply_msg or reply_msg.session_id != session_id:
+            raise ValueError("Reply target not found in this session")
+
+    def _update_message_stats(
+        self,
+        session: SessionTable,
+        group: GroupTable,
+        channel: ChannelTable,
+        content_length: int,
+    ) -> None:
+        """Update statistics after sending a message."""
+        now = datetime.now(UTC)
+        session.message_count += 1
+        session.total_content_length += content_length
+        session.last_activity_at = now
+        group.total_messages += 1
+        group.last_activity = now
+        channel.message_count += 1
+        channel.last_activity = now
+
     async def send_message(
         self,
         req: MessageCreateRequest,
@@ -469,12 +517,6 @@ class MessagingService:
         """
         Send a message to a session.
 
-        - Validates session is active
-        - Validates agent has write access (if agent_slug provided)
-        - Updates session statistics
-        - Checks session boundaries (auto-close if exceeded)
-        - Publishes message event
-
         Args:
             req: Message creation request
             agent_slug: Agent slug for channel access validation (optional)
@@ -486,34 +528,13 @@ class MessagingService:
             ValueError: If session not found or not active
             ChannelAccessDeniedError: If agent cannot write to channel
         """
-        # Get session
+        session, group, channel = await self._get_message_context(req.session_id)
-        session = await self.get_session(req.session_id)
-        if not session:
-            raise ValueError(f"Session {req.session_id} not found")
 
-        if session.status != SessionStatus.ACTIVE:
-            raise ValueError("Session is not active")
-
-        # Get group and channel for access check
-        group = await self.get_group(cast("UUID", session.group_id))
-        if not group:
-            raise ValueError(f"Group {session.group_id} not found")
-
-        channel = await self.get_channel(cast("UUID", group.channel_id))
-        if not channel:
-            raise ValueError(f"Channel {group.channel_id} not found")
-
-        # Validate write access
         if agent_slug:
             validate_channel_access(agent_slug, channel.slug, "write")
-
-        # Validate reply target if provided
         if req.reply_to:
-            reply_msg = await self.get_message(req.reply_to)
+            await self._validate_reply_target(req.reply_to, req.session_id)
-            if not reply_msg or reply_msg.session_id != req.session_id:
-                raise ValueError("Reply target not found in this session")
 
-        # Create message
         content_length = len(req.content)
         message = MessageTable(
             agent_id=req.agent_id,
@@ -529,25 +550,10 @@ class MessagingService:
             task_id=req.task_id,
             commit_ref=req.commit_ref,
         )
-
         self.session.add(message)
-
+        self._update_message_stats(session, group, channel, content_length)
-        # Update session statistics
-        session.message_count += 1
-        session.total_content_length += content_length
-        session.last_activity_at = datetime.now(UTC)
-
-        # Update group statistics
-        group.total_messages += 1
-        group.last_activity = datetime.now(UTC)
-
-        # Update channel statistics
-        channel.message_count += 1
-        channel.last_activity = datetime.now(UTC)
-
         await self.session.flush()
 
-        # Check session boundaries - close if exceeded
         if self._check_session_boundaries(session):
             await self.close_session(cast("UUID", session.id), "Boundary exceeded")
 

roboco/services/metrics.py

@@ -494,6 +494,46 @@ class MetricsService:
     # HEALTH STATUS
     # =========================================================================
 
+    def _determine_health_status(
+        self,
+        blocked_ratio: float,
+        active_count: int,
+        completed_count: int,
+    ) -> str:
+        """Determine health status from metrics."""
+        critical_threshold = 0.3
+        slow_threshold = 0.15
+        stale_threshold = 5
+
+        if blocked_ratio > critical_threshold:
+            return "critical"
+        if blocked_ratio > slow_threshold:
+            return "slow"
+        if active_count > stale_threshold and completed_count == 0:
+            return "slow"
+        return "ok"
+
+    async def _get_task_count(
+        self,
+        status_filter: list[TaskStatus] | TaskStatus,
+        team: Team | None,
+        since: datetime | None = None,
+    ) -> int:
+        """Get count of tasks matching criteria."""
+        conditions: list[Any] = []
+        if isinstance(status_filter, list):
+            conditions.append(TaskTable.status.in_(status_filter))
+        else:
+            conditions.append(TaskTable.status == status_filter)
+        if team:
+            conditions.append(TaskTable.team == team)
+        if since:
+            conditions.append(TaskTable.completed_at >= since)
+
+        query = select(func.count(TaskTable.id)).where(and_(*conditions))
+        result = await self.session.execute(query)
+        return result.scalar() or 0
+
     async def get_health_status(self, team: Team | None = None) -> dict[str, Any]:
         """
         Get health status for a team or the whole organization.
@@ -504,67 +544,27 @@ class MetricsService:
         - Average task age
         """
         week_ago = datetime.now(UTC) - timedelta(days=7)
-
+        active_statuses = [
-        # Build base query
-        base_filter = []
-        if team:
-            base_filter.append(TaskTable.team == team)
-
-        # Get counts
-        active_query = select(func.count(TaskTable.id)).where(
-            and_(
-                TaskTable.status.in_(
-                    [
             TaskStatus.CLAIMED,
             TaskStatus.IN_PROGRESS,
             TaskStatus.VERIFYING,
             TaskStatus.AWAITING_QA,
             TaskStatus.BLOCKED,
         ]
-                ),
-                *base_filter,
-            )
-        )
-        active_result = await self.session.execute(active_query)
-        active_count = active_result.scalar() or 0
 
-        blocked_query = select(func.count(TaskTable.id)).where(
+        active_count = await self._get_task_count(active_statuses, team)
-            and_(
+        blocked_count = await self._get_task_count(TaskStatus.BLOCKED, team)
-                TaskTable.status == TaskStatus.BLOCKED,
+        completed_count = await self._get_task_count(
-                *base_filter,
+            TaskStatus.COMPLETED, team, since=week_ago
         )
-        )
-        blocked_result = await self.session.execute(blocked_query)
-        blocked_count = blocked_result.scalar() or 0
 
-        completed_query = select(func.count(TaskTable.id)).where(
-            and_(
-                TaskTable.completed_at >= week_ago,
-                TaskTable.status == TaskStatus.COMPLETED,
-                *base_filter,
-            )
-        )
-        completed_result = await self.session.execute(completed_query)
-        completed_count = completed_result.scalar() or 0
-
-        # Calculate ratios
         blocked_ratio = blocked_count / active_count if active_count > 0 else 0
-
+        status_str = self._determine_health_status(
-        # Determine status
+            blocked_ratio, active_count, completed_count
-        three_tenths = 0.3
+        )
-        fifteen_hundredths = 0.15
-        five = 5
-        if blocked_ratio > three_tenths:
-            status = "critical"
-        elif blocked_ratio > fifteen_hundredths or (
-            active_count > five and completed_count == 0
-        ):
-            status = "slow"
-        else:
-            status = "ok"
 
         return {
-            "status": status,
+            "status": status_str,
             "team": team.value if team else "all",
             "active_tasks": active_count,
             "blocked_tasks": blocked_count,