feat(AGX1-275): per-RPC task permission rewire and 404/403 wrap

agentex/src/api/routes/agents.py

@@ -38,6 +38,7 @@
     DAuthorizedResourceIds,
 )
 from src.utils.logging import make_logger
+from src.utils.task_authorization import _check_task_or_collapse_to_404
 
 logger = make_logger(__name__)
 
@@ -333,26 +334,31 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await _check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.update,
                 )
             elif task_name is not None:
-                # Task name provided - check if task exists
+                # try/else (not try/except wrapping the whole block): a denied
+                # update on an existing task must surface as 404 to the caller,
+                # NOT silently fall through to the create check below — that
+                # would let "I'm denied update" masquerade as "task is absent"
+                # and grant create access.
                 try:
                     existing_task = await task_service.get_task(name=task_name)
-                    # Task exists - require execute permission on the specific task
-                    await authorization_service.check(
-                        resource=AgentexResource.task(existing_task.id),
-                        operation=AuthorizedOperationType.execute,
-                    )
                 except ItemDoesNotExist:
                     # Task doesn't exist - will be created, require create permission
                     await authorization_service.check(
                         resource=AgentexResource.task("*"),
                         operation=AuthorizedOperationType.create,
                     )
+                else:
+                    await _check_task_or_collapse_to_404(
+                        authorization_service,
+                        existing_task.id,
+                        AuthorizedOperationType.update,
+                    )
             else:
                 # No identifier provided - creating new task, require create permission
                 await authorization_service.check(
@@ -365,20 +371,22 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await _check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.update,
                 )
             elif task_name is not None:
-                # Task name provided - look up task and check execute permission
+                # `get_task` raises ItemDoesNotExist for absent rows (= 404
+                # naturally); the wrap collapses present-but-denied to the same
+                # shape so callers can't distinguish.
                 existing_task = await task_service.get_task(name=task_name)
-                await authorization_service.check(
-                    resource=AgentexResource.task(existing_task.id),
-                    operation=AuthorizedOperationType.execute,
+                await _check_task_or_collapse_to_404(
+                    authorization_service,
+                    existing_task.id,
+                    AuthorizedOperationType.update,
                 )
             else:
-                # No identifier provided - this shouldn't happen but handle gracefully
                 raise ValueError(
                     "Either task_id or task_name must be provided for event/send"
                 )
@@ -388,20 +396,19 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await _check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.cancel,
                 )
             elif task_name is not None:
-                # Task name provided - look up task and check execute permission
                 existing_task = await task_service.get_task(name=task_name)
