feat(BA-5777): add batch RBAC filtering infrastructure

- BaseBatchAction becomes a generic dataclass with entity_ids: list[str] +
  typed_entity_ids() abstract for native-type access
- BatchActionValidator exposes name() and validate() returning
  BatchValidationResult (allowed IDs + DeniedEntity with deny_reason)
- BatchActionProcessor runs each validator inside _validator_scope,
  narrows entity_ids in place, and returns BatchProcessResult carrying
  per-validator decisions for partial-success responses
- Remove orphan check_and_transit_status_multi path and 10 unused
  intermediate batch base classes

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: fregataa

src/ai/backend/manager/actions/action/batch.py

@@ -1,26 +1,26 @@
 from abc import abstractmethod
-from typing import TypeVar, override
+from dataclasses import dataclass
+from typing import Any, TypeVar, override
 
 from .base import BaseAction, BaseActionResult
-from .types import BatchFieldData
 
 
-class BaseBatchAction(BaseAction):
-    @override
-    def entity_id(self) -> str | None:
-        return None
+@dataclass
+class BaseBatchAction[T](BaseAction):
+    """Base class for actions operating on a batch of entities.
 
-    @abstractmethod
-    def entity_ids(self) -> list[str]:
-        raise NotImplementedError
+    ``entity_ids`` is stored as ``list[str]`` so ``BatchActionProcessor`` can
+    narrow it in place after each RBAC validator's verdict arrives without
+    caring about the concrete ID type. Concrete subclasses implement
+    ``typed_entity_ids()`` to surface the native ``list[T]`` (e.g.
+    ``list[SessionId]``) the service layer expects.
+    """
+
+    entity_ids: list[str]
 
     @abstractmethod
-    def field_data(self) -> BatchFieldData | None:
-        """
-        Returns batch field data containing the field type and IDs when the
-        action's targets exist as fields of another entity.
-        Returns None if these entities are not fields.
-        """
+    def typed_entity_ids(self) -> list[T]:
+        """Return ``entity_ids`` converted back to the native ID type ``T``."""
         raise NotImplementedError
 
 
@@ -34,5 +34,5 @@ def entity_ids(self) -> list[str]:
         raise NotImplementedError
 
 
-TBatchAction = TypeVar("TBatchAction", bound=BaseBatchAction)
+TBatchAction = TypeVar("TBatchAction", bound=BaseBatchAction[Any])
 TBatchActionResult = TypeVar("TBatchActionResult", bound=BaseBatchActionResult)

src/ai/backend/manager/actions/processor/batch.py

@@ -1,7 +1,10 @@
 import logging
 import uuid
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
 from datetime import UTC, datetime
+from typing import Any
 
 from ai.backend.logging.utils import BraceStyleAdapter
 from ai.backend.manager.actions.action import (
@@ -12,15 +15,45 @@
     BaseBatchActionResult,
 )
 from ai.backend.manager.actions.monitors.monitor import ActionMonitor
-from ai.backend.manager.actions.validator.batch import BatchActionValidator
+from ai.backend.manager.actions.validator.batch import (
+    BatchActionValidator,
+    BatchValidationResult,
+)
 
 from .base import ActionRunner
 
 log = BraceStyleAdapter(logging.getLogger(__spec__.name))
 
 
+@dataclass(frozen=True)
+class BatchValidatorDecision:
+    """One validator's per-entity verdict observed during batch processing.
+
+    Mirrors the ``SubStepResult`` pattern used by the scheduler history so
+    callers can trace where in the validator chain each ID was filtered and
+    *why*. ``results`` carries the validator's classification unchanged.
+    """
+
+    validator_name: str
+    results: BatchValidationResult
+
+
+@dataclass(frozen=True)
+class BatchProcessResult[TBatchActionResult: BaseBatchActionResult]:
+    """Outcome of a ``BatchActionProcessor`` run.
+
+    ``result`` is what the service function returned for the permitted subset
+    of entity IDs. ``validator_decisions`` keeps the per-validator trace in
+    iteration order; callers assemble the partial-success response by
+    walking it (each decision carries the denied IDs and their reasons).
+    """
+
+    result: TBatchActionResult
+    validator_decisions: list[BatchValidatorDecision]
+
+
 class BatchActionProcessor[
-    TBatchAction: BaseBatchAction,
+    TBatchAction: BaseBatchAction[Any],
     TBatchActionResult: BaseBatchActionResult,
 ]:
     _validators: Sequence[BatchActionValidator]
