refactor(BA-5777): address PR review feedback on bulk processor

- Rename BulkValidatorDecision to ValidatorDecision; the dataclass records
  one validator's verdict, so the Bulk prefix implied a set and conflicted
  with the surrounding Bulk* processor/result types
- Replace the _validator_scope asynccontextmanager with a plain async
  helper _run_validator; the CM yielded once with no pre/post split, so
  the bookend ceremony bought nothing over a regular call
- Drop the unused name parameter and _name attribute from the test's
  _AllowSetValidator (name() classmethod hardcodes the identifier)
- Note in _run that the service function runs once on the post-filter
  action so only entities that passed every validator reach it

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

Author: fregataa

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

@@ -1,7 +1,6 @@
 import logging
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
-from contextlib import asynccontextmanager
+from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import Any
@@ -26,7 +25,7 @@
 
 
 @dataclass(frozen=True)
-class BulkValidatorDecision:
+class ValidatorDecision:
     """One validator's per-entity verdict observed during bulk processing.
 
     Mirrors the ``SubStepResult`` pattern used by the scheduler history so
@@ -49,7 +48,7 @@ class BulkProcessResult[TBulkActionResult: BaseBulkActionResult]:
     """
 
     result: TBulkActionResult
-    validator_decisions: list[BulkValidatorDecision]
+    validator_decisions: list[ValidatorDecision]
 
 
 class BulkActionProcessor[
@@ -70,32 +69,28 @@ def __init__(
 
         self._validators = validators or []
 
-    @asynccontextmanager
-    async def _validator_scope(
+    async def _run_validator(
         self,
         validator: BulkActionValidator,
         action: TBulkAction,
         meta: BaseActionTriggerMeta,
-    ) -> AsyncIterator[BulkValidationResult]:
-        """Run one validator inside a bookend scope.
+    ) -> BulkValidationResult:
+        """Invoke one validator and emit its timing/trace log.
 
-        Yields the validator's ``BulkValidationResult`` so the caller can
-        record the decision inside the block. Timing and per-validator
-        logging live here rather than inside each validator implementation.
+        Timing and per-validator logging live here rather than inside each
+        validator implementation so the cross-cutting concern has one home.
         """
         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(
-                "bulk validator {} saw {} ids, denied {} in {:.3f}s",
-                validator.name(),
-                len(validation.allowed_entity_ids) + len(validation.denied_entities),
-                len(validation.denied_entities),
-                duration,
-            )
+        duration = (datetime.now(UTC) - started_at).total_seconds()
+        log.debug(
+            "bulk validator {} saw {} ids, denied {} in {:.3f}s",
+            validator.name(),
+            len(action.entity_ids),
+            len(validation.denied_entities),
+            duration,
+        )
+        return validation
 
     def _process_action(
         self,
@@ -120,20 +115,21 @@ async def _run(self, action: TBulkAction) -> BulkProcessResult[TBulkActionResult
         action_id = uuid.uuid4()
         action_trigger_meta = BaseActionTriggerMeta(action_id=action_id, started_at=started_at)
 
+        # Run every validator over the surviving ID set, then invoke the
+        # service function once on the final narrowed action — the service
+        # must only see IDs that passed every validator.
         current_action: TBulkAction = action
-        decisions: list[BulkValidatorDecision] = []
+        decisions: list[ValidatorDecision] = []
 
         for validator in self._validators:
-            async with self._validator_scope(
-                validator, current_action, action_trigger_meta
-            ) as validation:
-                decisions.append(
-                    BulkValidatorDecision(
-                        validator_name=validator.name(),
-                        results=validation,
-                    )
+            validation = await self._run_validator(validator, current_action, action_trigger_meta)
+            decisions.append(
+                ValidatorDecision(
+                    validator_name=validator.name(),
+                    results=validation,
                 )
-                current_action = self._process_action(current_action, validation)
+            )
+            current_action = self._process_action(current_action, validation)
 
         action_result = await self._runner.run(current_action, action_trigger_meta)
         return BulkProcessResult(result=action_result, validator_decisions=decisions)

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

@@ -20,7 +20,7 @@ class BulkValidationResult:
 
     ``BulkActionProcessor`` intersects ``allowed_entity_ids`` across
     validators and records each ``DeniedEntity`` — with its reason — on the
-    corresponding ``BulkValidatorDecision`` so the final response can
+    corresponding ``ValidatorDecision`` so the final response can
     surface *why* each ID was filtered out.
     """
 
@@ -32,7 +32,7 @@ class BulkActionValidator(ABC):
     @classmethod
     @abstractmethod
     def name(cls) -> str:
-        """Stable identifier used in ``BulkValidatorDecision.validator_name``.
+        """Stable identifier used in ``ValidatorDecision.validator_name``.
 
         Chosen by the implementation so logs and partial-success responses can
         attribute denials to a specific validator independently of the Python

tests/unit/manager/actions/test_bulk_processor.py

@@ -56,9 +56,8 @@ def entity_ids(self) -> list[str]:
 class _AllowSetValidator(BulkActionValidator):
     """Approves any ID in ``allowed``; anything else visible is denied."""
 
-    def __init__(self, allowed: set[str], name: str = "allow-set") -> None:
+    def __init__(self, allowed: set[str]) -> None:
         self._allowed = set(allowed)
-        self._name = name
 
     @classmethod
     @override