diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 356c8124..080129e5 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -712,10 +712,11 @@ async def block_task( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" ) - # Only assigned agent or PM can block a task + # Only assigned agent, PM, or the CEO can block a task if task.assigned_to != agent.agent_id and agent.role not in ( AgentRole.CELL_PM, AgentRole.MAIN_PM, + AgentRole.CEO, ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, @@ -771,10 +772,11 @@ async def unblock_task( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" ) - # Only assigned agent or PM can unblock a task + # Only assigned agent, PM, or the CEO can unblock a task if task.assigned_to != agent.agent_id and agent.role not in ( AgentRole.CELL_PM, AgentRole.MAIN_PM, + AgentRole.CEO, ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, diff --git a/roboco/services/permissions.py b/roboco/services/permissions.py index 2fd5090d..fa2fc1de 100644 --- a/roboco/services/permissions.py +++ b/roboco/services/permissions.py @@ -299,6 +299,12 @@ class PermissionService(SingletonService): task_team: Team | None = None, ) -> bool: """Check if agent can perform a task action.""" + # The CEO is the ultimate authority and may perform any task action on + # any task — unblock, reassign, cancel, override status, etc. Every + # route that gates a write through this helper therefore lets the CEO + # through (the panel operates as the CEO). + if agent.role == AgentRole.CEO: + return True allowed_actions = TASK_PERMISSIONS.get(agent.role, set()) if action in allowed_actions: diff --git a/tests/unit/services/test_permissions.py b/tests/unit/services/test_permissions.py index 2d0d7d7e..2492144c 100644 --- a/tests/unit/services/test_permissions.py +++ b/tests/unit/services/test_permissions.py @@ -153,6 +153,21 @@ def test_cell_pm_can_close_in_own_cell(svc: PermissionService) -> None: assert svc.can_perform_task_action(cell_pm, TaskAction.CLOSE, Team.BACKEND) is True +def test_ceo_can_perform_any_task_action(svc: PermissionService) -> None: + """The CEO is the ultimate authority — it may perform ANY task action on any + task (assign/reassign, change priority, close, claim, view). The panel + operates as the CEO, so this override is what unblocks the whole UI.""" + ceo = _ctx(AgentRole.CEO) + for action in ( + TaskAction.ASSIGN, + TaskAction.CHANGE_PRIORITY, + TaskAction.CLOSE, + TaskAction.CLAIM, + TaskAction.VIEW_ALL, + ): + assert svc.can_perform_task_action(ceo, action, Team.BACKEND) is True + + def test_get_task_actions_returns_set(svc: PermissionService) -> None: dev = _ctx(AgentRole.DEVELOPER, team=Team.BACKEND) actions = svc.get_task_actions(dev)