fix: gate the events/validation routes; unify the handler auth gate

server/osa/application/api/v1/routes/events.py

@@ -1,14 +1,21 @@
-"""Events API routes - changefeed for federation."""
+"""Events API routes - changefeed for federation.
+
+Thin HTTP ↔ DTO coercion only: the changefeed read (look-ahead pagination,
+payload shaping) and its explicit ``public()`` gate live in
+``ListEventsHandler`` (arch-survey 2026-08-16 F1 — this route previously
+injected the EventLog service directly, bypassing the gate machinery).
+"""
 
 from datetime import datetime
+from typing import Literal
 from uuid import UUID
 
 from dishka.integrations.fastapi import DishkaRoute, FromDishka
 from fastapi import APIRouter, Query
 from pydantic import BaseModel
 
 from osa.domain.shared.event import EventId
-from osa.domain.shared.event_log import EventLog
+from osa.domain.shared.event_log import ListEvents, ListEventsHandler
 
 router = APIRouter(
     prefix="/events",
@@ -36,41 +43,32 @@ class EventListResponse(BaseModel):
 
 @router.get("")
 async def list_events(
-    event_log: FromDishka[EventLog],
+    handler: FromDishka[ListEventsHandler],
     limit: int = Query(50, ge=1, le=500, description="Maximum number of events"),
     after: UUID | None = Query(None, description="Cursor: return events after this ID"),
     types: list[str] | None = Query(None, description="Filter by event types"),
-    order: str = Query("asc", description="Order: 'asc' (oldest first) or 'desc' (newest first)"),
+    order: Literal["asc", "desc"] = Query(
+        "asc", description="'asc' (oldest first, federation) or 'desc' (newest first)"
+    ),
 ) -> EventListResponse:
     """List events from the event log (changefeed).
 
     Use order=asc (default) for federation, order=desc for viewing recent events.
     Use the cursor to paginate through results.
     """
-    newest_first = order == "desc"
-    after_id = EventId(after) if after else None
-    events = await event_log.list_events(
-        limit=limit + 1, after=after_id, event_types=types, newest_first=newest_first
+    page = await handler.run(
+        ListEvents(
+            limit=limit,
+            after=EventId(after) if after else None,
+            types=types,
+            order=order,
+        )
     )
-
-    # Check if there are more results
-    has_more = len(events) > limit
-    if has_more:
-        events = events[:limit]
-
-    # Cursor is the ID of the last event
-    cursor = str(events[-1].id) if events else None
-
     return EventListResponse(
         events=[
-            EventResponse(
-                id=e.id,
-                type=type(e).__name__,
-                created_at=e.created_at,
-                data=e.model_dump(mode="json", exclude={"id", "created_at"}),
-            )
-            for e in events
+            EventResponse(id=e.id, type=e.type, created_at=e.created_at, data=e.data)
+            for e in page.events
         ],
-        cursor=cursor,
-        has_more=has_more,
+        cursor=page.cursor,
+        has_more=page.has_more,
     )

server/osa/application/api/v1/routes/validation.py

@@ -1,17 +1,22 @@
-"""Validation API routes."""
+"""Validation API routes.
+
+Thin HTTP ↔ DTO coercion only: the status→shape rule and the explicit
+``public()`` gate live in ``GetValidationRunHandler``; a missing run raises
+``NotFoundError``, mapped centrally (arch-survey 2026-08-16 F1 — this route
+previously injected ValidationService directly and owned the decision tree).
+"""
 
 from datetime import datetime
 
 from dishka.integrations.fastapi import DishkaRoute, FromDishka
-from fastapi import APIRouter, HTTPException, status
+from fastapi import APIRouter
 from pydantic import BaseModel, Field
 
-from osa.domain.validation.model import (
-    HookStatus,
-    RunStatus,
+from osa.domain.validation.model import HookStatus, RunStatus
+from osa.domain.validation.query.get_validation_run import (
+    GetValidationRun,
+    GetValidationRunHandler,
 )
-from osa.domain.validation.service.validation import ValidationService
-
 
 router = APIRouter(
     prefix="/validation",
@@ -66,40 +71,15 @@ class ValidationStatusResponse(BaseModel):
 )
 async def get_validation_status(
     run_id: str,
-    service: FromDishka[ValidationService],
+    handler: FromDishka[GetValidationRunHandler],
 ) -> ValidationStatusResponse:
-    run = await service.get_run(run_id)
-    if not run:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Validation run not found: {run_id}",
-        )
-
-    results_dto = [
-        HookResultDTO(
-            hook_name=r.hook_name.root,
-            status=r.status,
-            rejection_reason=r.rejection_reason,
-            error_message=r.error_message,
-            duration_seconds=r.duration_seconds,
-        )
-        for r in run.results
-    ]
-
-    summary = None
-    progress = None
-
-    if run.status in (RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.REJECTED):
-        summary = run.summary
-    elif run.status == RunStatus.RUNNING:
-        progress = {"status": "running"}
-
+    result = await handler.run(GetValidationRun(run_id=run_id))
     return ValidationStatusResponse(
-        run_id=run_id,
-        status=run.status,
-        summary=summary,
-        progress=progress,
-        results=results_dto,
-        started_at=run.started_at,
-        completed_at=run.completed_at,
+        run_id=result.run_id,
+        status=result.status,
+        summary=result.summary,
+        progress=result.progress.model_dump() if result.progress else None,
+        results=[HookResultDTO(**r.model_dump()) for r in result.results],
+        started_at=result.started_at,
+        completed_at=result.completed_at,
     )

server/osa/domain/shared/command.py

@@ -1,111 +1,33 @@
-"""Command and CommandHandler base classes with authorization gate."""
+"""Command and CommandHandler base classes with authorization gate.
+
+The gate wrapper and metaclass live once in :mod:`osa.domain.shared.handler`;
+this module contributes only the write-side vocabulary: the ``Command`` DTO
+base and the ``Result``-bound result TypeVar.
+"""
 
 from __future__ import annotations
 
-from abc import ABCMeta, abstractmethod
-from collections.abc import Callable, Coroutine
-from dataclasses import dataclass
-from functools import wraps
-from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, dataclass_transform
+from abc import abstractmethod
+from typing import TYPE_CHECKING, ClassVar, Generic, TypeVar
 
 from pydantic import BaseModel
 
+from osa.domain.shared.handler import HandlerMeta, Result
+
 if TYPE_CHECKING:
     from osa.domain.shared.authorization.gate import Gate
 
-
-class Command(BaseModel): ...
+__all__ = ["Command", "CommandHandler", "Result"]
 
 
-class Result(BaseModel): ...
+class Command(BaseModel): ...
 
 
 C = TypeVar("C", bound=Command)
 R = TypeVar("R", bound=Result)
 
-# Unbound async handler method: (self, cmd) -> Coroutine -> Result
-_HandlerMethod = Callable[..., Coroutine[Any, Any, Any]]
-
-
-def _wrap_run_with_auth(cls: type, original_run: _HandlerMethod) -> _HandlerMethod:
-    """Wrap the run() method with __auth__ gate evaluation."""
-
-    @wraps(original_run)
-    async def auth_wrapped_run(self: Any, cmd: Any) -> Any:
-        from osa.domain.shared.authorization.gate import AtLeast, Gate, Public, RequiresScope
-        from osa.domain.shared.error import AuthorizationError, ConfigurationError
-
-        auth_gate = getattr(type(self), "__auth__", None)
-
-        if not isinstance(auth_gate, Gate):
-            raise ConfigurationError(f"Handler {type(self).__name__} has no __auth__ declaration")
-
-        if isinstance(auth_gate, Public):
-            return await original_run(self, cmd)
-
-        if isinstance(auth_gate, AtLeast):
-            from osa.domain.auth.model.principal import Principal
-
-            principal = getattr(self, "principal", None)
-            if not isinstance(principal, Principal):
-                raise AuthorizationError(
-                    "Authentication required",
-                    code="missing_token",
-                )
-
-            if not principal.has_role(auth_gate.role):
-                raise AuthorizationError(
-                    f"Access denied: insufficient role for {type(self).__name__}",
-                    code="access_denied",
-                )
-
-            return await original_run(self, cmd)
-
-        if isinstance(auth_gate, RequiresScope):
-            from osa.domain.auth.model.principal import Principal
-            from osa.domain.auth.model.role import Role
-
-            principal = getattr(self, "principal", None)
-            if not isinstance(principal, Principal):
-                raise AuthorizationError(
-                    "Authentication required",
-                    code="missing_token",
-                )
-
-            if not (principal.has_scope(auth_gate.scope) or principal.has_role(Role.ADMIN)):
-                raise AuthorizationError(
-                    f"Access denied: missing scope {auth_gate.scope!r} for {type(self).__name__}",
-                    code="access_denied",
-                )
-
-            return await original_run(self, cmd)
-
-        raise ConfigurationError(  # pragma: no cover — future gate types handled here
-            f"Handler {type(self).__name__} has unhandled __auth__ type: {type(auth_gate).__name__}"
-        )
-
-    return auth_wrapped_run
-
-
-@dataclass_transform()
-class _CommandHandlerMeta(ABCMeta):
-    """Metaclass that combines ABC with auto-dataclass and __auth__ gate for subclasses."""
-
-    def __new__(mcs, name: str, bases: tuple[type, ...], namespace: dict[str, Any]):
-        cls = super().__new__(mcs, name, bases, namespace)
-        if any(isinstance(b, mcs) for b in bases):
-            cls = dataclass(cls)
-
-            # Wrap run() with auth gate
-            original_run = cls.__dict__.get("run")
-            if original_run is not None:
-                wrapped = _wrap_run_with_auth(cls, original_run)
-                cls.run = wrapped
-
-        return cls
-
 
-class CommandHandler(Generic[C, R], metaclass=_CommandHandlerMeta):
+class CommandHandler(Generic[C, R], metaclass=HandlerMeta):
     """Base class for command handlers. Subclasses are automatically dataclasses.
 
     Declare __auth__ to enforce role-based access:

server/osa/domain/shared/event_log.py

@@ -1,7 +1,14 @@
-"""EventLog - service for querying the event store (changefeed)."""
+"""EventLog — the event-store changefeed: service + its query handler."""
 
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import Field
+
+from osa.domain.shared.authorization.gate import public
 from osa.domain.shared.event import Event, EventId
 from osa.domain.shared.port.event_repository import EventRepository
+from osa.domain.shared.query import Query, QueryHandler, Result
 from osa.domain.shared.service import Service
 
 
@@ -43,3 +50,68 @@ async def count(self, event_types: list[str] | None = None) -> int:
     async def get(self, event_id: EventId) -> Event | None:
         """Get a single event by ID."""
         return await self._repo.get(event_id)
+
+
+class ListEvents(Query):
+    """Changefeed page request. ``order="asc"`` (oldest first) is the
+    federation direction; ``"desc"`` serves recent-events views. Anything else
+    is a validation error — never a silent default."""
+
+    limit: int = Field(default=50, ge=1, le=500)
+    after: EventId | None = None
+    types: list[str] | None = None
+    order: Literal["asc", "desc"] = "asc"
+
+
+class EventEntry(Result):
+    """One changefeed event: identity, type discriminator, and the payload
+    (the event body minus identity/timestamp, which ride alongside)."""
+
+    id: EventId
+    type: str
+    created_at: datetime
+    data: dict[str, Any]
+
+
+class EventPage(Result):
+    events: list[EventEntry]
+    cursor: str | None
+    has_more: bool
+
+
+class ListEventsHandler(QueryHandler[ListEvents, EventPage]):
+    """The /events changefeed read (arch-survey 2026-08-16 F1).
+
+    ``public()`` is deliberate, not an omission: the changefeed is the
+    federation surface (CLAUDE.md API §9) and mirroring nodes are anonymous.
+    Tightening access is a one-line gate change here, boot-validated.
+    """
+
+    __auth__ = public()
+
+    event_log: EventLog
+
+    async def run(self, cmd: ListEvents) -> EventPage:
+        # limit+1 look-ahead: one surplus row proves a further page exists.
+        events = await self.event_log.list_events(
+            limit=cmd.limit + 1,
+            after=cmd.after,
+            event_types=cmd.types,
+            newest_first=cmd.order == "desc",
+        )
+        has_more = len(events) > cmd.limit
+        if has_more:
+            events = events[: cmd.limit]
+        return EventPage(
+            events=[
+                EventEntry(
+                    id=e.id,
+                    type=type(e).__name__,
+                    created_at=e.created_at,
+                    data=e.model_dump(mode="json", exclude={"id", "created_at"}),
+                )
+                for e in events
+            ],
+            cursor=str(events[-1].id) if events else None,
+            has_more=has_more,
+        )