@@ -37,14 +70,72 @@ def __init__(
 
         self._validators = validators or []
 
-    async def _run(self, action: TBatchAction) -> TBatchActionResult:
+    @asynccontextmanager
+    async def _validator_scope(
+        self,
+        validator: BatchActionValidator,
+        action: TBatchAction,
+        meta: BaseActionTriggerMeta,
+    ) -> AsyncIterator[BatchValidationResult]:
+        """Run one validator inside a bookend scope.
+
+        Yields the validator's ``BatchValidationResult`` so the caller can
+        record the decision inside the block. Timing and per-validator
+        logging live here rather than inside each validator implementation.
+        """
+        started_at = datetime.now(UTC)
+        validation = await validator.validate(action, meta)
+        try:
+            yield validation
+        finally:
+            duration = (datetime.now(UTC) - started_at).total_seconds()
+            log.debug(
+                "batch validator {} saw {} ids, denied {} in {:.3f}s",
+                validator.name(),
+                len(validation.allowed_entity_ids) + len(validation.denied_entities),
+                len(validation.denied_entities),
+                duration,
+            )
+
+    def _process_action(
+        self,
+        current_action: TBatchAction,
+        validation: BatchValidationResult,
+    ) -> TBatchAction:
+        """Narrow ``current_action.entity_ids`` in place using one validator's verdict.
+
+        Returns the action unchanged when the validator denied nothing.
+        """
+        if not validation.denied_entities:
+            return current_action
+        allowed_set = set(validation.allowed_entity_ids)
+        current_action.entity_ids = [eid for eid in current_action.entity_ids if eid in allowed_set]
+        return current_action
+
+    async def _run(self, action: TBatchAction) -> BatchProcessResult[TBatchActionResult]:
         started_at = datetime.now(UTC)
         action_id = uuid.uuid4()
         action_trigger_meta = BaseActionTriggerMeta(action_id=action_id, started_at=started_at)
+
+        current_action: TBatchAction = action
+        decisions: list[BatchValidatorDecision] = []
+
         for validator in self._validators:
-            await validator.validate(action, action_trigger_meta)
+            async with self._validator_scope(
+                validator, current_action, action_trigger_meta
+            ) as validation:
+                decisions.append(
+                    BatchValidatorDecision(
+                        validator_name=validator.name(),
+                        results=validation,
+                    )
+                )
+                current_action = self._process_action(current_action, validation)
 
-        return await self._runner.run(action, action_trigger_meta)
+        action_result = await self._runner.run(current_action, action_trigger_meta)
+        return BatchProcessResult(result=action_result, validator_decisions=decisions)
 
-    async def wait_for_complete(self, action: TBatchAction) -> TBatchActionResult:
+    async def wait_for_complete(
+        self, action: TBatchAction
+    ) -> BatchProcessResult[TBatchActionResult]:
         return await self._run(action)

src/ai/backend/manager/actions/validator/batch.py

@@ -1,10 +1,57 @@
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Any
 
 from ai.backend.manager.actions.action import BaseActionTriggerMeta
 from ai.backend.manager.actions.action.batch import BaseBatchAction
 
 
+@dataclass(frozen=True)
+class DeniedEntity:
+    """A batch entity that a validator rejected, paired with its reason."""
+
+    entity_id: str
+    deny_reason: str
+
+
+@dataclass(frozen=True)
+class BatchValidationResult:
+    """Per-entity validation outcome for a batch action.
+
+    ``BatchActionProcessor`` intersects ``allowed_entity_ids`` across
+    validators and records each ``DeniedEntity`` — with its reason — on the
+    corresponding ``BatchValidatorDecision`` so the final response can
+    surface *why* each ID was filtered out.
+    """
+
+    allowed_entity_ids: list[str]
+    denied_entities: list[DeniedEntity]
+
+
 class BatchActionValidator(ABC):
+    @classmethod
+    @abstractmethod
+    def name(cls) -> str:
+        """Stable identifier used in ``BatchValidatorDecision.validator_name``.
+
+        Chosen by the implementation so logs and partial-success responses can
+        attribute denials to a specific validator independently of the Python
+        class name.
+        """
+        raise NotImplementedError
+
     @abstractmethod
-    async def validate(self, action: BaseBatchAction, meta: BaseActionTriggerMeta) -> None:
-        raise NotImplementedError("Subclasses must implement the validate method")
+    async def validate(
+        self, action: BaseBatchAction[Any], meta: BaseActionTriggerMeta
+    ) -> BatchValidationResult:
+        """Validate the batch action and return per-entity permission results.
+
+        Implementations must classify every ID in ``action.entity_ids`` as
+        either allowed or denied. Validators that cannot make a decision for
+        an ID should treat it as allowed.
+
+        The processor wraps each call in its own async context manager so
+        cross-cutting concerns (timing, audit) live in one place — validators
+        do not need to own them.
+        """
+        raise NotImplementedError

src/ai/backend/manager/actions/validators/rbac/batch.py

@@ -1,8 +1,11 @@
-from typing import override
+from typing import Any, override
 
 from ai.backend.manager.actions.action import BaseActionTriggerMeta
 from ai.backend.manager.actions.action.batch import BaseBatchAction
-from ai.backend.manager.actions.validator.batch import BatchActionValidator
+from ai.backend.manager.actions.validator.batch import (
+    BatchActionValidator,
+    BatchValidationResult,
+)
 from ai.backend.manager.repositories.permission_controller.repository import (
     PermissionControllerRepository,
 )
@@ -15,7 +18,18 @@ def __init__(
     ) -> None:
         self._repository = repository
 
+    @classmethod
+    @override
+    def name(cls) -> str:
+        return "rbac"
+
     @override
-    async def validate(self, action: BaseBatchAction, meta: BaseActionTriggerMeta) -> None:
-        # TODO: implement RBAC validation logic
-        pass
+    async def validate(
+        self, action: BaseBatchAction[Any], meta: BaseActionTriggerMeta
+    ) -> BatchValidationResult:
+        # TODO: wire this to PermissionControllerRepository.check_batch_permission_with_scope_chain().
+        # Until then, every entity is treated as allowed so legacy behavior is preserved.
+        return BatchValidationResult(
+            allowed_entity_ids=list(action.entity_ids),
+            denied_entities=[],
+        )

src/ai/backend/manager/services/artifact/actions/base.py

@@ -2,7 +2,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import EntityType
-from ai.backend.manager.actions.action import BaseAction, BaseBatchAction
+from ai.backend.manager.actions.action import BaseAction
 from ai.backend.manager.actions.action.scope import BaseScopeAction, BaseScopeActionResult
 from ai.backend.manager.actions.action.single_entity import (
     BaseSingleEntityAction,
@@ -19,14 +19,6 @@ def entity_type(cls) -> EntityType:
         return EntityType.ARTIFACT
 
 
-@dataclass
-class ArtifactBatchAction(BaseBatchAction):
-    @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.ARTIFACT
-
-
 @dataclass
 class ArtifactScopeAction(BaseScopeAction):
     @override

src/ai/backend/manager/services/artifact_registry/actions/base.py

@@ -1,7 +1,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import EntityType
-from ai.backend.manager.actions.action import BaseAction, BaseBatchAction
+from ai.backend.manager.actions.action import BaseAction
 from ai.backend.manager.actions.action.scope import BaseScopeAction, BaseScopeActionResult
 from ai.backend.manager.actions.action.single_entity import (
     BaseSingleEntityAction,
@@ -17,13 +17,6 @@ def entity_type(cls) -> EntityType:
         return EntityType.ARTIFACT_REGISTRY
 
 
-class ArtifactBatchRegistryAction(BaseBatchAction):
-    @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.ARTIFACT_REGISTRY
-
-
 class ArtifactRegistryScopeAction(BaseScopeAction):
     @override
     @classmethod

src/ai/backend/manager/services/artifact_revision/actions/base.py

@@ -2,7 +2,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import EntityType
-from ai.backend.manager.actions.action import BaseAction, BaseBatchAction
+from ai.backend.manager.actions.action import BaseAction
 
 
 @dataclass
@@ -11,11 +11,3 @@ class ArtifactRevisionAction(BaseAction):
     @classmethod
     def entity_type(cls) -> EntityType:
         return EntityType.ARTIFACT_REVISION
-
-
-@dataclass
-class ArtifactRevisionBatchAction(BaseBatchAction):
-    @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.ARTIFACT_REVISION

src/ai/backend/manager/services/container_registry/actions/base.py

@@ -2,7 +2,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import EntityType
-from ai.backend.manager.actions.action import BaseAction, BaseBatchAction
+from ai.backend.manager.actions.action import BaseAction
 
 
 @dataclass
@@ -11,11 +11,3 @@ class ContainerRegistryAction(BaseAction):
     @classmethod
     def entity_type(cls) -> EntityType:
         return EntityType.CONTAINER_REGISTRY
-
-
-@dataclass
-class ContainerRegistryBatchAction(BaseBatchAction):
-    @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.CONTAINER_REGISTRY

src/ai/backend/manager/services/image/actions/base.py

@@ -2,7 +2,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import EntityType
-from ai.backend.manager.actions.action import BaseAction, BaseBatchAction
+from ai.backend.manager.actions.action import BaseAction
 from ai.backend.manager.actions.action.single_entity import (
     BaseSingleEntityAction,
     BaseSingleEntityActionResult,
@@ -18,14 +18,6 @@ def entity_type(cls) -> EntityType:
         return EntityType.IMAGE
 
 
-@dataclass
-class ImageBatchAction(BaseBatchAction):
-    @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.IMAGE
-
-
 @dataclass
 class ImageSingleEntityAction(BaseSingleEntityAction):
     @override