From b82cc37657ba81f2bd0175907cd589c348ee6ee1 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Sun, 16 Aug 2026 13:05:10 +0100 Subject: [PATCH 1/2] refactor: one home for the handler auth gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gate wrapper, metaclass, and Result base were duplicated between shared/command.py and shared/query.py (~100 lines) and had already drifted: the query copy built a logger inside the request path and debug-logged role checks — reads were logged, writes weren't, by accident. The startup validator already treated the two handler kinds as one thing. Everything shared now lives once in shared/handler.py (wrap_run_with_auth + HandlerMeta + Result); CommandHandler/QueryHandler become facades contributing only their DTO base and result TypeVar bound (Command with Result-bound R; Query with unbound R, reason documented). The request-path debug log is dropped deliberately: startup prints the gate table once, and denials raise typed AuthorizationErrors recorded by the central error mapper. Parity tests pin the contract for both facades — public/at_least/ requires_scope/missing-gate behave identically on the write and read sides, and the Result base is one class — so the gate semantics cannot fork again. Arch-survey 2026-08-16 finding F2. --- server/osa/domain/shared/command.py | 104 ++----------- server/osa/domain/shared/handler.py | 110 +++++++++++++ server/osa/domain/shared/query.py | 116 ++------------ .../domain/shared/test_handler_gate_parity.py | 145 ++++++++++++++++++ 4 files changed, 281 insertions(+), 194 deletions(-) create mode 100644 server/osa/domain/shared/handler.py create mode 100644 server/tests/unit/domain/shared/test_handler_gate_parity.py diff --git a/server/osa/domain/shared/command.py b/server/osa/domain/shared/command.py index ce0f4031..239a130f 100644 --- a/server/osa/domain/shared/command.py +++ b/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: diff --git a/server/osa/domain/shared/handler.py b/server/osa/domain/shared/handler.py new file mode 100644 index 00000000..5e3a0bd7 --- /dev/null +++ b/server/osa/domain/shared/handler.py @@ -0,0 +1,110 @@ +"""The single home of handler mechanics: auth-gate wrapping + the metaclass. + +``CommandHandler`` and ``QueryHandler`` are thin facades over this module — +they differ only in their DTO base (``Command`` vs ``Query``) and their result +TypeVar bound. Everything they share lives here exactly once, so the gate +semantics cannot fork between the read and write sides (arch-survey +2026-08-16 F2: the wrapper had been duplicated and had already drifted). + +Gate evaluation is deliberately log-free on the request path: the startup +validator prints the full gate table once, and denials raise typed +``AuthorizationError``s that the central error mapper records. +""" + +from __future__ import annotations + +from abc import ABCMeta +from collections.abc import Callable, Coroutine +from dataclasses import dataclass +from functools import wraps +from typing import Any, dataclass_transform + +from pydantic import BaseModel + + +class Result(BaseModel): ... + + +# 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 a handler's ``run()`` 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 HandlerMeta(ABCMeta): + """ABC + auto-dataclass + ``__auth__`` gate wrap, applied to subclasses only. + + The facade classes themselves (``CommandHandler``/``QueryHandler``) have no + base carrying this metaclass, so they are left untouched; every concrete + handler subclass is dataclass-ified and gate-wrapped. + """ + + 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) + + original_run = cls.__dict__.get("run") + if original_run is not None: + cls.run = wrap_run_with_auth(cls, original_run) + + return cls diff --git a/server/osa/domain/shared/query.py b/server/osa/domain/shared/query.py index e7f2d894..783e4e13 100644 --- a/server/osa/domain/shared/query.py +++ b/server/osa/domain/shared/query.py @@ -1,23 +1,26 @@ -"""Query and QueryHandler base classes with authorization gate.""" +"""Query and QueryHandler base classes with authorization gate. + +The gate wrapper and metaclass live once in :mod:`osa.domain.shared.handler`; +this module contributes only the read-side vocabulary: the ``Query`` DTO base +and an *unbound* 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 Query(BaseModel): ... +__all__ = ["Query", "QueryHandler", "Result"] -class Result(BaseModel): ... +class Query(BaseModel): ... C = TypeVar("C", bound=Query) @@ -26,101 +29,8 @@ class Result(BaseModel): ... # the conventional base for handler-specific DTOs. R = TypeVar("R") -# Unbound async handler method: (self, cmd) -> Coroutine -> Result -_HandlerMethod = Callable[..., Coroutine[Any, Any, Any]] - - -def _wrap_query_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): - import logging as _logging - - from osa.domain.auth.model.principal import Principal - - _auth_logger = _logging.getLogger("osa.authz") - - principal = getattr(self, "principal", None) - if not isinstance(principal, Principal): - raise AuthorizationError( - "Authentication required", - code="missing_token", - ) - - _auth_logger.debug( - "Auth check: handler=%s, required=%s, principal_roles=%s, user_id=%s", - type(self).__name__, - auth_gate.role, - principal.roles, - principal.user_id, - ) - - 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 _QueryHandlerMeta(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_query_run_with_auth(cls, original_run) - cls.run = wrapped - - return cls - -class QueryHandler(Generic[C, R], metaclass=_QueryHandlerMeta): +class QueryHandler(Generic[C, R], metaclass=HandlerMeta): """Base class for query handlers. Subclasses are automatically dataclasses. Declare __auth__ to enforce role-based access: diff --git a/server/tests/unit/domain/shared/test_handler_gate_parity.py b/server/tests/unit/domain/shared/test_handler_gate_parity.py new file mode 100644 index 00000000..626d091a --- /dev/null +++ b/server/tests/unit/domain/shared/test_handler_gate_parity.py @@ -0,0 +1,145 @@ +"""The auth gate behaves identically on the write and read sides (#F2, 2026-08-16). + +The wrapper and metaclass live once in ``shared/handler.py``; these tests pin +the behavioral contract for BOTH facades so the gate semantics can never fork +between CommandHandler and QueryHandler again (they had drifted once: the query +copy debug-logged, the command copy didn't). +""" + +from uuid import uuid4 + +import pytest + +from osa.domain.auth.model.principal import Principal +from osa.domain.auth.model.role import Role +from osa.domain.auth.model.value import ProviderIdentity, UserId +from osa.domain.shared.authorization.gate import at_least, public, requires_scope +from osa.domain.shared.command import Command, CommandHandler +from osa.domain.shared.command import Result as CommandResult +from osa.domain.shared.error import AuthorizationError, ConfigurationError +from osa.domain.shared.query import Query, QueryHandler +from osa.domain.shared.query import Result as QueryResult + + +class Ping(Command): + pass + + +class PingQ(Query): + pass + + +class Pong(CommandResult): + ok: bool = True + + +def _principal( + *, roles: frozenset[Role] = frozenset(), scopes: frozenset[str] = frozenset() +) -> Principal: + return Principal( + user_id=UserId(uuid4()), + provider_identity=ProviderIdentity(provider="test", external_id="u1"), + roles=roles, + scopes=scopes, + ) + + +def _make_pair(gate, *, with_principal: bool): + """One CommandHandler and one QueryHandler with the same gate + body.""" + + if with_principal: + + class Cmd(CommandHandler[Ping, Pong]): + __auth__ = gate + principal: Principal | None + + async def run(self, cmd: Ping) -> Pong: + return Pong() + + class Qry(QueryHandler[PingQ, Pong]): + __auth__ = gate + principal: Principal | None + + async def run(self, cmd: PingQ) -> Pong: + return Pong() + + return Cmd, Qry + + class CmdNoP(CommandHandler[Ping, Pong]): + __auth__ = gate + + async def run(self, cmd: Ping) -> Pong: + return Pong() + + class QryNoP(QueryHandler[PingQ, Pong]): + __auth__ = gate + + async def run(self, cmd: PingQ) -> Pong: + return Pong() + + return CmdNoP, QryNoP + + +@pytest.mark.asyncio +class TestGateParity: + async def test_public_runs_without_principal_on_both(self): + cmd_cls, qry_cls = _make_pair(public(), with_principal=False) + assert (await cmd_cls().run(Ping())).ok + assert (await qry_cls().run(PingQ())).ok + + async def test_at_least_missing_token_on_both(self): + cmd_cls, qry_cls = _make_pair(at_least(Role.ADMIN), with_principal=True) + for handler, dto in ((cmd_cls(principal=None), Ping()), (qry_cls(principal=None), PingQ())): + with pytest.raises(AuthorizationError) as exc: + await handler.run(dto) + assert exc.value.code == "missing_token" + + async def test_at_least_access_denied_on_both(self): + cmd_cls, qry_cls = _make_pair(at_least(Role.ADMIN), with_principal=True) + weak = _principal(roles=frozenset({Role.DEPOSITOR})) + for handler, dto in ((cmd_cls(principal=weak), Ping()), (qry_cls(principal=weak), PingQ())): + with pytest.raises(AuthorizationError) as exc: + await handler.run(dto) + assert exc.value.code == "access_denied" + + async def test_at_least_admits_sufficient_role_on_both(self): + cmd_cls, qry_cls = _make_pair(at_least(Role.ADMIN), with_principal=True) + admin = _principal(roles=frozenset({Role.ADMIN})) + assert (await cmd_cls(principal=admin).run(Ping())).ok + assert (await qry_cls(principal=admin).run(PingQ())).ok + + async def test_requires_scope_admits_scope_or_admin_on_both(self): + gate = requires_scope("things:write") + cmd_cls, qry_cls = _make_pair(gate, with_principal=True) + scoped = _principal(scopes=frozenset({"things:write"})) + admin = _principal(roles=frozenset({Role.ADMIN})) + unscoped = _principal(scopes=frozenset({"other:read"})) + + assert (await cmd_cls(principal=scoped).run(Ping())).ok + assert (await qry_cls(principal=admin).run(PingQ())).ok + for handler, dto in ( + (cmd_cls(principal=unscoped), Ping()), + (qry_cls(principal=unscoped), PingQ()), + ): + with pytest.raises(AuthorizationError) as exc: + await handler.run(dto) + assert exc.value.code == "access_denied" + + async def test_missing_gate_is_a_configuration_error_on_both(self): + class NoGateCmd(CommandHandler[Ping, Pong]): + async def run(self, cmd: Ping) -> Pong: + return Pong() + + class NoGateQry(QueryHandler[PingQ, Pong]): + async def run(self, cmd: PingQ) -> Pong: + return Pong() + + with pytest.raises(ConfigurationError): + await NoGateCmd().run(Ping()) + with pytest.raises(ConfigurationError): + await NoGateQry().run(PingQ()) + + +def test_result_base_is_shared(): + """One Result class, re-exported by both facades — no forked DTO bases.""" + assert CommandResult is QueryResult From d8f7c7ed4ead6ae40c91bced1d824a323bab8a68 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Sun, 16 Aug 2026 13:08:34 +0100 Subject: [PATCH 2/2] fix: gate the events and validation routes through handlers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET /events and GET /validation/runs/{id} were the only two routes injecting a Service directly instead of a handler — so no __auth__ gate existed on either path and the startup validator (which walks handler factories) could not see the hole. /events returned full event payloads to anonymous callers; the validation route also owned the only business-logic decision tree in a route file. Both now go through gated query handlers, and the access level is a deliberate, boot-validated declaration instead of an accident: - ListEventsHandler (shared/event_log.py) owns the changefeed read — limit+1 look-ahead, cursor, payload shaping — with public() documented as the federation-surface choice. The order param is now Literal["asc","desc"]: garbage values 422 instead of silently meaning ascending. - GetValidationRunHandler (validation/query/) owns the status→shape rule (summary iff terminal, typed RunProgress iff running) and raises NotFoundError for the central mapper — the route's HTTPException and its untyped progress dict are gone. public() preserves today's anonymous polling; an ownership check is now a one-line gate change. Wire shapes preserved (response models unchanged; contract suite green). The 404 body for a missing validation run now uses the central error envelope rather than a bespoke detail string. Arch-survey 2026-08-16 finding F1 (fixes the two exceptions to the 'routes carry no auth logic' convention; F7's route decision tree rides along). --- .../osa/application/api/v1/routes/events.py | 50 +++++----- .../application/api/v1/routes/validation.py | 62 +++++-------- server/osa/domain/shared/event_log.py | 74 ++++++++++++++- .../validation/query/get_validation_run.py | 82 ++++++++++++++++ .../osa/domain/validation/util/di/provider.py | 4 + server/osa/infrastructure/event/di.py | 4 +- .../domain/shared/test_list_events_handler.py | 79 ++++++++++++++++ .../test_get_validation_run_handler.py | 93 +++++++++++++++++++ 8 files changed, 379 insertions(+), 69 deletions(-) create mode 100644 server/osa/domain/validation/query/get_validation_run.py create mode 100644 server/tests/unit/domain/shared/test_list_events_handler.py create mode 100644 server/tests/unit/domain/validation/test_get_validation_run_handler.py diff --git a/server/osa/application/api/v1/routes/events.py b/server/osa/application/api/v1/routes/events.py index 5d5b5944..3f44b998 100644 --- a/server/osa/application/api/v1/routes/events.py +++ b/server/osa/application/api/v1/routes/events.py @@ -1,6 +1,13 @@ -"""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 @@ -8,7 +15,7 @@ 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, ) diff --git a/server/osa/application/api/v1/routes/validation.py b/server/osa/application/api/v1/routes/validation.py index 031c6fd7..61db37c2 100644 --- a/server/osa/application/api/v1/routes/validation.py +++ b/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, ) diff --git a/server/osa/domain/shared/event_log.py b/server/osa/domain/shared/event_log.py index dc9eb337..658bbe2a 100644 --- a/server/osa/domain/shared/event_log.py +++ b/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, + ) diff --git a/server/osa/domain/validation/query/get_validation_run.py b/server/osa/domain/validation/query/get_validation_run.py new file mode 100644 index 00000000..4db81e80 --- /dev/null +++ b/server/osa/domain/validation/query/get_validation_run.py @@ -0,0 +1,82 @@ +"""GetValidationRun — status + results of one validation run. + +Owns the rule the route used to hard-code (arch-survey 2026-08-16 F1): which +fields are meaningful at which run status. Terminal runs carry the summary; +running runs carry progress; pending runs carry neither. A missing run raises +``NotFoundError`` for the central error mapper — no route-level HTTPException. +""" + +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field + +from osa.domain.shared.authorization.gate import public +from osa.domain.shared.error import NotFoundError +from osa.domain.shared.query import Query, QueryHandler, Result +from osa.domain.validation.model.hook_result import HookStatus +from osa.domain.validation.model.value import RunStatus +from osa.domain.validation.service.validation import ValidationService + +_TERMINAL = (RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.REJECTED) + + +class GetValidationRun(Query): + run_id: str + + +class HookResultEntry(Result): + hook_name: str + status: HookStatus + rejection_reason: str | None = None + error_message: str | None = None + duration_seconds: float + + +class RunProgress(BaseModel): + """Typed progress marker — present only while the run is RUNNING.""" + + status: Literal["running"] = "running" + + +class ValidationRunStatus(Result): + run_id: str + status: RunStatus + summary: HookStatus | None + progress: RunProgress | None + results: list[HookResultEntry] = Field(default_factory=list) + started_at: datetime | None + completed_at: datetime | None + + +class GetValidationRunHandler(QueryHandler[GetValidationRun, ValidationRunStatus]): + """``public()`` is deliberate: depositors poll their run status anonymously + today (the route had no gate at all). An ownership check is future work — + and now a one-line gate change away, boot-validated.""" + + __auth__ = public() + + service: ValidationService + + async def run(self, cmd: GetValidationRun) -> ValidationRunStatus: + run = await self.service.get_run(cmd.run_id) + if run is None: + raise NotFoundError(f"Validation run not found: {cmd.run_id}") + return ValidationRunStatus( + run_id=cmd.run_id, + status=run.status, + summary=run.summary if run.status in _TERMINAL else None, + progress=RunProgress() if run.status == RunStatus.RUNNING else None, + results=[ + HookResultEntry( + 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 + ], + started_at=run.started_at, + completed_at=run.completed_at, + ) diff --git a/server/osa/domain/validation/util/di/provider.py b/server/osa/domain/validation/util/di/provider.py index eeb5dba9..63ca45b9 100644 --- a/server/osa/domain/validation/util/di/provider.py +++ b/server/osa/domain/validation/util/di/provider.py @@ -6,6 +6,9 @@ from osa.domain.validation.command.create_release import CreateReleaseHandler from osa.domain.validation.command.set_live import SetLiveHandler from osa.domain.validation.query.get_hook_run import GetHookRunHandler +from osa.domain.validation.query.get_validation_run import ( + GetValidationRunHandler, +) from osa.domain.validation.query.get_hook_run_logs import GetHookRunLogsHandler from osa.domain.validation.query.get_release import GetReleaseHandler from osa.domain.validation.query.list_hooks import ListHooksHandler @@ -35,6 +38,7 @@ class ValidationProvider(Provider): # Hook-run provenance + logs read handlers (#147). get_hook_run_handler = provide(GetHookRunHandler, scope=Scope.UOW) + get_validation_run_handler = provide(GetValidationRunHandler, scope=Scope.UOW) get_hook_run_logs_handler = provide(GetHookRunLogsHandler, scope=Scope.UOW) @provide(scope=Scope.UOW) diff --git a/server/osa/infrastructure/event/di.py b/server/osa/infrastructure/event/di.py index 7f192f61..1eddc1d5 100644 --- a/server/osa/infrastructure/event/di.py +++ b/server/osa/infrastructure/event/di.py @@ -12,7 +12,7 @@ from osa.application.workflow.process_submission import ProcessSubmission from osa.config import Config from osa.domain.shared.event import EventHandler -from osa.domain.shared.event_log import EventLog +from osa.domain.shared.event_log import EventLog, ListEventsHandler from osa.domain.shared.model.subscription_registry import SubscriptionRegistry from osa.domain.shared.outbox import Outbox from osa.domain.shared.port.event_repository import EventRepository @@ -103,6 +103,8 @@ def get_outbox(self, repo: EventRepository, registry: SubscriptionRegistry) -> O def get_event_log(self, repo: EventRepository) -> EventLog: return EventLog(repo) + list_events_handler = provide(ListEventsHandler, scope=Scope.UOW) + @provide(scope=Scope.APP) def get_handler_types(self) -> HandlerTypes: """Return all handler types (core + extra) for WorkerPool registration.""" diff --git a/server/tests/unit/domain/shared/test_list_events_handler.py b/server/tests/unit/domain/shared/test_list_events_handler.py new file mode 100644 index 00000000..1f745475 --- /dev/null +++ b/server/tests/unit/domain/shared/test_list_events_handler.py @@ -0,0 +1,79 @@ +"""ListEvents query handler (arch-survey 2026-08-16 F1). + +The /events route previously injected EventLog directly — full event payloads +to anonymous callers with no gate anywhere. The handler now owns the changefeed +read (limit+1 look-ahead, cursor, payload shaping) behind an explicit, +boot-validated gate. Public is the deliberate choice: the changefeed is the +federation surface (CLAUDE.md API §9) — the gate documents that decision. +""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest + +from osa.domain.shared.authorization.gate import Public +from osa.domain.shared.event import Event, EventId +from osa.domain.shared.event_log import ListEvents, ListEventsHandler + + +class SomethingHappened(Event): + detail: str + + +def _event(detail: str) -> SomethingHappened: + return SomethingHappened( + id=EventId(uuid4()), created_at=datetime(2026, 1, 1, tzinfo=UTC), detail=detail + ) + + +def _handler(events: list[Event]) -> ListEventsHandler: + log = AsyncMock() + log.list_events.return_value = events + return ListEventsHandler(event_log=log) + + +class TestGate: + def test_gate_is_explicitly_public(self): + assert isinstance(ListEventsHandler.__auth__, Public) + + +@pytest.mark.asyncio +class TestChangefeedPage: + async def test_look_ahead_sets_has_more_and_trims(self): + events = [_event(f"e{i}") for i in range(3)] + handler = _handler(events) + page = await handler.run(ListEvents(limit=2)) + # Service asked for limit+1; surplus row trimmed, has_more set. + handler.event_log.list_events.assert_awaited_once() + assert handler.event_log.list_events.await_args.kwargs["limit"] == 3 + assert len(page.events) == 2 + assert page.has_more is True + assert page.cursor == str(events[1].id) + + async def test_last_page_has_no_more(self): + events = [_event("only")] + page = await _handler(events).run(ListEvents(limit=2)) + assert len(page.events) == 1 + assert page.has_more is False + assert page.cursor == str(events[0].id) + + async def test_empty_page_has_no_cursor(self): + page = await _handler([]).run(ListEvents(limit=2)) + assert page.events == [] and page.cursor is None and page.has_more is False + + async def test_payload_shaping(self): + events = [_event("hello")] + page = await _handler(events).run(ListEvents(limit=5)) + entry = page.events[0] + assert entry.type == "SomethingHappened" + assert entry.data == {"detail": "hello"} # id/created_at excluded from payload + assert entry.id == events[0].id + + async def test_order_and_type_filters_forwarded(self): + handler = _handler([]) + await handler.run(ListEvents(limit=5, types=["RecordPublished"], order="desc")) + kwargs = handler.event_log.list_events.await_args.kwargs + assert kwargs["event_types"] == ["RecordPublished"] + assert kwargs["newest_first"] is True diff --git a/server/tests/unit/domain/validation/test_get_validation_run_handler.py b/server/tests/unit/domain/validation/test_get_validation_run_handler.py new file mode 100644 index 00000000..0a4baa80 --- /dev/null +++ b/server/tests/unit/domain/validation/test_get_validation_run_handler.py @@ -0,0 +1,93 @@ +"""GetValidationRun query handler (arch-survey 2026-08-16 F1). + +The /validation/runs/{id} route previously injected ValidationService directly +— no __auth__ gate existed on the path, and the route owned the only +business-logic decision tree in a route file (which fields are meaningful at +which run status). Both now live here: the handler carries an explicit gate +and owns the status→shape rule; the route is thin. +""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock + +import pytest + +from osa.domain.shared.authorization.gate import Public +from osa.domain.shared.error import NotFoundError +from osa.domain.shared.model.srn import Domain, LocalId, ValidationRunSRN +from osa.domain.validation.model.entity import ValidationRun +from osa.domain.validation.model.hook_result import HookResult, HookStatus +from osa.domain.validation.model.value import RunStatus +from osa.domain.validation.query.get_validation_run import ( + GetValidationRun, + GetValidationRunHandler, +) + + +def _run(status: RunStatus, results: list[HookResult] | None = None) -> ValidationRun: + return ValidationRun( + srn=ValidationRunSRN(domain=Domain("localhost"), id=LocalId("run-000000001"), version=None), + status=status, + results=results or [], + started_at=datetime(2026, 1, 1, tzinfo=UTC), + completed_at=None, + ) + + +def _passed_result() -> HookResult: + from osa.domain.shared.model.hook import HookName + + return HookResult(hook_name=HookName("quality"), status=HookStatus.PASSED, duration_seconds=1.5) + + +def _handler(run: ValidationRun | None) -> GetValidationRunHandler: + service = AsyncMock() + service.get_run.return_value = run + return GetValidationRunHandler(service=service) + + +class TestGate: + def test_gate_is_explicitly_public(self): + """The changed contract of F1: an explicit, boot-validated gate exists. + Public is the deliberate, documented choice (depositors poll their run + status anonymously today) — tightening it later is a one-line edit.""" + assert isinstance(GetValidationRunHandler.__auth__, Public) + + +@pytest.mark.asyncio +class TestStatusShaping: + async def test_terminal_statuses_carry_summary_and_no_progress(self): + for status in (RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.REJECTED): + result = await _handler(_run(status, [_passed_result()])).run( + GetValidationRun(run_id="run-000000001") + ) + assert result.summary == HookStatus.PASSED, status + assert result.progress is None, status + + async def test_running_carries_progress_and_no_summary(self): + result = await _handler(_run(RunStatus.RUNNING)).run( + GetValidationRun(run_id="run-000000001") + ) + assert result.summary is None + assert result.progress is not None and result.progress.status == "running" + + async def test_pending_carries_neither(self): + result = await _handler(_run(RunStatus.PENDING)).run( + GetValidationRun(run_id="run-000000001") + ) + assert result.summary is None and result.progress is None + + async def test_results_are_mapped(self): + result = await _handler(_run(RunStatus.COMPLETED, [_passed_result()])).run( + GetValidationRun(run_id="run-000000001") + ) + assert len(result.results) == 1 + entry = result.results[0] + assert entry.hook_name == "quality" + assert entry.status == HookStatus.PASSED + assert entry.duration_seconds == 1.5 + + async def test_missing_run_raises_not_found(self): + """Central error mapping, not a route-level HTTPException.""" + with pytest.raises(NotFoundError): + await _handler(None).run(GetValidationRun(run_id="nope-00000001"))