diff --git a/roboco/agents_config.py b/roboco/agents_config.py index 6ed8041b..9c9b0240 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -27,6 +27,7 @@ different purposes. MCP is coarse-grained (tool-level), API is fine-grained from typing import Final +from roboco.models.base import NotificationPriority, NotificationType from roboco.seeds.initial_data import AGENT_UUIDS # Reverse mapping: UUID -> slug (computed from seeds) @@ -390,3 +391,10 @@ NOTIFICATION_PERMISSIONS: Final[dict[str, dict]] = { "can_send": False, }, } + +VALID_NOTIFICATION_TYPES: Final[frozenset[str]] = frozenset( + t.value for t in NotificationType +) +VALID_NOTIFICATION_PRIORITIES: Final[frozenset[str]] = frozenset( + p.value for p in NotificationPriority +) diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 79f5d6ea..0e0f7e0c 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -4,7 +4,7 @@ Task API Routes Full CRUD operations and lifecycle management for tasks. """ -from typing import Annotated +from typing import Annotated, Any from uuid import UUID from fastapi import APIRouter, Body, HTTPException, Query, status @@ -25,6 +25,7 @@ from roboco.api.schemas.tasks import ( CheckpointRequest, ClaimRequest, CommitRequest, + CompleteTaskRequest, EscalateRequest, EscalateResponse, ProgressRequest, @@ -967,12 +968,17 @@ async def complete_task( db: DbSession, agent: CurrentAgentContext, permissions: PermissionServiceDep, + data: Annotated[CompleteTaskRequest | None, Body()] = None, ) -> TaskResponse: """Mark task as completed (PM only). Two completion paths: 1. Developer work: task must be in awaiting_pm_review (went through QA/Docs) 2. PM's own task: task can be in_progress if assigned to the completing PM + + PM Override for cancelled subtasks: + If force_with_cancelled=True, PM can complete despite cancelled subtasks. + Requires justification. Does NOT apply to pending/in_progress subtasks. """ service = get_task_service(db) task = await service.get(task_id) @@ -981,7 +987,7 @@ async def complete_task( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" ) - # Only PMs can complete tasks + # Permission check - only PMs can complete tasks can_close = permissions.can_perform_task_action(agent, TaskAction.CLOSE, task.team) if not can_close: raise HTTPException( @@ -989,38 +995,28 @@ async def complete_task( detail="Only PMs can complete tasks", ) - # Check ALL subtasks are completed before completing parent - # Cancelled subtasks block completion - they must be resolved first - subtasks = await service.get_subtasks(task_id) - incomplete_subtasks = [ - st - for st in subtasks - if st.status != TaskStatus.COMPLETED - ] - if incomplete_subtasks: - max_titles_shown = 3 - incomplete_info = [ - f"{st.title} ({st.status.value})" - for st in incomplete_subtasks[:max_titles_shown] - ] - detail = ( - f"Cannot complete task - {len(incomplete_subtasks)} subtask(s) " - f"not completed: {', '.join(incomplete_info)}" - ) - if len(incomplete_subtasks) > max_titles_shown: - detail += f" (+{len(incomplete_subtasks) - max_titles_shown} more)" + # Extract request data + force_complete = data.force_with_cancelled if data else False + justification = data.justification if data else None + + # Validate justification if force is requested + if force_complete and not justification: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=detail, + detail="force_with_cancelled requires justification", ) - # Pass agent_id so service can check if PM is completing their own task - task = await service.complete(task_id, agent_id=agent.agent_id) + task = await service.complete( + task_id, + agent_id=agent.agent_id, + force_with_cancelled=force_complete, + justification=justification, + ) if not task: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail="Cannot complete task - must be in awaiting_pm_review or " - "in_progress (if your own task)", + detail="Cannot complete task - check status and subtasks. " + "Use force_with_cancelled=true if only cancelled subtasks remain.", ) await db.commit() return task_to_response(task) @@ -1234,12 +1230,20 @@ async def substitute_task( # Determine new status based on reason new_status = _REASON_TO_STATUS.get(reason, TaskStatus.PENDING) - # Update task - update_data = { + # QA/Documenter completing their own work goes to PM review (can't self-review) + if reason == SubstituteReason.TASK_COMPLETE and agent.role in ("qa", "documenter"): + new_status = TaskStatus.AWAITING_PM_REVIEW + + # Preserve original developer for self-review prevention when going to QA + update_data: dict[str, Any] = { "status": new_status.value, "assigned_to": None, # Clear assignment "dev_notes": f"[SUBSTITUTE] Reason: {reason.value}\n{data.details}", } + if new_status == TaskStatus.AWAITING_QA and task.assigned_to: + # Set original_developer BEFORE clearing assigned_to + update_data["quick_context"] = f"original_developer:{task.assigned_to}" + task = await service.update(task_id, **update_data) if not task: raise HTTPException( diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index ead722ab..9eca4ce0 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -303,6 +303,22 @@ class QANotes(BaseModel): notes: str +class CompleteTaskRequest(BaseModel): + """Request to complete a task with optional force flag.""" + + force_with_cancelled: bool = Field( + default=False, + description="Force complete even if some subtasks are cancelled. " + "PM takes responsibility for judging work is done. " + "Only applies to cancelled subtasks, not pending/in_progress.", + ) + justification: str | None = Field( + default=None, + description="Required when force_with_cancelled=True. " + "PM's justification for completing despite cancelled subtasks.", + ) + + class SoftBlockRequest(BaseModel): """Request to soft-block a task due to an external factor.""" diff --git a/roboco/enforcement/task_lifecycle.py b/roboco/enforcement/task_lifecycle.py index 255a70e0..a9e46c0b 100644 --- a/roboco/enforcement/task_lifecycle.py +++ b/roboco/enforcement/task_lifecycle.py @@ -34,8 +34,8 @@ VALID_TRANSITIONS: dict[str, list[str]] = { "awaiting_documentation", "cancelled", ], - # Needs revision - back to work or cancel - "needs_revision": ["in_progress", "cancelled"], + # Needs revision - developer claims, works, or PM cancels + "needs_revision": ["claimed", "in_progress", "cancelled"], # Awaiting QA - QA claims, passes, fails, or blocks "awaiting_qa": [ "claimed", @@ -46,8 +46,8 @@ VALID_TRANSITIONS: dict[str, list[str]] = { ], # Awaiting documentation - documenter claims or marks done "awaiting_documentation": ["claimed", "awaiting_pm_review", "cancelled"], - # Awaiting PM review - PM reviews and completes, or cancels - "awaiting_pm_review": ["completed", "cancelled"], + # Awaiting PM review - PM claims to review, then completes or cancels + "awaiting_pm_review": ["claimed", "completed", "cancelled"], # Terminal states - cannot transition out "completed": [], "cancelled": [], @@ -73,6 +73,8 @@ ROLE_RESTRICTED_TRANSITIONS: dict[tuple[str, str], list[str]] = { # Only documenter can claim docs tasks and mark complete ("awaiting_documentation", "claimed"): ["documenter"], ("awaiting_documentation", "awaiting_pm_review"): ["documenter"], + # Only PM can claim PM review tasks + ("awaiting_pm_review", "claimed"): _CANCEL_ROLES, # Only PM can complete tasks (either after PM review or their own work) ("awaiting_pm_review", "completed"): _CANCEL_ROLES, ("in_progress", "completed"): _CANCEL_ROLES, # PM completing their own task diff --git a/roboco/mcp/notify_server.py b/roboco/mcp/notify_server.py index 5bac6d25..2dea5cbe 100644 --- a/roboco/mcp/notify_server.py +++ b/roboco/mcp/notify_server.py @@ -25,12 +25,15 @@ from mcp.server.fastmcp import FastMCP from roboco.agents_config import ( NOTIFICATION_PERMISSIONS, + VALID_NOTIFICATION_PRIORITIES, + VALID_NOTIFICATION_TYPES, can_send_notifications, get_agent_cell, get_agent_role, ) from roboco.mcp.schemas import SendNotificationInput from roboco.mcp.utils import ApiClient, format_error_response +from roboco.models.base import NotificationPriority, NotificationType # ============================================================================= # HELPER FUNCTIONS @@ -82,13 +85,6 @@ def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str return can_send, reason -# Valid notification types and priorities -VALID_NOTIFICATION_TYPES = frozenset( - ["info", "alert", "task", "escalation", "approval"] -) -VALID_PRIORITIES = frozenset(["low", "normal", "high", "urgent"]) - - def _validate_notification_type(notification_type: str) -> dict[str, Any] | None: """Validate notification type. Returns error dict or None if valid.""" if notification_type not in VALID_NOTIFICATION_TYPES: @@ -101,10 +97,11 @@ def _validate_notification_type(notification_type: str) -> dict[str, Any] | None def _validate_priority(priority: str) -> dict[str, Any] | None: """Validate priority. Returns error dict or None if valid.""" - if priority not in VALID_PRIORITIES: + if priority not in VALID_NOTIFICATION_PRIORITIES: + valid = sorted(VALID_NOTIFICATION_PRIORITIES) return format_error_response( "INVALID_PRIORITY", - f"Invalid priority. Must be one of: {sorted(VALID_PRIORITIES)}", + f"Invalid priority. Must be one of: {valid}", ) return None @@ -392,8 +389,8 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP: recipients=[escalate_to], subject=f"[ESCALATION] {subject}", body=description, - notification_type="escalation", - priority="high", + notification_type=NotificationType.BLOCKER_ESCALATION.value, + priority=NotificationPriority.HIGH.value, requires_ack=True, related_task_id=task_id, ) @@ -416,8 +413,8 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP: recipients=[approver], subject=f"[APPROVAL NEEDED] {subject}", body=what_needs_approval, - notification_type="approval", - priority="normal", + notification_type=NotificationType.REVIEW_REQUEST.value, + priority=NotificationPriority.NORMAL.value, requires_ack=True, related_task_id=task_id, ) diff --git a/roboco/mcp/task_server.py b/roboco/mcp/task_server.py index f2f719f9..01a30cd8 100644 --- a/roboco/mcp/task_server.py +++ b/roboco/mcp/task_server.py @@ -551,7 +551,11 @@ def _register_pm_completion_tools( """Register PM-only task completion tools.""" @mcp.tool() - async def roboco_task_complete(task_id: str) -> dict[str, Any]: + async def roboco_task_complete( + task_id: str, + force_with_cancelled: bool = False, + justification: str | None = None, + ) -> dict[str, Any]: """ Mark task as completed (PM only). @@ -561,14 +565,24 @@ def _register_pm_completion_tools( ENFORCEMENT: - Only PMs can use this tool - Task must be in 'awaiting_pm_review' status + - All subtasks must be completed (or use force_with_cancelled) + + PM Override for cancelled subtasks: + If some subtasks were cancelled but PM judges work is done anyway, + use force_with_cancelled=True with justification explaining why. + Only works if ALL non-completed subtasks are cancelled. Args: task_id: The task UUID + force_with_cancelled: Override cancelled subtask check + justification: Required when force_with_cancelled=True Returns: Completed task """ - return await handle_task_complete(client, task_id, agent_id) + return await handle_task_complete( + client, task_id, agent_id, force_with_cancelled, justification + ) def _register_pm_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None: diff --git a/roboco/mcp/tasks/handlers/lifecycle.py b/roboco/mcp/tasks/handlers/lifecycle.py index d0e659d4..bea8c152 100644 --- a/roboco/mcp/tasks/handlers/lifecycle.py +++ b/roboco/mcp/tasks/handlers/lifecycle.py @@ -61,8 +61,13 @@ async def handle_docs_complete( ): return error - payload = {"notes": doc_notes} if doc_notes else {} - docs_resp = await client.post(f"/tasks/{task_id}/docs-complete", json=payload) + # Only send payload if doc_notes provided (QANotes.notes is required) + if doc_notes: + docs_resp = await client.post( + f"/tasks/{task_id}/docs-complete", json={"notes": doc_notes} + ) + else: + docs_resp = await client.post(f"/tasks/{task_id}/docs-complete") if not docs_resp.ok: return format_error_response( @@ -98,26 +103,23 @@ async def _check_children_completed( Returns error if any children are not completed, None if OK. """ try: - # Fetch children/subtasks for this task resp = await client.get(f"/tasks/{task_id}/subtasks") if not resp.ok: - # If endpoint doesn't exist or fails, skip check (backwards compat) return None subtasks = resp.json() if not subtasks: - return None # No children, OK to complete + return None - incomplete: list[dict[str, str]] = [] - for subtask in subtasks: - subtask_status = subtask.get("status") - # ONLY "completed" is acceptable - cancelled/pending/etc. block completion - if subtask_status != "completed": - incomplete.append({ - "id": str(subtask.get("id", "unknown")), - "title": subtask.get("title", "Untitled"), - "status": subtask_status or "unknown", - }) + incomplete = [ + { + "id": str(subtask.get("id", "unknown")), + "title": subtask.get("title", "Untitled"), + "status": subtask.get("status") or "unknown", + } + for subtask in subtasks + if subtask.get("status") != "completed" + ] if incomplete: return format_error_response( @@ -134,18 +136,59 @@ async def _check_children_completed( return None except Exception: - # If check fails for any reason, allow completion (backwards compat) return None +async def _validate_children_or_force( + client: ApiClient, task_id: str, force: bool, justification: str | None +) -> dict[str, Any] | None: + """Validate children completion or force override. Returns error or None.""" + if force: + if not justification: + return format_error_response( + "JUSTIFICATION_REQUIRED", + "force_with_cancelled requires justification explaining " + "why cancelled subtasks don't block completion.", + ) + return None + return await _check_children_completed(client, task_id) + + +def _validate_completion_status( + task: dict[str, Any], agent_id: str +) -> dict[str, Any] | None: + """Validate task is in completable status. Returns error or None.""" + current_status = task.get("status") + is_own_task = current_status == "in_progress" and _is_pm_own_task(task, agent_id) + is_review_task = current_status == "awaiting_pm_review" + + if is_own_task or is_review_task: + return None + + return format_error_response( + "INVALID_STATE", + f"Cannot complete task in '{current_status}' status. " + "Expected 'awaiting_pm_review' (dev work) or 'in_progress' (own task).", + {"current_status": current_status}, + ) + + async def handle_task_complete( - client: ApiClient, task_id: str, agent_id: str + client: ApiClient, + task_id: str, + agent_id: str, + force_with_cancelled: bool = False, + justification: str | None = None, ) -> dict[str, Any]: """Handle task completion (PM only). Two completion paths: 1. Completing developer work: task must be in 'awaiting_pm_review' 2. Completing PM's own task: task can be in 'in_progress' if assigned to PM + + PM Override for cancelled subtasks: + Use force_with_cancelled=True with justification to complete despite + cancelled subtasks. Only works if ALL non-completed children are cancelled. """ if error := _validate_pm_role(agent_id, "complete tasks"): return error @@ -155,25 +198,19 @@ async def handle_task_complete( return error assert task is not None - current_status = task.get("status") - - # PM completing their own task - allow from in_progress - if current_status == "in_progress" and _is_pm_own_task(task, agent_id): - pass # Valid - PM completing their own work - # Normal path - developer work went through QA/Docs - elif current_status != "awaiting_pm_review": - return format_error_response( - "INVALID_STATE", - f"Cannot complete task in '{current_status}' status. " - "Expected 'awaiting_pm_review' (dev work) or 'in_progress' (own task).", - {"current_status": current_status}, - ) - - # Check all children are completed before allowing parent completion - if error := await _check_children_completed(client, task_id): + if error := _validate_completion_status(task, agent_id): return error - complete_resp = await client.post(f"/tasks/{task_id}/complete") + if error := await _validate_children_or_force( + client, task_id, force_with_cancelled, justification + ): + return error + + payload: dict[str, Any] = {} + if force_with_cancelled: + payload = {"force_with_cancelled": True, "justification": justification} + + complete_resp = await client.post(f"/tasks/{task_id}/complete", json=payload) if not complete_resp.ok: return format_error_response( "COMPLETE_FAILED", diff --git a/roboco/mcp/tasks/handlers/management.py b/roboco/mcp/tasks/handlers/management.py index 7e5c1bb1..79d87a00 100644 --- a/roboco/mcp/tasks/handlers/management.py +++ b/roboco/mcp/tasks/handlers/management.py @@ -77,6 +77,60 @@ def validate_assignee_can_work_on_team( return None +def get_role_mismatch_warning(task: dict[str, Any], assignee: str) -> str | None: + """Check if assignee role matches the task type and return warning if mismatch. + + Returns a warning string or None if no mismatch. + This is a SOFT warning, not a blocking error - PM has flexibility. + """ + assignee_role = get_agent_role(assignee) + task_status = task.get("status") + task_title = task.get("title", "").lower() + + # Role suggestions based on task status + status_role_map = { + "awaiting_qa": "qa", + "awaiting_documentation": "documenter", + "awaiting_pm_review": ("cell_pm", "main_pm"), + } + + # Check status-based role matching + if task_status in status_role_map: + expected = status_role_map[task_status] + if isinstance(expected, tuple): + if assignee_role not in expected: + return ( + f"Task is {task_status} - typically assigned to " + f"{' or '.join(expected)}, not {assignee_role}." + ) + elif assignee_role != expected: + return ( + f"Task is {task_status} - typically assigned to " + f"{expected}, not {assignee_role}." + ) + + # Check title-based hints for pending tasks + if task_status == "pending": + # QA-related keywords in title + qa_keywords = ["qa", "test", "validation", "quality", "review"] + if any(kw in task_title for kw in qa_keywords) and assignee_role != "qa": + return ( + f"Task title suggests QA work but assignee is {assignee_role}. " + "Consider assigning to a QA agent." + ) + + # Docs-related keywords in title + doc_keywords = ["doc", "documentation", "readme", "guide", "reference"] + is_doc_task = any(kw in task_title for kw in doc_keywords) + if is_doc_task and assignee_role != "documenter": + return ( + f"Task title suggests documentation but assignee is {assignee_role}. " + "Consider assigning to a documenter." + ) + + return None + + def validate_cell_pm_assignment( role: str, agent_team: str | None, @@ -275,6 +329,13 @@ async def handle_task_create( task = assigned_task guidance = _format_create_guidance(task, input_data.assigned_to) + + # Check for role mismatch and add warning if found + if input_data.assigned_to: + role_warning = get_role_mismatch_warning(task, input_data.assigned_to) + if role_warning: + guidance += f"\n\n⚠️ ROLE WARNING: {role_warning}" + return format_task_response(task, "CREATED", guidance) @@ -312,6 +373,12 @@ async def handle_task_assign( f"Task assigned to {input_data.assignee} and set to pending. " "Orchestrator will spawn them to claim and work on it." ) + + # Check for role mismatch and add warning if found + role_warning = get_role_mismatch_warning(task, input_data.assignee) + if role_warning: + guidance += f"\n\n⚠️ ROLE WARNING: {role_warning}" + return format_task_response(assigned_task, "ASSIGNED", guidance) diff --git a/roboco/mcp/tasks/handlers/substitute.py b/roboco/mcp/tasks/handlers/substitute.py index 1da07af9..90689218 100644 --- a/roboco/mcp/tasks/handlers/substitute.py +++ b/roboco/mcp/tasks/handlers/substitute.py @@ -21,11 +21,13 @@ from roboco.models import SubstituteReason, TaskStatus HTTP_NOT_FOUND = 404 # Map substitute reasons to target task statuses +# NOTE: OUT_OF_SCOPE_ROLE goes to awaiting_pm_review to avoid reassignment loops +# (e.g., documenter can't self-document → substitute → gets reassigned same doc → loop) REASON_TO_STATUS: dict[SubstituteReason, TaskStatus] = { SubstituteReason.TASK_COMPLETE: TaskStatus.AWAITING_QA, SubstituteReason.LOW_CONTEXT: TaskStatus.PENDING, SubstituteReason.OUT_OF_SCOPE_TEAM: TaskStatus.PENDING, - SubstituteReason.OUT_OF_SCOPE_ROLE: TaskStatus.PENDING, + SubstituteReason.OUT_OF_SCOPE_ROLE: TaskStatus.AWAITING_PM_REVIEW, # PM decides SubstituteReason.MAX_RETRIES: TaskStatus.PENDING, SubstituteReason.BLOCKED_EXTERNAL: TaskStatus.BLOCKED, } @@ -190,6 +192,11 @@ async def handle_task_substitute( "Task marked as blocked. PM will be notified. " "You are now free to claim new work with roboco_task_scan()." ) + elif substitute_reason == SubstituteReason.OUT_OF_SCOPE_ROLE: + next_action = ( + "Task sent to PM for reassignment (role conflict). " + "You are now free to claim new work with roboco_task_scan()." + ) else: next_action = ( "Task released and will be reassigned. " diff --git a/roboco/mcp/tasks/utils.py b/roboco/mcp/tasks/utils.py index 5bfccdd6..5ca39f81 100644 --- a/roboco/mcp/tasks/utils.py +++ b/roboco/mcp/tasks/utils.py @@ -91,7 +91,8 @@ def get_next_step_guidance(status: str) -> tuple[str, str]: "needs_revision": ( "FIX_ISSUES", "QA found issues. Read the QA notes carefully. " - "Fix all issues, then re-submit for QA.", + "Call roboco_task_start() to resume work, fix all issues, " + "then re-submit for QA with roboco_task_submit_qa().", ), "awaiting_documentation": ( "DOCUMENT", diff --git a/roboco/models/audit.py b/roboco/models/audit.py index 9d9e5a79..d573e20f 100644 --- a/roboco/models/audit.py +++ b/roboco/models/audit.py @@ -30,6 +30,9 @@ class AuditEventType(str, Enum): ACCESS_GRANTED = "access_granted" ACCESS_REVOKED = "access_revoked" + # PM override events + PM_OVERRIDE = "pm_override" + @dataclass class PermissionDenialContext: diff --git a/roboco/models/organization.py b/roboco/models/organization.py index 031178f9..3976fc7e 100644 --- a/roboco/models/organization.py +++ b/roboco/models/organization.py @@ -4,33 +4,35 @@ Organization Models Defines organizational structures: Cell, Board, Organization. """ -from typing import Any +from typing import TYPE_CHECKING from pydantic import BaseModel from roboco.models import Team +if TYPE_CHECKING: + from roboco.agents.base import Agent + class Cell(BaseModel): """A cell in the organization (backend, frontend, ux_ui).""" name: str team: Team - pm: Any # Agent - developers: list[Any] = [] # List of Agent - qa: Any | None = None # Agent - documenter: Any | None = None # Agent + pm: "Agent" + developers: list["Agent"] = [] + qa: "Agent | None" = None + documenter: "Agent | None" = None model_config = {"arbitrary_types_allowed": True} class Board(BaseModel): - """The board of the organization.""" + """The board of the organization (3 agents reporting to CEO).""" - product_owner: Any # Agent - head_marketing: Any # Agent - auditor: Any # Agent - main_pm: Any # Agent + product_owner: "Agent" + head_marketing: "Agent" + auditor: "Agent" model_config = {"arbitrary_types_allowed": True} @@ -39,6 +41,9 @@ class Organization(BaseModel): """The complete organization structure.""" board: Board - cells: dict[str, Cell] = {} + main_pm: "Agent" + backend_cell: Cell + frontend_cell: Cell + ux_cell: Cell model_config = {"arbitrary_types_allowed": True} diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index e6d1b3ed..cb27db21 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -270,6 +270,8 @@ class AgentOrchestrator: "mcp__roboco-notify__*", # Journal - always needed for reflection "mcp__roboco-journal__*", + # Knowledge base/RAG - needed for research + "mcp__roboco-optimal__*", # File operations for documenters and developers # Note: // prefix = absolute path (container paths like /app/docs) "Write(//app/docs/**)", @@ -1696,11 +1698,12 @@ Start now: roboco_task_get("{task_id}") to review and close the parent task. Monitors: tasks with completed subtasks but parent still open - Spawns: be-pm, fe-pm, ux-pm (based on parent team) + Spawns: be-pm, fe-pm, ux-pm, main-pm (based on parent team) """ # Find parent tasks that might have children ready for closure # Include "paused" - PM pauses while waiting, respawned when subtasks done - parent_statuses = ["claimed", "in_progress", "paused"] + # Include "awaiting_pm_review" - parent awaiting review when children done + parent_statuses = ["claimed", "in_progress", "paused", "awaiting_pm_review"] for status in parent_statuses: tasks = await self._fetch_tasks(client, status) @@ -1722,8 +1725,12 @@ Start now: roboco_task_get("{task_id}") continue # Not ready for closure # Parent has all subtasks completed - spawn PM to close - team = task.get("team", "backend") - pm_id = self._TEAM_PM_MAP.get(team, "be-pm") + team = task.get("team") + if team in ["backend", "frontend", "ux_ui"]: + pm_id = self._TEAM_PM_MAP.get(team, "be-pm") + else: + # main_pm, board, or no team → Main PM handles closure + pm_id = "main-pm" if self._is_agent_active(pm_id): continue # PM already working @@ -1985,18 +1992,15 @@ Begin with step 1: roboco_task_get("{task_id}") async def _dispatch_pm_review_work(self, client: httpx.AsyncClient) -> None: """ - Dispatch PM review work to cell PMs. + Dispatch PM review work to cell PMs or Main PM. Monitors: awaiting_pm_review tasks - Spawns: be-pm, fe-pm, ux-pm + Spawns: be-pm, fe-pm, ux-pm, main-pm """ tasks = await self._fetch_tasks(client, "awaiting_pm_review") for task in tasks: team = task.get("team") - if team not in ["backend", "frontend", "ux_ui"]: - continue - assigned_to = task.get("assigned_to") # If already assigned, check if that agent is running @@ -2012,8 +2016,13 @@ Begin with step 1: roboco_task_get("{task_id}") ) continue - # Unassigned task - select PM for this team - pm_id = self._TEAM_PM_MAP.get(team, "be-pm") + # Unassigned task - select PM based on team + # Cell tasks go to Cell PM, cross-cell/main_pm tasks go to Main PM + if team in ["backend", "frontend", "ux_ui"]: + pm_id = self._TEAM_PM_MAP.get(team, "be-pm") + else: + # main_pm, board, or no team → Main PM handles it + pm_id = "main-pm" if self._is_agent_active(pm_id): continue diff --git a/roboco/services/audit.py b/roboco/services/audit.py index c760e475..52e41044 100644 --- a/roboco/services/audit.py +++ b/roboco/services/audit.py @@ -157,6 +157,30 @@ class AuditService(SingletonService): timestamp=datetime.now(UTC).isoformat(), ) + async def log_pm_override( + self, + agent_id: str | UUID, + task_id: str | UUID, + action: str, + justification: str, + cancelled_subtask_ids: list[str] | None = None, + ) -> None: + """Log when a PM uses an override capability. + + PM overrides are legitimate but need auditing - e.g., completing + a task despite cancelled subtasks when PM judges work is done. + """ + self.log.info( + "PM override used", + event_type=AuditEventType.PM_OVERRIDE.value, + agent_id=str(agent_id), + task_id=str(task_id), + action=action, + justification=justification, + cancelled_subtask_ids=cancelled_subtask_ids, + timestamp=datetime.now(UTC).isoformat(), + ) + # ============================================================================= # SINGLETON INSTANCE diff --git a/roboco/services/task.py b/roboco/services/task.py index 81d8f1cd..b0dedfb6 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -65,9 +65,18 @@ def _get_valid_claim_statuses( if allow_reassign: statuses.add(TaskStatus.CLAIMED) return statuses + elif role in ("cell_pm", "main_pm"): + # PMs can claim: + # - PENDING: standard task claiming + # - AWAITING_PM_REVIEW: tasks submitted for PM approval + statuses = {TaskStatus.PENDING, TaskStatus.AWAITING_PM_REVIEW} + if allow_reassign: + statuses.add(TaskStatus.CLAIMED) + return statuses else: - # Developer, PM, and other roles - statuses = {TaskStatus.PENDING} + # Developer and other roles + # NEEDS_REVISION for when task is reassigned after QA rejection + statuses = {TaskStatus.PENDING, TaskStatus.NEEDS_REVISION} if allow_reassign: statuses.add(TaskStatus.CLAIMED) return statuses @@ -406,6 +415,20 @@ class TaskService(BaseService): return "agent not in task's team" return None + def _validate_not_self_review( + self, task: TaskTable, agent: AgentTable | None, agent_id: UUID + ) -> str | None: + """Prevent QA/Documenter from claiming tasks they developed.""" + if not agent or not agent.role: + return None + role = agent.role.value if hasattr(agent.role, "value") else str(agent.role) + if role not in ("qa", "documenter"): + return None + original_dev = extract_original_developer(task.quick_context) + if original_dev and original_dev == str(agent_id): + return "cannot review your own work (self-review)" + return None + def _set_original_developer_context( self, task: TaskTable, agent: AgentTable | None ) -> None: @@ -453,7 +476,12 @@ class TaskService(BaseService): self.log.warning(f"Cannot claim task - {error}", task_id=str(task_id)) return None - # Set context for QA/Documenter claims + # Prevent self-review: QA/Documenter cannot claim tasks they developed + if error := self._validate_not_self_review(task, agent, agent_id): + self.log.warning(f"Cannot claim task - {error}", task_id=str(task_id)) + return None + + # Set context for QA/Documenter claims (only if not already set) self._set_original_developer_context(task, agent) # Update assignment @@ -466,6 +494,7 @@ class TaskService(BaseService): TaskStatus.PENDING, TaskStatus.AWAITING_QA, TaskStatus.AWAITING_DOCUMENTATION, + TaskStatus.AWAITING_PM_REVIEW, } if task.status in claimable_statuses: self._validate_and_set_status(task, TaskStatus.CLAIMED, agent_role) @@ -922,6 +951,8 @@ class TaskService(BaseService): self, task_id: UUID, agent_id: UUID | None = None, + force_with_cancelled: bool = False, + justification: str | None = None, ) -> TaskTable | None: """ Mark task as completed (PM only). @@ -930,10 +961,16 @@ class TaskService(BaseService): 1. Developer work: task must be in AWAITING_PM_REVIEW (went through QA/Docs) 2. PM's own task: task can be IN_PROGRESS if assigned to the completing PM + PM Override for cancelled subtasks: + Use force_with_cancelled=True with justification to complete despite + cancelled subtasks. Only works if ALL non-completed children are cancelled. + Args: task_id: The task to complete agent_id: Optional agent UUID - if provided, allows PM to complete their own in_progress tasks + force_with_cancelled: Override cancelled subtask check + justification: Required when force_with_cancelled=True Returns: The completed task or None if completion not allowed @@ -961,6 +998,37 @@ class TaskService(BaseService): ) return None + # Check subtasks completion + subtasks = await self.get_subtasks(task_id) + incomplete_subtasks = [ + st for st in subtasks if st.status != TaskStatus.COMPLETED + ] + + if incomplete_subtasks: + cancelled_only = all( + st.status == TaskStatus.CANCELLED for st in incomplete_subtasks + ) + + if force_with_cancelled and cancelled_only and justification: + # Log the PM override + self.log.info( + "PM override: completing task with cancelled subtasks", + task_id=str(task_id), + agent_id=str(agent_id) if agent_id else None, + justification=justification, + cancelled_subtask_ids=[str(st.id) for st in incomplete_subtasks], + ) + else: + # Block completion + self.log.warning( + "Cannot complete task - incomplete subtasks exist", + task_id=str(task_id), + incomplete_count=len(incomplete_subtasks), + cancelled_only=cancelled_only, + force_requested=force_with_cancelled, + ) + return None + task.completed_at = datetime.now(UTC) # Validate transition with PM role requirement self._validate_and_set_status(task, TaskStatus.COMPLETED, "cell_pm")