-                await authorization_service.check(
-                    resource=AgentexResource.task(existing_task.id),
-                    operation=AuthorizedOperationType.execute,
+                await _check_task_or_collapse_to_404(
+                    authorization_service,
+                    existing_task.id,
+                    AuthorizedOperationType.cancel,
                 )
             else:
-                # No identifier provided - this shouldn't happen but handle gracefully
                 raise ValueError(
                     "Either task_id or task_name must be provided for task/cancel"
                 )

agentex/src/api/routes/checkpoints.py

@@ -2,6 +2,10 @@
 
 from fastapi import APIRouter, Response
 
+from src.api.schemas.authorization_types import (
+    AgentexResourceType,
+    AuthorizedOperationType,
+)
 from src.api.schemas.checkpoints import (
     BlobResponse,
     CheckpointListItem,
@@ -14,10 +18,6 @@
     PutWritesRequest,
     WriteResponse,
 )
-from src.api.schemas.authorization_types import (
-    AgentexResourceType,
-    AuthorizedOperationType,
-)
 from src.domain.use_cases.checkpoints_use_case import DCheckpointsUseCase
 from src.utils.authorization_shortcuts import DAuthorizedBodyId
 from src.utils.logging import make_logger
@@ -95,7 +95,7 @@ async def put_checkpoint(
     request: PutCheckpointRequest,
     checkpoints_use_case: DCheckpointsUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute, field_name="thread_id"
+        AgentexResourceType.task, AuthorizedOperationType.update, field_name="thread_id"
     ),
 ) -> PutCheckpointResponse:
     blobs = [
@@ -133,7 +133,7 @@ async def put_writes(
     request: PutWritesRequest,
     checkpoints_use_case: DCheckpointsUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute, field_name="thread_id"
+        AgentexResourceType.task, AuthorizedOperationType.update, field_name="thread_id"
     ),
 ) -> Response:
     writes = [

agentex/src/api/routes/messages.py

@@ -82,7 +82,7 @@ async def batch_create_messages(
     request: BatchCreateTaskMessagesRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> list[TaskMessage]:
     # Convert each content from API schema to entity schema
@@ -110,7 +110,7 @@ async def batch_update_messages(
     request: BatchUpdateTaskMessagesRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> list[TaskMessage]:
     task_message_entities = await message_use_case.update_batch(
@@ -131,7 +131,7 @@ async def create_message(
     request: CreateTaskMessageRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> TaskMessage:
     task_message_entity = await message_use_case.create(
@@ -152,7 +152,7 @@ async def update_message(
     message_id: str,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> TaskMessage:
     task_message_entity = await message_use_case.update(

agentex/src/api/schemas/authorization_types.py

@@ -9,6 +9,7 @@ class AuthorizedOperationType(StrEnum):
     update = "update"
     delete = "delete"
     execute = "execute"
+    cancel = "cancel"
 
 
 class AgentexResourceType(StrEnum):

agentex/src/utils/authorization_shortcuts.py

@@ -13,6 +13,7 @@
 from src.domain.repositories.task_repository import DTaskRepository
 from src.domain.repositories.task_state_repository import DTaskStateRepository
 from src.domain.services.authorization_service import DAuthorizationService
+from src.utils.task_authorization import _check_task_or_collapse_to_404
 
 
 async def _get_parent_task_id(
@@ -51,12 +52,10 @@ async def _ensure_authorized_id(
             task_id = await _get_parent_task_id(
                 resource_type, resource_id, event_repository, state_repository
             )
-            await authorization.check(
-                resource=AgentexResource.task(task_id),
-                operation=operation,
-            )
+            await _check_task_or_collapse_to_404(authorization, task_id, operation)
+        elif resource_type == AgentexResourceType.task:
+            await _check_task_or_collapse_to_404(authorization, resource_id, operation)
         else:
-            # For direct resources, check directly
             await authorization.check(
                 resource=AgentexResource(type=resource_type, selector=resource_id),
                 operation=operation,
@@ -88,12 +87,10 @@ async def _ensure_authorized_query(
             task_id = await _get_parent_task_id(
                 resource_type, resource_id, event_repository, state_repository
             )
-            await authorization.check(
-                resource=AgentexResource.task(task_id),
-                operation=operation,
-            )
+            await _check_task_or_collapse_to_404(authorization, task_id, operation)
+        elif resource_type == AgentexResourceType.task:
+            await _check_task_or_collapse_to_404(authorization, resource_id, operation)
         else:
-            # For direct resources, check directly
             await authorization.check(
                 resource=AgentexResource(type=resource_type, selector=resource_id),
                 operation=operation,
@@ -118,10 +115,13 @@ async def _ensure_authorized_body_field(
         body = await request.json()
         field_value = body[field_name]
 
-        await authorization.check(
-            resource=AgentexResource(type=resource_type, selector=field_value),
-            operation=operation,
-        )
+        if resource_type == AgentexResourceType.task:
+            await _check_task_or_collapse_to_404(authorization, field_value, operation)
+        else:
+            await authorization.check(
+                resource=AgentexResource(type=resource_type, selector=field_value),
+                operation=operation,
+            )
         return field_value
 
     return Annotated[str, Depends(_ensure_authorized_body_field)]
@@ -164,12 +164,21 @@ async def _ensure_authorized_name(
         resource_id = resource_name
         repository = registry[resource_type]
 
+        # Lookup-before-authz: if the name isn't present, `repository.get` raises
+        # ItemDoesNotExist (→ 404), which is what we want for absent resources.
+        # The present-but-denied case is handled per-resource below.
         resource = await repository.get(name=resource_id)
 
-        await authorization.check(
-            resource=AgentexResource(type=resource_type, selector=resource.id),
-            operation=operation,
-        )
+        if resource_type == AgentexResourceType.task:
+            # Tasks: collapse denial to 404 so name probes can't distinguish
+            # "present in another tenant" from "absent" (tasks.name is globally
+            # unique — any 403 leak here probes the whole system, not a tenant).
+            await _check_task_or_collapse_to_404(authorization, resource.id, operation)
+        else:
+            await authorization.check(
+                resource=AgentexResource(type=resource_type, selector=resource.id),
+                operation=operation,
+            )
         return resource_id
 
     return Annotated[str, Depends(_ensure_authorized_name)]

agentex/src/utils/task_authorization.py

@@ -0,0 +1,39 @@
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
+
+
+async def _check_task_or_collapse_to_404(
+    authorization,
+    task_id: str,
+    operation: AuthorizedOperationType,
+) -> None:
+    """Issue a check on a task resource. On any denial, surface 404 — even when
+    the task exists.
+
+    The 403/404 split cannot be done safely here: ``TaskORM`` has no tenant
+    column, ``task_repository.get`` does an unfiltered primary-key lookup, and
+    ``AuthorizationError`` carries no deny-reason discriminator. Returning 403
+    when the row exists and 404 when it doesn't leaks cross-tenant existence
+    (caller probes "does task X exist?" and gets distinguishable responses).
+
+    Until tasks carry tenant scope (or agentex-auth's deny distinguishes
+    "cross-tenant" from "in-tenant lacking permission"), the safer default is
+    to collapse both into 404. Trade-off: a user with ``read`` but not
+    ``update`` permission on an in-tenant task sees 404 on update attempts
+    instead of 403. UX regression for in-tenant permission granularity, but
+    eliminates the cross-tenant existence leak.
+
+    TODO(AGX1-290): Restore the 403/404 split for same-tenant calls once
+    tasks carry tenant/account_id scope at the data layer.
+    """
+    try:
+        await authorization.check(
+            resource=AgentexResource.task(task_id),
+            operation=operation,
+        )
+    except AuthorizationError:
+        raise ItemDoesNotExist(f"Item with id '{task_id}' does not exist.") from None