diff --git a/roboco/agents_config.py b/roboco/agents_config.py index 9d0f1887..424a4c93 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -218,6 +218,7 @@ CHANNEL_ACCESS: Final[dict[str, dict[str, list[str]]]] = { # This is the SINGLE SOURCE OF TRUTH for role hierarchy # Used by PermissionService to build AgentRole -> PermissionLevel mapping ROLE_PERMISSION_LEVELS: Final[dict[str, str]] = { + "system": "CEO", # System/orchestrator has CEO-level access for internal operations "ceo": "CEO", "product_owner": "BOARD", "head_marketing": "BOARD", diff --git a/roboco/api/app.py b/roboco/api/app.py index e55cd72a..0829d0fb 100644 --- a/roboco/api/app.py +++ b/roboco/api/app.py @@ -12,6 +12,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from roboco.api.middleware import setup_middleware +from roboco.api.routes.agents import router as agents_router from roboco.api.routes.channels import router as channels_router from roboco.api.routes.dashboard import router as dashboard_router from roboco.api.routes.health import router as health_router @@ -146,6 +147,12 @@ def create_app() -> FastAPI: # API v1 api_prefix = "/api/v1" + app.include_router( + agents_router, + prefix=f"{api_prefix}/agents", + tags=["Agents"], + ) + app.include_router( channels_router, prefix=f"{api_prefix}/channels", diff --git a/roboco/api/deps.py b/roboco/api/deps.py index 7f6b2ac3..09a83525 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -6,13 +6,15 @@ Shared dependencies for FastAPI routes. import contextlib from collections.abc import Callable, Coroutine -from typing import Annotated, Any +from typing import Annotated, Any, cast from uuid import UUID from fastapi import Depends, Header, HTTPException, status +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from roboco.db.base import get_db +from roboco.db.tables import AgentTable from roboco.models import AgentRole, Team from roboco.services.permissions import AgentContext, PermissionService @@ -20,6 +22,41 @@ from roboco.services.permissions import AgentContext, PermissionService DbSession = Annotated[AsyncSession, Depends(get_db)] +async def resolve_agent_id(agent_id_str: str, db: AsyncSession) -> UUID: + """ + Resolve agent ID from string (UUID or slug). + + Args: + agent_id_str: Either a UUID string or agent slug (e.g., "be-dev-1") + db: Database session + + Returns: + UUID of the agent + + Raises: + HTTPException: If agent not found or invalid format + """ + # First, try to parse as UUID + try: + return UUID(agent_id_str) + except ValueError: + pass + + # Not a UUID, try to look up by slug + result = await db.execute( + select(AgentTable.id).where(AgentTable.slug == agent_id_str) + ) + agent_uuid = result.scalar_one_or_none() + + if agent_uuid is None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Agent not found: {agent_id_str}", + ) + + return cast("UUID", agent_uuid) + + class _ServiceHolder: """Holder for singleton service instances.""" @@ -37,22 +74,25 @@ PermissionServiceDep = Annotated[PermissionService, Depends(get_permission_servi async def get_current_agent_id( + db: DbSession, x_agent_id: Annotated[str | None, Header()] = None, ) -> UUID: """ Get the current agent ID from request headers. + Accepts either a UUID string or agent slug (e.g., "be-dev-1"). In production, this would validate a JWT token and extract the agent ID. For now, we use a simple header-based approach for development. Args: - x_agent_id: Agent ID from X-Agent-ID header + x_agent_id: Agent ID (UUID or slug) from X-Agent-ID header + db: Database session for slug resolution Returns: UUID of the current agent Raises: - HTTPException: If agent ID is missing or invalid + HTTPException: If agent ID is missing or invalid/not found """ if not x_agent_id: raise HTTPException( @@ -60,13 +100,7 @@ async def get_current_agent_id( detail="Missing X-Agent-ID header", ) - try: - return UUID(x_agent_id) - except ValueError as e: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Invalid agent ID format: {e}", - ) from e + return await resolve_agent_id(x_agent_id, db) # Type alias for current agent dependency @@ -74,19 +108,21 @@ CurrentAgentId = Annotated[UUID, Depends(get_current_agent_id)] async def get_optional_agent_id( + db: DbSession, x_agent_id: Annotated[str | None, Header()] = None, ) -> UUID | None: """ Get the current agent ID if provided. + Accepts either a UUID string or agent slug (e.g., "be-dev-1"). Unlike get_current_agent_id, this doesn't raise an error if missing. """ if not x_agent_id: return None try: - return UUID(x_agent_id) - except ValueError: + return await resolve_agent_id(x_agent_id, db) + except HTTPException: return None @@ -94,6 +130,7 @@ OptionalAgentId = Annotated[UUID | None, Depends(get_optional_agent_id)] async def get_agent_context( + db: DbSession, x_agent_id: Annotated[str | None, Header()] = None, x_agent_role: Annotated[str | None, Header()] = None, x_agent_team: Annotated[str | None, Header()] = None, @@ -105,7 +142,7 @@ async def get_agent_context( For development, we use headers. Required headers: - X-Agent-ID: UUID of the agent + X-Agent-ID: UUID or slug of the agent (e.g., "be-dev-1") X-Agent-Role: Role (e.g., 'developer', 'cell_pm') Optional headers: @@ -123,13 +160,8 @@ async def get_agent_context( detail="Missing X-Agent-Role header", ) - try: - agent_id = UUID(x_agent_id) - except ValueError as e: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Invalid agent ID format: {e}", - ) from e + # Resolve agent ID (UUID or slug) + agent_id = await resolve_agent_id(x_agent_id, db) try: role = AgentRole(x_agent_role.lower()) diff --git a/roboco/api/routes/__init__.py b/roboco/api/routes/__init__.py index fd7b72b3..dd624db5 100644 --- a/roboco/api/routes/__init__.py +++ b/roboco/api/routes/__init__.py @@ -5,6 +5,7 @@ All FastAPI route modules. """ from roboco.api.routes import ( + agents, channels, dashboard, health, @@ -20,6 +21,7 @@ from roboco.api.routes import ( ) __all__ = [ + "agents", "channels", "dashboard", "health", diff --git a/roboco/api/routes/agents.py b/roboco/api/routes/agents.py new file mode 100644 index 00000000..44c11f06 --- /dev/null +++ b/roboco/api/routes/agents.py @@ -0,0 +1,122 @@ +""" +Agent Routes + +Provides agent lookup and information endpoints. +""" + +from typing import cast +from uuid import UUID + +from fastapi import APIRouter, HTTPException, Query, status +from pydantic import BaseModel +from sqlalchemy import select + +from roboco.api.deps import DbSession +from roboco.db.tables import AgentTable +from roboco.models import AgentRole, Team + +router = APIRouter() + + +class AgentResponse(BaseModel): + """Response model for agent information.""" + + id: UUID + name: str + slug: str + role: AgentRole + team: Team | None + + class Config: + """Pydantic config.""" + + from_attributes = True + + +@router.get("") +async def list_agents( + db: DbSession, + slug: str | None = Query(None, description="Filter by agent slug"), + role: str | None = Query(None, description="Filter by role"), + team: str | None = Query(None, description="Filter by team"), +) -> list[AgentResponse]: + """ + List agents with optional filters. + + Supports filtering by slug, role, or team. + """ + query = select(AgentTable) + + if slug: + query = query.where(AgentTable.slug == slug) + if role: + try: + role_enum = AgentRole(role.lower()) + query = query.where(AgentTable.role == role_enum) + except ValueError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid role: {role}", + ) from None + if team: + try: + team_enum = Team(team.lower()) + query = query.where(AgentTable.team == team_enum) + except ValueError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid team: {team}", + ) from None + + result = await db.execute(query) + agents = result.scalars().all() + + return [ + AgentResponse( + id=cast("UUID", agent.id), + name=agent.name, + slug=agent.slug, + role=agent.role, + team=agent.team, + ) + for agent in agents + ] + + +@router.get("/{agent_id}") +async def get_agent( + agent_id: str, + db: DbSession, +) -> AgentResponse: + """ + Get agent by ID (UUID or slug). + + Accepts either a UUID string or agent slug (e.g., "be-dev-1"). + """ + # Try to parse as UUID first + try: + uuid = UUID(agent_id) + result = await db.execute( + select(AgentTable).where(AgentTable.id == uuid) + ) + except ValueError: + # Not a UUID, try slug lookup + result = await db.execute( + select(AgentTable).where(AgentTable.slug == agent_id) + ) + + agent = result.scalar_one_or_none() + + if agent is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Agent not found: {agent_id}", + ) + + return AgentResponse( + id=cast("UUID", agent.id), + name=agent.name, + slug=agent.slug, + role=agent.role, + team=agent.team, + ) diff --git a/roboco/api/routes/sessions.py b/roboco/api/routes/sessions.py index 508bac92..f0552aed 100644 --- a/roboco/api/routes/sessions.py +++ b/roboco/api/routes/sessions.py @@ -22,6 +22,7 @@ from roboco.api.schemas.sessions import ( ) from roboco.db.tables import GroupTable, SessionTable from roboco.models import SessionStatus +from roboco.services.permissions import has_privileged_access from roboco.utils.converters import require_uuid router = APIRouter() @@ -58,9 +59,14 @@ async def list_sessions( detail="Group not found", ) - # Check channel access + # Check channel access (privileged roles bypass membership check) channel = group.channel - if agent_id not in channel.members and agent_id not in channel.silent_observers: + has_access = ( + agent_id in channel.members + or agent_id in channel.silent_observers + or await has_privileged_access(db, agent_id) + ) + if not has_access: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You don't have access to this group", @@ -163,9 +169,12 @@ async def create_session( detail="Group not found", ) - # Check write access to channel + # Check write access to channel (privileged roles bypass membership check) channel = group.channel - if agent_id not in channel.writers: + has_write_access = agent_id in channel.writers or await has_privileged_access( + db, agent_id + ) + if not has_write_access: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You don't have write access to this group", diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 3660a4e6..2f758200 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -7,7 +7,8 @@ Full CRUD operations and lifecycle management for tasks. from typing import Annotated from uuid import UUID -from fastapi import APIRouter, HTTPException, Query, status +from fastapi import APIRouter, Body, HTTPException, Query, status +from sqlalchemy import select from roboco.api.deps import ( CurrentAgentContext, @@ -17,6 +18,7 @@ from roboco.api.deps import ( from roboco.api.schemas.tasks import ( CheckpointRequest, CheckpointResponse, + ClaimRequest, CommitRefResponse, CommitRequest, ListTasksQuery, @@ -30,7 +32,7 @@ from roboco.api.schemas.tasks import ( TaskUpdate, TeamTasksQuery, ) -from roboco.db.tables import TaskTable +from roboco.db.tables import AgentTable, TaskTable from roboco.models.base import TaskStatus, Team from roboco.models.task import TaskCreate from roboco.services.audit import get_audit_service @@ -421,6 +423,7 @@ async def get_task( @router.put("/{task_id}", response_model=TaskResponse) +@router.patch("/{task_id}", response_model=TaskResponse) async def update_task( task_id: UUID, data: TaskUpdate, @@ -428,7 +431,7 @@ async def update_task( agent: CurrentAgentContext, permissions: PermissionServiceDep, ) -> TaskResponse: - """Update a task.""" + """Update a task. Supports both PUT and PATCH for partial updates.""" service = get_task_service(db) task = await service.get(task_id) if not task: @@ -515,8 +518,14 @@ async def claim_task( db: DbSession, agent: CurrentAgentContext, permissions: PermissionServiceDep, + data: Annotated[ClaimRequest | None, Body()] = None, ) -> TaskResponse: - """Claim a task.""" + """ + Claim a task. + + Privileged roles (system, PM) can claim tasks on behalf of other agents + by providing agent_id in the request body. + """ service = get_task_service(db) task = await service.get(task_id) if not task: @@ -531,7 +540,33 @@ async def claim_task( detail="Not authorized to claim tasks", ) - task = await service.claim(task_id, agent.agent_id) + # Determine the agent to claim for + # Privileged roles can claim on behalf of other agents + can_assign = permissions.can_perform_task_action( + agent, TaskAction.ASSIGN, task.team + ) + if data and data.agent_id and can_assign: + # Resolve agent_id from UUID string or slug + agent_id_str = data.agent_id + try: + # Try parsing as UUID first + claim_agent_id = UUID(agent_id_str) + except ValueError: + # Not a UUID, look up by slug + result = await db.execute( + select(AgentTable.id).where(AgentTable.slug == agent_id_str) + ) + agent_uuid = result.scalar_one_or_none() + if not agent_uuid: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Agent not found: {agent_id_str}", + ) from None + claim_agent_id = agent_uuid + else: + claim_agent_id = agent.agent_id + + task = await service.claim(task_id, claim_agent_id) if not task: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, diff --git a/roboco/api/schemas/__init__.py b/roboco/api/schemas/__init__.py index 8a9ca8b5..fcf12b8c 100644 --- a/roboco/api/schemas/__init__.py +++ b/roboco/api/schemas/__init__.py @@ -91,6 +91,7 @@ from roboco.api.schemas.stream import ( ) from roboco.api.schemas.tasks import ( CheckpointRequest, + ClaimRequest, CommitRequest, ListTasksQuery, ProgressRequest, @@ -116,6 +117,7 @@ __all__ = [ "ChannelResponse", # Tasks "CheckpointRequest", + "ClaimRequest", # Optimal "ClearIndexResponse", "CommitRequest", diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index b6d77d61..308d901d 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -186,6 +186,16 @@ class CommitRequest(BaseModel): message: str +class ClaimRequest(BaseModel): + """Request to claim a task on behalf of an agent. + + Used by privileged roles (system, PM) to claim tasks for other agents. + Accepts either a UUID or agent slug (e.g., "be-dev-1"). + """ + + agent_id: str = Field(..., description="The agent ID (UUID) or slug to claim for") + + class QANotes(BaseModel): """QA review notes.""" diff --git a/roboco/db/seed.py b/roboco/db/seed.py index ff7c0cbc..f552fb25 100644 --- a/roboco/db/seed.py +++ b/roboco/db/seed.py @@ -5,6 +5,7 @@ Functions to populate the database with initial data. Separates database operations from bootstrap orchestration. """ +import contextlib from uuid import UUID as UUIDType import structlog @@ -100,8 +101,15 @@ async def create_agents(session: AsyncSession) -> dict[str, str]: team_str = agent_data.get("team") team = Team(team_str) if team_str else None + # If slug is a valid UUID, use it as the database ID + # (important for CEO so X-Agent-ID header matches the DB id) + explicit_id: UUIDType | None = None + with contextlib.suppress(ValueError): + explicit_id = UUIDType(slug) + # Create agent using ORM agent = AgentTable( + id=explicit_id, # Will use slug as ID if it's a valid UUID name=agent_data["name"], slug=slug, role=role, diff --git a/roboco/llm/toon_adapter.py b/roboco/llm/toon_adapter.py index e3af9557..99e5922e 100644 --- a/roboco/llm/toon_adapter.py +++ b/roboco/llm/toon_adapter.py @@ -62,7 +62,9 @@ class ToonAdapter: if isinstance(data, BaseModel): data = data.model_dump() - result: str = toon.encode(data, indent=self.config.indent) + # Note: toon.encode() does not support indent parameter + # TOON format is already compact by design + result: str = toon.encode(data) return result def decode(self, toon_str: str) -> dict[str, Any] | list[Any]: diff --git a/roboco/mcp/journal_server.py b/roboco/mcp/journal_server.py index 4b6d3031..449247bd 100644 --- a/roboco/mcp/journal_server.py +++ b/roboco/mcp/journal_server.py @@ -20,6 +20,7 @@ import httpx from fastapi import status from mcp.server.fastmcp import FastMCP +from roboco.agents_config import get_agent_role from roboco.config import settings from roboco.llm import ToonAdapter from roboco.mcp.schemas import ( @@ -54,6 +55,14 @@ def _format_error_response( } +def _get_agent_headers(agent_id: str) -> dict[str, str]: + """Get standard headers for API calls.""" + return { + "X-Agent-Id": agent_id, + "X-Agent-Role": get_agent_role(agent_id), + } + + async def _post_journal_entry( endpoint: str, payload: dict[str, Any], @@ -64,7 +73,7 @@ async def _post_journal_entry( resp = await client.post( f"{settings.internal_api_url}/journals/me/{endpoint}", json=payload, - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code not in [200, 201]: return None, _format_error_response( @@ -221,7 +230,7 @@ async def _handle_search(query: str, top_k: int, agent_id: str) -> dict[str, Any resp = await client.post( f"{settings.internal_api_url}/journals/me/search", json=payload, - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code != status.HTTP_200_OK: @@ -249,11 +258,11 @@ async def _handle_stats(agent_id: str) -> dict[str, Any]: async with httpx.AsyncClient() as client: stats_resp = await client.get( f"{settings.internal_api_url}/journals/me/stats", - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) growth_resp = await client.get( f"{settings.internal_api_url}/journals/me/growth", - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) stats = ( @@ -299,7 +308,7 @@ async def _handle_recent( resp = await client.get( f"{settings.internal_api_url}/journals/me/entries", params=params, - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code != status.HTTP_200_OK: diff --git a/roboco/mcp/message_server.py b/roboco/mcp/message_server.py index e0fa1aeb..a79d80d8 100644 --- a/roboco/mcp/message_server.py +++ b/roboco/mcp/message_server.py @@ -20,7 +20,7 @@ import httpx from fastapi import status from mcp.server.fastmcp import FastMCP -from roboco.agents_config import CHANNEL_ACCESS +from roboco.agents_config import CHANNEL_ACCESS, get_agent_role from roboco.config import settings from roboco.llm import ToonAdapter from roboco.mcp.schemas import ( @@ -33,6 +33,14 @@ from roboco.mcp.schemas import ( _toon = ToonAdapter() +def _get_agent_headers(agent_id: str) -> dict[str, str]: + """Get standard headers for API calls.""" + return { + "X-Agent-Id": agent_id, + "X-Agent-Role": get_agent_role(agent_id), + } + + # ============================================================================= # HELPER FUNCTIONS # ============================================================================= @@ -112,26 +120,76 @@ def _validate_message_send( return None +async def _get_default_group( + client: httpx.AsyncClient, + channel_id: str, + headers: dict[str, str], +) -> str | dict[str, Any]: + """Get the default (first) group for a channel. Returns group_id or error dict.""" + groups_resp = await client.get( + f"{settings.internal_api_url}/channels/{channel_id}/groups", + headers=headers, + ) + + if groups_resp.status_code != status.HTTP_200_OK: + return _format_error_response( + "GROUPS_ERROR", + "Failed to get channel groups", + {"status": groups_resp.status_code}, + ) + + groups = groups_resp.json() + if not groups: + return _format_error_response("NO_GROUPS", "Channel has no groups") + + # Return first active group, or first group if none are active + for group in groups: + if group.get("is_active", True): + return str(group["id"]) + return str(groups[0]["id"]) + + async def _get_or_create_session( client: httpx.AsyncClient, channel_id: str, + headers: dict[str, str], ) -> str | dict[str, Any]: """Get or create session for channel. Returns session_id or error dict.""" - session_resp = await client.get( - f"{settings.internal_api_url}/channels/{channel_id}/session" + # First get the default group for this channel + group_result = await _get_default_group(client, channel_id, headers) + if isinstance(group_result, dict): + return group_result # Error response + group_id = group_result + + # Check if group has an active session + sessions_resp = await client.get( + f"{settings.internal_api_url}/sessions", + params={"group_id": group_id, "limit": 1}, + headers=headers, ) - if session_resp.status_code == status.HTTP_200_OK: - return str(session_resp.json()["id"]) + if sessions_resp.status_code == status.HTTP_200_OK: + data = sessions_resp.json() + items = data.get("items", []) + # Find an active session + for session in items: + if session.get("status") == "active": + return str(session["id"]) + # Create new session create_resp = await client.post( f"{settings.internal_api_url}/sessions", - json={"channel_id": channel_id}, + json={"group_id": group_id}, + headers=headers, ) if create_resp.status_code in [status.HTTP_200_OK, status.HTTP_201_CREATED]: return str(create_resp.json()["id"]) - return _format_error_response("SESSION_ERROR", "Failed to get or create session") + return _format_error_response( + "SESSION_ERROR", + "Failed to create session", + {"api_error": create_resp.text}, + ) # ============================================================================= @@ -164,7 +222,7 @@ async def _handle_channel_list(agent_id: str) -> dict[str, Any]: } -async def _handle_channel_history( +async def _handle_channel_history( # noqa: PLR0911 agent_id: str, channel_slug: str, limit: int, @@ -179,38 +237,86 @@ async def _handle_channel_history( limit = min(limit, 100) since = datetime.now(UTC) - timedelta(hours=hours_back) + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: + # Get channel by slug channels_resp = await client.get( f"{settings.internal_api_url}/channels", params={"slug": channel_slug}, + headers=headers, ) if channels_resp.status_code != status.HTTP_200_OK: return _format_error_response("API_ERROR", "Failed to fetch channels") - channels = channels_resp.json() - if not channels: + channels_data = channels_resp.json() + items = channels_data.get("items", channels_data) + if not items: return _format_error_response( "NOT_FOUND", f"Channel #{channel_slug} not found" ) - channel_id = channels[0]["id"] + channel = items[0] if isinstance(items, list) else items + channel_id = channel["id"] - messages_resp = await client.get( - f"{settings.internal_api_url}/channels/{channel_id}/messages", - params={"after": since.isoformat(), "limit": limit}, + # Get groups for this channel + group_result = await _get_default_group(client, channel_id, headers) + if isinstance(group_result, dict): + return group_result # Error response + group_id = group_result + + # Get sessions for this group + sessions_resp = await client.get( + f"{settings.internal_api_url}/sessions", + params={"group_id": group_id, "limit": 5}, + headers=headers, ) - if messages_resp.status_code != status.HTTP_200_OK: - return _format_error_response("API_ERROR", "Failed to fetch messages") + if sessions_resp.status_code != status.HTTP_200_OK: + return _format_error_response("API_ERROR", "Failed to fetch sessions") - messages = messages_resp.json() + sessions_data = sessions_resp.json() + sessions = sessions_data.get("items", []) + + if not sessions: + return { + "channel": channel_slug, + "messages": [], + "total": 0, + "has_more": False, + "since": since.isoformat(), + } + + # Get messages from all recent sessions + all_messages = [] + for session in sessions: + session_id = session["id"] + messages_resp = await client.get( + f"{settings.internal_api_url}/messages", + params={ + "session_id": session_id, + "after": since.isoformat(), + "limit": limit, + }, + headers=headers, + ) + + if messages_resp.status_code == status.HTTP_200_OK: + msg_data = messages_resp.json() + all_messages.extend(msg_data.get("items", [])) + + if len(all_messages) >= limit: + break + + # Sort by timestamp descending and limit + all_messages.sort(key=lambda m: m.get("timestamp", ""), reverse=True) + all_messages = all_messages[:limit] return { "channel": channel_slug, - "messages": messages.get("items", []), - "total": messages.get("total", 0), - "has_more": messages.get("has_more", False), + "messages": all_messages, + "total": len(all_messages), + "has_more": len(all_messages) >= limit, "since": since.isoformat(), } @@ -225,10 +331,12 @@ async def _handle_message_send( ): return validation_error + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: channels_resp = await client.get( f"{settings.internal_api_url}/channels", params={"slug": data.channel_slug}, + headers=headers, ) if channels_resp.status_code != status.HTTP_200_OK or not channels_resp.json(): @@ -239,7 +347,7 @@ async def _handle_message_send( channel = channels_resp.json()[0] channel_id = channel["id"] - session_result = await _get_or_create_session(client, channel_id) + session_result = await _get_or_create_session(client, channel_id, headers) if isinstance(session_result, dict): return session_result session_id = session_result @@ -257,7 +365,7 @@ async def _handle_message_send( send_resp = await client.post( f"{settings.internal_api_url}/messages", json=message_data, - headers={"X-Agent-Id": agent_id}, + headers=headers, ) if send_resp.status_code not in [status.HTTP_200_OK, status.HTTP_201_CREATED]: @@ -273,10 +381,14 @@ async def _handle_message_send( } -async def _handle_message_get(message_id: str) -> dict[str, Any]: +async def _handle_message_get(message_id: str, agent_id: str) -> dict[str, Any]: """Handle message retrieval.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - resp = await client.get(f"{settings.internal_api_url}/messages/{message_id}") + resp = await client.get( + f"{settings.internal_api_url}/messages/{message_id}", + headers=headers, + ) if resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response( @@ -396,7 +508,7 @@ def create_message_mcp_server(agent_id: str) -> FastMCP: @mcp.tool() async def roboco_message_get(message_id: str) -> dict[str, Any]: """Get a specific message by ID.""" - return await _handle_message_get(message_id) + return await _handle_message_get(message_id, agent_id) @mcp.tool() async def roboco_ask_question( diff --git a/roboco/mcp/notify_server.py b/roboco/mcp/notify_server.py index dc8621c4..797443ac 100644 --- a/roboco/mcp/notify_server.py +++ b/roboco/mcp/notify_server.py @@ -30,6 +30,14 @@ from roboco.mcp.schemas import SendNotificationInput # ============================================================================= +def _get_agent_headers(agent_id: str) -> dict[str, str]: + """Get standard headers for API calls.""" + return { + "X-Agent-Id": agent_id, + "X-Agent-Role": get_agent_role(agent_id), + } + + def _check_cell_scope(sender_id: str) -> tuple[bool, str]: """Check if sender can notify within their cell.""" sender_cell = get_agent_cell(sender_id) @@ -126,7 +134,7 @@ async def _handle_list( resp = await client.get( f"{settings.internal_api_url}/notifications", params=params, - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code != status.HTTP_200_OK: @@ -162,7 +170,7 @@ async def _handle_get(agent_id: str, notification_id: str) -> dict[str, Any]: async with httpx.AsyncClient() as client: resp = await client.get( f"{settings.internal_api_url}/notifications/{notification_id}", - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code == status.HTTP_404_NOT_FOUND: @@ -193,7 +201,7 @@ async def _handle_ack(agent_id: str, notification_id: str) -> dict[str, Any]: async with httpx.AsyncClient() as client: resp = await client.post( f"{settings.internal_api_url}/notifications/{notification_id}/ack", - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code == status.HTTP_404_NOT_FOUND: @@ -280,7 +288,7 @@ async def _handle_send(agent_id: str, data: SendNotificationInput) -> dict[str, resp = await client.post( f"{settings.internal_api_url}/notifications", json=payload, - headers={"X-Agent-Id": agent_id}, + headers=_get_agent_headers(agent_id), ) if resp.status_code not in [status.HTTP_200_OK, status.HTTP_201_CREATED]: diff --git a/roboco/mcp/task_server.py b/roboco/mcp/task_server.py index bd07365f..ffdae7c8 100644 --- a/roboco/mcp/task_server.py +++ b/roboco/mcp/task_server.py @@ -26,9 +26,55 @@ import httpx from fastapi import status from mcp.server.fastmcp import FastMCP +from roboco.agents_config import get_agent_role from roboco.config import settings from roboco.llm import ToonAdapter + +def _get_agent_headers(agent_id: str) -> dict[str, str]: + """Get standard headers for API calls.""" + return { + "X-Agent-Id": agent_id, + "X-Agent-Role": get_agent_role(agent_id), + } + + +# Cache for agent slug -> UUID resolution +_agent_uuid_cache: dict[str, str] = {} + + +async def _resolve_agent_uuid(agent_id: str, headers: dict[str, str]) -> str | None: + """Resolve agent slug to UUID. Returns None if not found.""" + # Check if already a valid UUID + try: + from uuid import UUID + + UUID(agent_id) + return agent_id # Already a UUID + except ValueError: + pass + + # Check cache + if agent_id in _agent_uuid_cache: + return _agent_uuid_cache[agent_id] + + # Query API to resolve slug to UUID + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{settings.internal_api_url}/agents", + params={"slug": agent_id}, + headers=headers, + ) + if resp.status_code == status.HTTP_200_OK: + agents = resp.json() + if agents: + uuid_str = str(agents[0]["id"]) + _agent_uuid_cache[agent_id] = uuid_str + return uuid_str + + return None + + # Global TOON adapter for encoding task data _toon = ToonAdapter() @@ -145,11 +191,13 @@ def _get_next_step_guidance(status: str) -> tuple[str, str]: async def _handle_task_scan(team: str | None, agent_id: str) -> dict[str, Any]: """Handle task scanning.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: # Get paused tasks for this agent paused_resp = await client.get( f"{settings.internal_api_url}/tasks", params={"assigned_to": agent_id, "status": "paused"}, + headers=headers, ) paused_tasks = ( paused_resp.json() if paused_resp.status_code == status.HTTP_200_OK else [] @@ -159,6 +207,7 @@ async def _handle_task_scan(team: str | None, agent_id: str) -> dict[str, Any]: assigned_resp = await client.get( f"{settings.internal_api_url}/tasks", params={"assigned_to": agent_id}, + headers=headers, ) assigned_data = ( assigned_resp.json() @@ -179,6 +228,7 @@ async def _handle_task_scan(team: str | None, agent_id: str) -> dict[str, Any]: available_resp = await client.get( f"{settings.internal_api_url}/tasks", params=params, + headers=headers, ) available_tasks = ( available_resp.json() @@ -216,10 +266,14 @@ async def _handle_task_scan(team: str | None, agent_id: str) -> dict[str, Any]: } -async def _handle_task_get(task_id: str) -> dict[str, Any]: +async def _handle_task_get(task_id: str, agent_id: str) -> dict[str, Any]: """Handle getting task details.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) if resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response( @@ -272,10 +326,14 @@ def _validate_task_claimable(task: dict) -> dict[str, Any] | None: return None -async def _get_project_context(project_id: str) -> dict[str, Any] | None: +async def _get_project_context(project_id: str, agent_id: str) -> dict[str, Any] | None: """Fetch project context if available.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - resp = await client.get(f"{settings.internal_api_url}/projects/{project_id}") + resp = await client.get( + f"{settings.internal_api_url}/projects/{project_id}", + headers=headers, + ) if resp.status_code == status.HTTP_200_OK: result: dict[str, Any] = resp.json() return result @@ -284,10 +342,12 @@ async def _get_project_context(project_id: str) -> dict[str, Any] | None: async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]: """Handle task claiming.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: active_resp = await client.get( f"{settings.internal_api_url}/tasks", params={"assigned_to": agent_id}, + headers=headers, ) if active_resp.status_code == status.HTTP_200_OK: active_tasks = active_resp.json() @@ -296,7 +356,10 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]: if error := _check_paused_tasks(active_tasks): return error - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) if task_resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response("NOT_FOUND", f"Task {task_id} not found") @@ -307,6 +370,7 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]: claim_resp = await client.post( f"{settings.internal_api_url}/tasks/{task_id}/claim", json={"agent_id": agent_id}, + headers=headers, ) if claim_resp.status_code != status.HTTP_200_OK: return _format_error_response( @@ -319,7 +383,7 @@ async def _handle_task_claim(task_id: str, agent_id: str) -> dict[str, Any]: project = None if claimed_task.get("project_id"): - project = await _get_project_context(claimed_task["project_id"]) + project = await _get_project_context(claimed_task["project_id"], agent_id) return _format_task_response( claimed_task, @@ -332,13 +396,30 @@ 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: +async def _validate_task_ownership( + task: dict, agent_id: str, headers: dict[str, str] +) -> dict[str, Any] | None: """Validate agent owns the task. Returns error or None.""" - if task.get("assigned_to") != agent_id: + assigned_to = task.get("assigned_to") + if not assigned_to: + return _format_error_response( + "NOT_ASSIGNED", + "This task is not assigned to anyone", + ) + + # Resolve agent_id (which may be a slug) to UUID for comparison + agent_uuid = await _resolve_agent_uuid(agent_id, headers) + if not agent_uuid: + return _format_error_response( + "AGENT_NOT_FOUND", + f"Could not resolve agent: {agent_id}", + ) + + if str(assigned_to) != agent_uuid: return _format_error_response( "NOT_OWNER", "You are not assigned to this task", - {"assigned_to": task.get("assigned_to")}, + {"assigned_to": assigned_to}, ) return None @@ -381,13 +462,17 @@ async def _handle_task_plan( agent_id: str, ) -> dict[str, Any]: """Handle task planning.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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): + if error := await _validate_task_ownership(task, agent_id, headers): return error if error := _validate_task_status_claimed(task): return error @@ -396,6 +481,7 @@ async def _handle_task_plan( update_resp = await client.patch( f"{settings.internal_api_url}/tasks/{task_id}", json={"plan": plan_data}, + headers=headers, ) if update_resp.status_code != status.HTTP_200_OK: return _format_error_response( @@ -423,10 +509,12 @@ async def _handle_task_plan( ) -def _validate_task_start(task: dict[str, Any], agent_id: str) -> dict[str, Any] | None: +async def _validate_task_start( + task: dict[str, Any], agent_id: str, headers: dict[str, str] +) -> dict[str, Any] | None: """Validate task can be started. Returns error dict or None.""" - if task.get("assigned_to") != agent_id: - return _format_error_response("NOT_OWNER", "You are not assigned to this task") + if error := await _validate_task_ownership(task, agent_id, headers): + return error task_status = task.get("status") if task_status not in ["claimed", "paused"]: @@ -459,19 +547,24 @@ def _validate_task_start(task: dict[str, Any], agent_id: str) -> dict[str, Any] async def _handle_task_start(task_id: str, agent_id: str) -> dict[str, Any]: """Handle task start.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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 validation_error := _validate_task_start(task, agent_id): + if validation_error := await _validate_task_start(task, agent_id, headers): return validation_error # Start the task start_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/start" + f"{settings.internal_api_url}/tasks/{task_id}/start", + headers=headers, ) if start_resp.status_code != status.HTTP_200_OK: @@ -500,17 +593,19 @@ async def _handle_task_progress( agent_id: str, ) -> dict[str, Any]: """Handle task progress update.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "in_progress": return _format_error_response( @@ -526,6 +621,7 @@ async def _handle_task_progress( "message": message, "percentage": percentage, }, + headers=headers, ) if progress_resp.status_code != status.HTTP_200_OK: @@ -557,17 +653,19 @@ async def _handle_task_block( "Both 'reason' and 'what_needed' are required to block a task.", ) + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "in_progress": return _format_error_response( @@ -583,6 +681,7 @@ async def _handle_task_block( "blocker_type": blocker_type, "what_needed": what_needed, }, + headers=headers, ) if block_resp.status_code != status.HTTP_200_OK: @@ -605,17 +704,19 @@ async def _handle_task_block( async def _handle_task_unblock(task_id: str, agent_id: str) -> dict[str, Any]: """Handle task unblocking.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "blocked": return _format_error_response( @@ -624,7 +725,8 @@ async def _handle_task_unblock(task_id: str, agent_id: str) -> dict[str, Any]: ) unblock_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/unblock" + f"{settings.internal_api_url}/tasks/{task_id}/unblock", + headers=headers, ) if unblock_resp.status_code != status.HTTP_200_OK: @@ -647,17 +749,19 @@ async def _handle_task_pause( agent_id: str, ) -> dict[str, Any]: """Handle task pausing.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "in_progress": return _format_error_response( @@ -674,11 +778,13 @@ async def _handle_task_pause( "remaining_work": remaining_work, "notes": reason, }, + headers=headers, ) # Pause the task pause_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/pause" + f"{settings.internal_api_url}/tasks/{task_id}/pause", + headers=headers, ) if pause_resp.status_code != status.HTTP_200_OK: @@ -700,17 +806,19 @@ async def _handle_task_submit_verification( task_id: str, agent_id: str ) -> dict[str, Any]: """Handle task verification submission.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "in_progress": return _format_error_response( @@ -728,7 +836,8 @@ async def _handle_task_submit_verification( ) verify_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/verify" + f"{settings.internal_api_url}/tasks/{task_id}/verify", + headers=headers, ) if verify_resp.status_code != status.HTTP_200_OK: @@ -765,17 +874,19 @@ async def _handle_task_submit_qa( "Both dev_notes and handoff_summary are required for QA submission.", ) + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) 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" - ) + if error := await _validate_task_ownership(task, agent_id, headers): + return error if task.get("status") != "verifying": return _format_error_response( @@ -790,11 +901,13 @@ async def _handle_task_submit_qa( "dev_notes": dev_notes, "documenter_handoff": handoff_summary, }, + headers=headers, ) # Submit for QA qa_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/submit-qa" + f"{settings.internal_api_url}/tasks/{task_id}/submit-qa", + headers=headers, ) if qa_resp.status_code != status.HTTP_200_OK: @@ -824,8 +937,12 @@ async def _handle_task_qa_pass( "Only QA agents can pass tasks through QA review.", ) + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) if task_resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response("NOT_FOUND", f"Task {task_id} not found") @@ -847,6 +964,7 @@ async def _handle_task_qa_pass( pass_resp = await client.post( f"{settings.internal_api_url}/tasks/{task_id}/pass-qa", json={"notes": qa_notes}, + headers=headers, ) if pass_resp.status_code != status.HTTP_200_OK: @@ -881,8 +999,12 @@ async def _handle_task_qa_fail( "Must specify at least one issue when failing QA.", ) + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) if task_resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response("NOT_FOUND", f"Task {task_id} not found") @@ -899,6 +1021,7 @@ async def _handle_task_qa_fail( fail_resp = await client.post( f"{settings.internal_api_url}/tasks/{task_id}/fail-qa", json={"notes": full_notes}, + headers=headers, ) if fail_resp.status_code != status.HTTP_200_OK: @@ -915,10 +1038,14 @@ async def _handle_task_qa_fail( ) -async def _handle_task_complete(task_id: str) -> dict[str, Any]: +async def _handle_task_complete(task_id: str, agent_id: str) -> dict[str, Any]: """Handle task completion.""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: - task_resp = await client.get(f"{settings.internal_api_url}/tasks/{task_id}") + task_resp = await client.get( + f"{settings.internal_api_url}/tasks/{task_id}", + headers=headers, + ) if task_resp.status_code == status.HTTP_404_NOT_FOUND: return _format_error_response("NOT_FOUND", f"Task {task_id} not found") @@ -931,7 +1058,8 @@ async def _handle_task_complete(task_id: str) -> dict[str, Any]: ) complete_resp = await client.post( - f"{settings.internal_api_url}/tasks/{task_id}/complete" + f"{settings.internal_api_url}/tasks/{task_id}/complete", + headers=headers, ) if complete_resp.status_code != status.HTTP_200_OK: @@ -948,11 +1076,13 @@ async def _handle_task_complete(task_id: str) -> dict[str, Any]: async def _handle_agent_idle(agent_id: str) -> dict[str, Any]: """Handle agent going idle (no work available).""" + headers = _get_agent_headers(agent_id) async with httpx.AsyncClient() as client: # Signal to orchestrator that this agent is idle resp = await client.post( f"{settings.internal_api_url}/orchestrator/agents/{agent_id}/mark-waiting", params={"waiting_for": "task_assignment"}, + headers=headers, ) if resp.status_code == status.HTTP_204_NO_CONTENT: @@ -1022,7 +1152,7 @@ def create_task_mcp_server(agent_id: str) -> FastMCP: Returns: Task details with current status and guidance """ - return await _handle_task_get(task_id) + return await _handle_task_get(task_id, agent_id) @mcp.tool() async def roboco_task_claim(task_id: str) -> dict[str, Any]: @@ -1288,7 +1418,7 @@ def create_task_mcp_server(agent_id: str) -> FastMCP: Returns: Completed task """ - return await _handle_task_complete(task_id) + return await _handle_task_complete(task_id, agent_id) @mcp.tool() async def roboco_agent_idle() -> dict[str, Any]: diff --git a/roboco/models/base.py b/roboco/models/base.py index a25712ca..01d42d41 100644 --- a/roboco/models/base.py +++ b/roboco/models/base.py @@ -170,8 +170,6 @@ class RobocoBase(BaseModel): validate_assignment=True, # Allow population by field name populate_by_name=True, - # Strict mode for better type checking - strict=True, # Extra fields are forbidden extra="forbid", ) diff --git a/roboco/models/permissions.py b/roboco/models/permissions.py index 8c32af20..55b8e43f 100644 --- a/roboco/models/permissions.py +++ b/roboco/models/permissions.py @@ -134,6 +134,16 @@ COMMUNICATION_MATRIX: dict[AgentRole, set[AgentRole]] = { # Per HOMELAB_TEAM_V0.md Section 12.3 TASK_PERMISSIONS: dict[AgentRole, set[str]] = { + # System role (orchestrator) has full access for internal operations + AgentRole.SYSTEM: { + TaskAction.VIEW_ALL, + TaskAction.CREATE, + TaskAction.ASSIGN, + TaskAction.CLAIM, + TaskAction.UPDATE_OWN, + TaskAction.CLOSE, + TaskAction.CHANGE_PRIORITY, + }, AgentRole.CEO: { TaskAction.VIEW_ALL, TaskAction.CREATE, diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 1f6d0604..b290f4bf 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -403,14 +403,20 @@ class AgentOrchestrator: "stream-json", "--verbose", # Always provide a prompt (required for non-interactive mode) - # NOTE: With the smart dispatcher, agents should ALWAYS receive - # a task assignment at spawn time. This default indicates a bug. + # If no task assignment provided, agent should follow standard workflow: + # SCAN for work -> CLAIM if available -> or IDLE if no work "-p", initial_prompt or ( - "ERROR: You were spawned without a task assignment. " - "This is a bug in the orchestrator. " - "Call roboco_agent_idle() immediately to shutdown." + "You may have been spawned without a specific task assignment. " + "Follow your standard workflow:\n\n" + "1. Call `roboco_task_scan()` to find available work for your role\n" + "2. If tasks are found, claim one with `roboco_task_claim(task_id)` " + "and begin the full task lifecycle " + "(UNDERSTAND -> PLAN -> EXECUTE -> VERIFY -> HANDOFF)\n" + "3. If no tasks are available, call `roboco_agent_idle()` " + "to shutdown gracefully\n\n" + "Start now by scanning for work." ), ] @@ -525,7 +531,7 @@ class AgentOrchestrator: cell_dir = "backend" elif team == "frontend": cell_dir = "frontend" - elif team == "uxui": + elif team == "ux_ui": cell_dir = "ux_ui" else: cell_dir = "board" @@ -542,7 +548,7 @@ class AgentOrchestrator: cell_dir = "backend" elif team == "frontend": cell_dir = "frontend" - elif team == "uxui": + elif team == "ux_ui": cell_dir = "ux_ui" else: cell_dir = "board" @@ -581,7 +587,7 @@ class AgentOrchestrator: if agent_id.startswith("fe-"): return "frontend" if agent_id.startswith("ux-"): - return "uxui" + return "ux_ui" return None # ========================================================================= @@ -896,7 +902,7 @@ Start by: Prefers agents that are not currently active. For developers, uses round-robin among candidates. """ - prefix_map = {"backend": "be", "frontend": "fe", "uxui": "ux"} + prefix_map = {"backend": "be", "frontend": "fe", "ux_ui": "ux"} prefix = prefix_map.get(cell) if not prefix: return None @@ -1074,7 +1080,7 @@ Start by: continue team = task.get("team") - if team not in ["backend", "frontend", "uxui"]: + if team not in ["backend", "frontend", "ux_ui"]: continue # Select best agent for this task @@ -1105,7 +1111,7 @@ Start by: for task in tasks: team = task.get("team") - if team not in ["backend", "frontend", "uxui"]: + if team not in ["backend", "frontend", "ux_ui"]: continue agent_id = self._select_agent_for_cell(team, "qa") @@ -1136,7 +1142,7 @@ Start by: for task in tasks: team = task.get("team") - if team not in ["backend", "frontend", "uxui"]: + if team not in ["backend", "frontend", "ux_ui"]: continue agent_id = self._select_agent_for_cell(team, "doc") @@ -1193,7 +1199,7 @@ Start by: for task in tasks: team = task.get("team") - if team not in ["backend", "frontend", "uxui"]: + if team not in ["backend", "frontend", "ux_ui"]: continue agent_id = self._select_agent_for_cell(team, "pm") diff --git a/roboco/services/permissions.py b/roboco/services/permissions.py index ddb45411..91a3b23c 100644 --- a/roboco/services/permissions.py +++ b/roboco/services/permissions.py @@ -21,9 +21,14 @@ Architecture: - No duplicate permission definitions - all derived from agents_config """ -from typing import Any +from typing import TYPE_CHECKING, Any +from uuid import UUID import structlog +from sqlalchemy import select + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession from roboco.agents_config import ( AGENT_ROLE_MAP, @@ -382,3 +387,30 @@ class PermissionService: role = get_role_string(agent_slug) perms = NOTIFICATION_PERMISSIONS.get(role, {}) return bool(perms.get("can_send", False)) + + +# ============================================================================= +# ASYNC DATABASE LOOKUPS +# ============================================================================= + +# Roles with full access to all channels (bypass membership checks) +PRIVILEGED_ROLES = frozenset({AgentRole.CEO, AgentRole.AUDITOR, AgentRole.MAIN_PM}) + + +async def has_privileged_access(db: "AsyncSession", agent_id: UUID) -> bool: + """ + Check if agent has a privileged role (CEO, Auditor, Main PM). + + These roles have full access to all channels regardless of membership. + Queries by both id and slug since agent_id could be either + (CEO uses UUID-style slug, others use short slugs like "be-dev-1"). + """ + from roboco.db.tables import AgentTable + + result = await db.execute( + select(AgentTable.role).where( + (AgentTable.id == agent_id) | (AgentTable.slug == str(agent_id)) + ) + ) + role = result.scalar_one_or_none() + return role in PRIVILEGED_ROLES if role else False diff --git a/roboco/services/task.py b/roboco/services/task.py index 9247c512..65494c4a 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -66,7 +66,7 @@ class TaskService: "Task created", task_id=str(task.id), title=req.title, - team=req.team.value, + team=req.team if isinstance(req.team, str) else req.team.value, ) return task