From ccd949c5c1d6a4d4c20253887f84cf4b9548ff2a Mon Sep 17 00:00:00 2001 From: Renn F Date: Sat, 13 Dec 2025 17:50:47 +0100 Subject: [PATCH] + Radon + Xenon + Pip-Audit + Bandit + Safety --- .safety-project.ini | 5 + roboco/agents/board.py | 79 +++++---- roboco/agents/documenter.py | 84 +++++---- roboco/agents/qa.py | 84 +++++---- roboco/api/routes/dashboard.py | 179 +++++++++---------- roboco/api/routes/messages.py | 147 +++++++++------- roboco/api/routes/notifications.py | 89 +++++----- roboco/bootstrap.py | 104 ++++++----- roboco/events/handlers.py | 268 ++++++++++++++--------------- roboco/mcp/task_server.py | 200 +++++++++++---------- roboco/services/messaging.py | 96 ++++++----- roboco/services/metrics.py | 110 ++++++------ 12 files changed, 767 insertions(+), 678 deletions(-) create mode 100644 .safety-project.ini diff --git a/.safety-project.ini b/.safety-project.ini new file mode 100644 index 00000000..b724a3e8 --- /dev/null +++ b/.safety-project.ini @@ -0,0 +1,5 @@ +[project] +id = roboco +url = /codebases/roboco/findings +name = roboco + diff --git a/roboco/agents/board.py b/roboco/agents/board.py index 0a099752..9d849b0b 100644 --- a/roboco/agents/board.py +++ b/roboco/agents/board.py @@ -681,20 +681,11 @@ 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: - """Perform a specific type of audit.""" - try: - # Query relevant data based on audit type - 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""" + async def _audit_code_quality(self, tasks: list[dict]) -> str | None: + """Audit code quality from completed tasks.""" + if not tasks: + return None + prompt = f""" Analyze these completed tasks for code quality patterns: {chr(10).join(f"- {t.get('title')}: {t.get('description', '')[:100]}" for t in tasks)} @@ -707,36 +698,44 @@ Look for: Report findings or None if all looks good. """ - return await self.think(prompt) + return await self.think(prompt) - elif audit_type == "documentation": - result = await self._api_call( - "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 {count} tasks with incomplete documentation" + async def _audit_documentation(self, tasks: list[dict]) -> str | None: + """Audit documentation completeness.""" + missing_docs = [t for t in tasks if not t.get("documentation_complete")] + if missing_docs: + return f"Found {len(missing_docs)} tasks with incomplete documentation" + return None - elif audit_type == "process_compliance": - # Check for process violations - result = await self._api_call( - "GET", - "/tasks", - params={"status": "completed", "limit": 10}, - ) - tasks = result.get("items", []) - violations = [] - 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)}" + async def _audit_process_compliance(self, tasks: list[dict]) -> str | None: + """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", []) + return await handler(tasks) except Exception as e: self.log.warning("Failed to perform audit", error=str(e)) return None diff --git a/roboco/agents/documenter.py b/roboco/agents/documenter.py index b71ac3ff..4d3e0247 100644 --- a/roboco/agents/documenter.py +++ b/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: - case DocTaskPhase.RECEIVE: - 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 - + result = await self._run_phase(self._doc_context) + return result is True 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 # ========================================================================= diff --git a/roboco/agents/qa.py b/roboco/agents/qa.py index d15efa18..ccfb528a 100644 --- a/roboco/agents/qa.py +++ b/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: - case QATaskPhase.RECEIVE: - 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 - + result = await self._run_phase(self._review_context) + return result is True except Exception as e: - self.log.error("Error in review phase", phase=ctx.phase.value, error=str(e)) - ctx.findings.append(f"Error during review: {e}") + self.log.error( + "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 # ========================================================================= diff --git a/roboco/api/routes/dashboard.py b/roboco/api/routes/dashboard.py index 8dca9fea..09139dbb 100644 --- a/roboco/api/routes/dashboard.py +++ b/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, - key_metrics=key_metrics, - auditor_alerts=auditor_alerts, - roadmap_progress=roadmap_progress, + health_status=await _get_team_health_list(metrics_service), + key_metrics=await _get_key_metrics(metrics_service), + auditor_alerts=_get_auditor_alerts(), + roadmap_progress=await _get_roadmap_progress(db), ) diff --git a/roboco/api/routes/messages.py b/roboco/api/routes/messages.py index cf02845c..a0b81600 100644 --- a/roboco/api/routes/messages.py +++ b/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_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 + session = await _get_session_with_group(db, data.session_id) group = session.group - channel_result = await db.execute( - select(ChannelTable).where(ChannelTable.id == group.channel_id) + channel = await _get_channel_with_access( + 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( - 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", - ) + await _validate_reply_target(db, data.reply_to, data.session_id) - # 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) - - # Update group stats + session.last_activity_at = now group.total_messages += 1 - group.last_activity = datetime.now(UTC) - - # Update channel stats + group.last_activity = now channel.message_count += 1 - channel.last_activity = datetime.now(UTC) + channel.last_activity = now - # Check if session should be closed - 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: + if _check_session_boundaries(session): session.status = SessionStatus.CLOSED - session.closed_at = datetime.now(UTC) + session.closed_at = now group.active_session_id = None await db.flush() diff --git a/roboco/api/routes/notifications.py b/roboco/api/routes/notifications.py index 95cf08d2..39076c1d 100644 --- a/roboco/api/routes/notifications.py +++ b/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,6 +90,49 @@ class NotificationCreateRequest(BaseModel): # ============================================================================= +def _build_notification_query( + agent_id: UUID, + params: ListNotificationsParams, +) -> Any: + """Build the notification query with filters.""" + 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) + + +def _notification_to_response( + n: NotificationTable, + agent_id: UUID, +) -> NotificationResponse: + """Convert a notification to response format.""" + return NotificationResponse( + id=require_uuid(n.id), + type=n.type, + priority=n.priority, + from_agent=require_uuid(n.from_agent), + to_agents=to_python_uuid_list(n.to_agents), + subject=n.subject, + body=n.body, + requires_ack=n.requires_ack, + is_acknowledged=agent_id in n.acked_by, + is_fully_acknowledged=all(a in n.acked_by for a in n.to_agents), + is_read=agent_id in n.read_by, + related_task_id=to_python_uuid(n.related_task_id), + timestamp=n.timestamp, + expires_at=n.expires_at, + ) + + @router.get( "", response_model=NotificationListResponse, @@ -102,53 +145,15 @@ async def list_notifications( 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) - - query = query.order_by(NotificationTable.timestamp.desc()).limit(params.limit) - - result = await db.execute(query) + query = _build_notification_query(agent_id, params) + result: Any = await db.execute(query) notifications = result.scalars().all() - # Count unread and pending ack 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 = [ - NotificationResponse( - id=require_uuid(n.id), - type=n.type, - priority=n.priority, - from_agent=require_uuid(n.from_agent), - to_agents=to_python_uuid_list(n.to_agents), - subject=n.subject, - body=n.body, - requires_ack=n.requires_ack, - is_acknowledged=agent_id in n.acked_by, - is_fully_acknowledged=all(a in n.acked_by for a in n.to_agents), - is_read=agent_id in n.read_by, - related_task_id=to_python_uuid(n.related_task_id), - timestamp=n.timestamp, - expires_at=n.expires_at, - ) - for n in notifications - ] + items = [_notification_to_response(n, agent_id) for n in notifications] return NotificationListResponse( items=items, diff --git a/roboco/bootstrap.py b/roboco/bootstrap.py index b8ba0115..cca823b5 100644 --- a/roboco/bootstrap.py +++ b/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(): - channel_id = channel_ids.get(channel_slug) - if not channel_id: - continue + await _configure_channel_members(session, channel_ids, agent_ids) - # 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) - - 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] + await _add_auditor_silent_access(session, channel_ids, UUIDType(auditor_db_id)) logger.info("Channel memberships configured") diff --git a/roboco/events/handlers.py b/roboco/events/handlers.py index 90e122eb..c78ac7f4 100644 --- a/roboco/events/handlers.py +++ b/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, - agent=agent_id, + event_type=event.type.value, + 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: - # Notify the cell PM - blocker_reason = event.data.get("reason", "Unknown blocker") - team = event.data.get("team") + handlers = { + EventType.TASK_BLOCKED: _handle_task_blocked, + EventType.TASK_AWAITING_QA: _handle_task_awaiting_qa, + EventType.TASK_QA_FAILED: _handle_task_qa_failed, + EventType.TASK_AWAITING_DOCS: _handle_task_awaiting_docs, + } - if team: - pm_id = f"{team[:2]}-pm" # e.g., "backend" -> "be-pm" - await notification_service.send_blocker_notification( - 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, - ) + handler = handlers.get(event.type) + if handler: + await handler(event, task_id, notification_service) 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: - """ - Handle QA result events. +async def _try_resolve_agent_wait( + agent_id: str | None, + 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: - - Resume developer agent if waiting on QA result - - Notify appropriate parties - """ + orchestrator = _BootstrapHolder.orchestrator + if not orchestrator: + 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,107 +215,42 @@ async def handle_qa_result(event: Event) -> None: developer=developer_id, ) - # Check if developer agent is in WAITING_LONG state - - # Get orchestrator instance (if running) - try: - 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 + await _try_resolve_agent_wait( + developer_id, + "qa_result", + {"passed": passed, "notes": event.data.get("qa_notes"), "task_id": task_id}, + ) async def handle_blocker_resolved(event: Event) -> None: - """ - Handle blocker resolution events. - - Triggers: - - Resume blocked agent - - Update task status - """ + """Handle blocker resolution events.""" task_id = event.data.get("task_id") agent_id = event.data.get("agent_id") resolution = event.data.get("resolution", "Resolved") - logger.info( - "Blocker resolved", - task_id=task_id, - agent=agent_id, + logger.info("Blocker resolved", task_id=task_id, agent=agent_id) + + await _try_resolve_agent_wait( + agent_id, + "blocker_resolution", + {"details": resolution, "task_id": task_id}, ) - # Resume agent if waiting - try: - from roboco.bootstrap import _BootstrapHolder # noqa: PLC0415 - - 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. - - Triggers: - - Resume agent waiting for answer - """ + """Handle question answered events.""" question_id = event.data.get("question_id") agent_id = event.data.get("asking_agent") answer = event.data.get("answer") - logger.info( - "Question answered", - question_id=question_id, - agent=agent_id, + logger.info("Question answered", question_id=question_id, agent=agent_id) + + await _try_resolve_agent_wait( + agent_id, + "answer", + {"answer": answer, "question_id": question_id}, ) - # Resume agent if waiting - try: - from roboco.bootstrap import _BootstrapHolder # noqa: PLC0415 - - 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: """Register all default event handlers.""" diff --git a/roboco/mcp/task_server.py b/roboco/mcp/task_server.py index 88ae32f0..836e08f4 100644 --- a/roboco/mcp/task_server.py +++ b/roboco/mcp/task_server.py @@ -259,61 +259,81 @@ async def _handle_task_get(task_id: str) -> dict[str, Any]: return _format_task_response(task, next_step, guidance) +def _check_blocking_tasks(active_tasks: list[dict]) -> dict[str, Any] | None: + """Check for blocking active tasks. Returns error or None.""" + blocking_statuses = ["claimed", "in_progress", "verifying"] + blocking = [t for t in active_tasks if t.get("status") in blocking_statuses] + if blocking: + return _format_error_response( + "ALREADY_ACTIVE", + f"You already have an active task: {blocking[0]['id']}. " + "Complete or pause it before claiming a new task.", + {"active_task_id": blocking[0]["id"]}, + ) + return None + + +def _check_paused_tasks(active_tasks: list[dict]) -> dict[str, Any] | None: + """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)} paused task(s). " + "Resume paused work before claiming new tasks.", + {"paused_task_ids": [t["id"] for t in paused]}, + ) + return None + + +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", + f"Cannot claim task in '{task.get('status')}' status. " + "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: - # Check for existing active tasks 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() - # 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"{blocking_tasks[0]['id']}. " - "Complete or pause it before claiming a new task.", - {"active_task_id": blocking_tasks[0]["id"]}, - ) + if error := _check_blocking_tasks(active_tasks): + return error + if error := _check_paused_tasks(active_tasks): + return error - # Check for paused tasks - paused_tasks = [t for t in active_tasks if t.get("status") == "paused"] - if paused_tasks: - return _format_error_response( - "PAUSED_TASKS_EXIST", - f"You have {len(paused_tasks)} paused task(s). " - "Resume paused work before claiming new tasks.", - {"paused_task_ids": [t["id"] for t in paused_tasks]}, - ) - - # 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() - if task.get("status") != "pending": - return _format_error_response( - "INVALID_STATE", - f"Cannot claim task in '{task.get('status')}' status. " - "Only 'pending' tasks can be claimed.", - {"current_status": task.get("status")}, - ) + 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: - 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() + project = await _get_project_context(claimed_task["project_id"]) return _format_task_response( claimed_task, @@ -344,70 +358,71 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]: ) +def _validate_task_ownership(task: dict, agent_id: str) -> dict[str, Any] | None: + """Validate agent owns the task. Returns error or None.""" + 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 '{task.get('status')}' status. " + "Task must be 'claimed'.", + {"current_status": task.get("status")}, + ) + return None + + +def _build_plan_data(plan_params: dict[str, Any]) -> dict[str, Any]: + """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(plan_params["sub_tasks"]) + ], + "risks": [{"description": r} for r in (plan_params.get("risks") or [])], + "open_questions": [ + {"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. - - 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") - + """Handle task planning.""" 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 error := _validate_task_ownership(task, agent_id): + return error + if error := _validate_task_status_claimed(task): + return error - 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")}, - ) - - if task.get("status") != "claimed": - return _format_error_response( - "INVALID_STATE", - f"Cannot submit plan for task in " - f"'{task.get('status')}' status. " - "Task must be 'claimed'.", - {"current_status": task.get("status")}, - ) - - # Submit the plan - plan_data = { - "approach": approach, - "sub_tasks": [ - { - "title": st.get("title", ""), - "description": st.get("description", ""), - "order": i, - } - for i, st in enumerate(sub_tasks) - ], - "risks": [{"description": r} for r in (risks or [])], - "open_questions": [ - {"question": q, "answered": False} for q in (open_questions or []) - ], - } - + 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)} " - "open question(s). " + f"Plan saved but you have {len(open_questions)} open question(s). " "Ask these questions in your cell channel before starting. " "Do NOT proceed until questions are answered.", ) diff --git a/roboco/services/messaging.py b/roboco/services/messaging.py index d8849134..1fb536f6 100644 --- a/roboco/services/messaging.py +++ b/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 = await self.get_session(req.session_id) - if not session: - raise ValueError(f"Session {req.session_id} not found") + session, group, channel = await self._get_message_context(req.session_id) - 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) - if not reply_msg or reply_msg.session_id != req.session_id: - raise ValueError("Reply target not found in this session") + await self._validate_reply_target(req.reply_to, req.session_id) - # 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) - - # 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) - + self._update_message_stats(session, group, channel, content_length) 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") diff --git a/roboco/services/metrics.py b/roboco/services/metrics.py index b468ead6..da175ebb 100644 --- a/roboco/services/metrics.py +++ b/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 = [ + TaskStatus.CLAIMED, + TaskStatus.IN_PROGRESS, + TaskStatus.VERIFYING, + TaskStatus.AWAITING_QA, + TaskStatus.BLOCKED, + ] - # 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_count = await self._get_task_count(active_statuses, team) + blocked_count = await self._get_task_count(TaskStatus.BLOCKED, team) + completed_count = await self._get_task_count( + TaskStatus.COMPLETED, team, since=week_ago ) - active_result = await self.session.execute(active_query) - active_count = active_result.scalar() or 0 - blocked_query = select(func.count(TaskTable.id)).where( - and_( - TaskTable.status == TaskStatus.BLOCKED, - *base_filter, - ) - ) - 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 - - # Determine status - 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" + status_str = self._determine_health_status( + blocked_ratio, active_count, completed_count + ) return { - "status": status, + "status": status_str, "team": team.value if team else "all", "active_tasks": active_count, "blocked_tasks": blocked_count,