deer-flow/backend/tests/test_extension_gateway_wiring.py

"""Gateway binding tests for app-scoped extension contributions."""

from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager
from types import SimpleNamespace

import pytest

from deerflow.extensions.registry import ExtensionRegistry


class _Service:
    def __init__(
        self,
        name: str,
        events: list[str],
        *,
        fail_start: bool = False,
        fail_stop: bool = False,
    ) -> None:
        self.name = name
        self.events = events
        self.fail_start = fail_start
        self.fail_stop = fail_stop
        self.deps = None

    async def start(self, deps) -> None:
        self.events.append(f"start:{self.name}")
        self.deps = deps
        if self.fail_start:
            raise RuntimeError(f"{self.name} failed")

    async def stop(self) -> None:
        self.events.append(f"stop:{self.name}")
        if self.fail_stop:
            raise RuntimeError(f"{self.name} failed")


@pytest.mark.asyncio
async def test_services_start_in_order_with_narrow_deps_and_fail_open():
    from deerflow.extensions.gateway import start_services

    events: list[str] = []
    first = _Service("first", events, fail_start=True)
    second = _Service("second", events)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)
    extensions = registry.build()
    session_factory = object()
    config = SimpleNamespace(
        token_budget=SimpleNamespace(enabled=False, max_tokens=999),
        subagents=SimpleNamespace(max_total_per_run=4),
    )

    diagnostics = await start_services(extensions, config, session_factory)

    assert events == ["start:first", "start:second"]
    assert first.deps is second.deps
    assert second.deps.app_store is extensions.app_store
    assert second.deps.session_factory is session_factory
    assert second.deps.policy.token_budget_enabled is False
    assert second.deps.policy.max_total_tokens is None
    assert second.deps.policy.max_subagents_per_run == 4
    assert [(diagnostic.source, diagnostic.level) for diagnostic in diagnostics] == [("first:install", "error")]


@pytest.mark.asyncio
async def test_service_originated_cancelled_error_does_not_abort_start_batch():
    from deerflow.extensions.gateway import start_services

    events: list[str] = []

    class _CancelsItself(_Service):
        async def start(self, deps) -> None:
            self.events.append(f"start:{self.name}")
            raise asyncio.CancelledError()

    first = _CancelsItself("first", events)
    second = _Service("second", events)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)

    diagnostics = await start_services(registry.build(), SimpleNamespace(), None)

    assert events == ["start:first", "start:second"]
    assert len(diagnostics) == 1
    assert diagnostics[0].source == "first:install"
    assert "CancelledError" in diagnostics[0].message


@pytest.mark.asyncio
async def test_services_stop_in_reverse_order_and_fail_open():
    from deerflow.extensions.gateway import stop_services

    events: list[str] = []
    first = _Service("first", events)
    second = _Service("second", events, fail_stop=True)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)

    diagnostics = await stop_services(registry.build())

    assert events == ["stop:second", "stop:first"]
    assert [(diagnostic.source, diagnostic.level) for diagnostic in diagnostics] == [("second:install", "error")]


@pytest.mark.asyncio
async def test_service_originated_cancelled_error_does_not_abort_stop_batch():
    from deerflow.extensions.gateway import stop_services

    events: list[str] = []

    class _CancelsItself(_Service):
        async def stop(self) -> None:
            self.events.append(f"stop:{self.name}")
            raise asyncio.CancelledError()

    first = _Service("first", events)
    second = _CancelsItself("second", events)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)

    diagnostics = await stop_services(registry.build())

    assert events == ["stop:second", "stop:first"]
    assert len(diagnostics) == 1
    assert diagnostics[0].source == "second:install"
    assert "CancelledError" in diagnostics[0].message


@pytest.mark.asyncio
async def test_each_service_stop_has_its_own_timeout_budget():
    from deerflow.extensions.gateway import stop_services

    events: list[str] = []

    class _HangingService(_Service):
        async def stop(self) -> None:
            self.events.append(f"stop:{self.name}")
            await asyncio.Event().wait()

    first = _Service("first", events)
    second = _HangingService("second", events)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)

    diagnostics = await stop_services(registry.build(), timeout_seconds=0.01)

    assert events == ["stop:second", "stop:first"]
    assert len(diagnostics) == 1
    assert diagnostics[0].source == "second:install"
    assert "timed out" in diagnostics[0].message


@pytest.mark.asyncio
async def test_service_originated_timeout_error_is_reported_as_failure_not_budget_expiry():
    from deerflow.extensions.gateway import stop_services

    events: list[str] = []

    class _RaisesTimeout(_Service):
        async def stop(self) -> None:
            self.events.append(f"stop:{self.name}")
            raise TimeoutError("extension deadline")

    first = _Service("first", events)
    second = _RaisesTimeout("second", events)
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.service(first)
    with registry.attributed_to("second:install"):
        registry.service(second)

    diagnostics = await stop_services(registry.build(), timeout_seconds=1.0)

    assert events == ["stop:second", "stop:first"]
    assert diagnostics[0].source == "second:install"
    assert "failed" in diagnostics[0].message
    assert "timed out" not in diagnostics[0].message


@pytest.mark.parametrize(
    ("owner_path", "owner_protocol", "candidate_path", "candidate_protocol", "rejected"),
    [
        ("/exact", "GET", "/exact", "GET", True),
        ("/items/{item_id}", "GET", "/items/{id}", "GET", True),
        ("/items/{item_id}", "GET", "/items/new", "GET", True),
        ("/items/{item_id}", "GET", "/items/prefix-{id}", "GET", True),
        ("/pre{tenant}", "GET", "/prefoo{id}", "GET", True),
        ("/items/new", "GET", "/items/{id}", "GET", False),
        ("/records/{value}", "GET", "/records/{id:int}", "GET", True),
        ("/records/{value:int}", "GET", "/records/0{id:int}", "GET", True),
        ("/records/{value:int}", "GET", "/records/new", "GET", False),
        ("/files/{rest:path}", "GET", "/files/{id:int}", "GET", True),
        ("/files/{rest:path}", "GET", "/files/{id}", "GET", False),
        ("/x/{rest:path}/tail", "GET", "/x/a/{id}/tail", "GET", False),
        ("/items/{item_id}", "GET", "/items/{id}", "POST", False),
        ("/live/{item_id}", "WS", "/live/{id}", "GET", False),
    ],
    ids=[
        "exact",
        "renamed-parameter",
        "dynamic-shadows-static",
        "dynamic-shadows-compound",
        "compound-trailing-str-shadows-narrower-compound",
        "static-does-not-shadow-dynamic",
        "str-covers-int",
        "int-shadows-digit-compound",
        "int-does-not-cover-static",
        "path-covers-descendant",
        "path-does-not-cover-newline-capable-str",
        "nonterminal-path-does-not-cover-newline-capable-str",
        "disjoint-http-methods",
        "websocket-does-not-shadow-http",
    ],
)
def test_router_conflicts_follow_starlette_dispatch_order(
    owner_path,
    owner_protocol,
    candidate_path,
    candidate_protocol,
    rejected,
):
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    app = FastAPI()
    if owner_protocol == "WS":
        app.add_api_websocket_route(owner_path, endpoint)
    else:
        app.add_api_route(owner_path, endpoint, methods=[owner_protocol])

    router = APIRouter()
    if candidate_protocol == "WS":
        router.add_api_websocket_route(candidate_path, endpoint)
    else:
        router.add_api_route(candidate_path, endpoint, methods=[candidate_protocol])
    registry = ExtensionRegistry()
    with registry.attributed_to("candidate:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert bool(diagnostics) is rejected
    if rejected:
        assert diagnostics[0].source == "candidate:install"
        assert "host" in diagnostics[0].message
        assert candidate_path in diagnostics[0].message
    else:
        assert any(getattr(route, "path", None) == candidate_path for route in app.routes)


@pytest.mark.parametrize("convertor_name", ["flip", "int"])
def test_re_registered_convertor_does_not_create_a_false_shadow(
    monkeypatch,
    convertor_name,
):
    from fastapi import APIRouter, FastAPI
    from starlette.convertors import CONVERTOR_TYPES, Convertor
    from starlette.routing import Match

    from deerflow.extensions.gateway import include_contributed_routers

    class DigitsConvertor(Convertor[str]):
        regex = "[0-9]+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    class LettersConvertor(Convertor[str]):
        regex = "[A-Z]+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    async def endpoint(value: str):
        return {"value": value}

    monkeypatch.setitem(CONVERTOR_TYPES, convertor_name, DigitsConvertor())
    route_path = f"/owned/{{value:{convertor_name}}}"
    app = FastAPI()
    app.add_api_route(route_path, endpoint, methods=["GET"])
    owner = app.routes[-1]

    monkeypatch.setitem(CONVERTOR_TYPES, convertor_name, LettersConvertor())
    router = APIRouter()
    router.add_api_route(route_path, endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("candidate:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    candidate = app.routes[-1]
    scope = {
        "type": "http",
        "path": "/owned/A",
        "method": "GET",
        "root_path": "",
    }
    assert owner.matches(scope)[0] is Match.NONE
    assert candidate.matches(scope)[0] is Match.FULL


def test_router_claim_uses_converter_semantics_at_include_time(monkeypatch):
    from fastapi import APIRouter, FastAPI
    from starlette.convertors import CONVERTOR_TYPES, Convertor
    from starlette.routing import Match

    from deerflow.extensions.gateway import include_contributed_routers

    class DigitsConvertor(Convertor[str]):
        regex = "[0-9]+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    class LettersConvertor(Convertor[str]):
        regex = "[A-Z]+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    async def endpoint(value: str):
        return {"value": value}

    monkeypatch.setitem(CONVERTOR_TYPES, "flip", DigitsConvertor())
    route_path = "/owned/{value:flip}"
    app = FastAPI()
    app.add_api_route(route_path, endpoint, methods=["GET"])
    owner = app.routes[-1]
    router = APIRouter()
    router.add_api_route(route_path, endpoint, methods=["GET"])

    monkeypatch.setitem(CONVERTOR_TYPES, "flip", LettersConvertor())
    registry = ExtensionRegistry()
    with registry.attributed_to("candidate:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    candidate = app.routes[-1]
    scope = {
        "type": "http",
        "path": "/owned/A",
        "method": "GET",
        "root_path": "",
    }
    assert owner.matches(scope)[0] is Match.NONE
    assert candidate.matches(scope)[0] is Match.FULL


def test_recompiled_converter_cannot_enter_a_public_namespace(monkeypatch):
    from fastapi import APIRouter, FastAPI
    from starlette.convertors import CONVERTOR_TYPES, Convertor

    from deerflow.extensions.gateway import include_contributed_routers

    class PublicPathConvertor(Convertor[str]):
        regex = r"webhooks/.+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    async def endpoint(value: str):
        return {"value": value}

    router = APIRouter()
    router.add_api_route("/api/{value:int}", endpoint, methods=["GET"])
    monkeypatch.setitem(CONVERTOR_TYPES, "int", PublicPathConvertor())
    registry = ExtensionRegistry()
    with registry.attributed_to("candidate:install"):
        registry.routers((router,))

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["candidate:install"]
    assert "public namespace" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == "/api/{value:int}" for route in app.routes)


def test_host_mount_claims_descendant_http_paths():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    app = FastAPI()
    app.mount("/assets", FastAPI())
    router = APIRouter()

    @router.get("/assets/{name:int}")
    async def asset(name: int):
        return {"name": name}

    registry = ExtensionRegistry()
    with registry.attributed_to("assets:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert len(diagnostics) == 1
    assert diagnostics[0].source == "assets:install"
    assert "host" in diagnostics[0].message


def test_dynamic_host_mount_claims_matching_descendants():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    app = FastAPI()
    app.mount("/pre{tenant}", FastAPI())
    router = APIRouter()

    @router.get("/prefoo/{item_id:int}")
    async def item(item_id: int):
        return {"item_id": item_id}

    registry = ExtensionRegistry()
    with registry.attributed_to("mount:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["mount:install"]
    assert "host" in diagnostics[0].message


@pytest.mark.parametrize(
    ("mount_path", "candidate_path", "witness"),
    [
        ("/assets", "/assets/{name}", "/assets/a\nb"),
        ("/pre{tenant}", "/prefoo{id}/{child}", "/prefoo1/a\nb"),
    ],
)
def test_host_mount_does_not_claim_newline_capable_str_descendants(
    mount_path,
    candidate_path,
    witness,
):
    from fastapi import APIRouter, FastAPI
    from starlette.routing import Match, Mount

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    app = FastAPI()
    app.mount(mount_path, FastAPI())
    host_mount = next(route for route in app.routes if isinstance(route, Mount) and route.path == mount_path)
    router = APIRouter()
    router.add_api_route(candidate_path, endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("mount:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    candidate_route = next(route for route in app.routes if getattr(route, "path", None) == candidate_path)
    scope = {
        "type": "http",
        "path": witness,
        "method": "GET",
        "root_path": "",
    }
    assert host_mount.matches(scope)[0] is Match.NONE
    assert candidate_route.matches(scope)[0] is Match.FULL


def test_contributed_websocket_route_is_rejected_until_host_auth_wraps_it():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def websocket_endpoint(websocket):
        await websocket.close()

    router = APIRouter()
    router.add_api_websocket_route("/extension-ws", websocket_endpoint)
    registry = ExtensionRegistry()
    with registry.attributed_to("websocket:install"):
        registry.routers((router,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["websocket:install"]
    assert "WebSocket" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == "/extension-ws" for route in app.routes)


@pytest.mark.parametrize(
    "path",
    [
        "/health-extension",
        "/docs-private",
        "/redoc-private",
        "/api/webhooks/extension",
        "/api/{rest:path}",
        "/api/{section}/extension",
    ],
)
def test_extension_routes_cannot_enter_host_public_namespaces(path):
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    router = APIRouter()
    router.add_api_route(path, endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("public:install"):
        registry.routers((router,))

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["public:install"]
    assert "public namespace" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == path for route in app.routes)


@pytest.mark.parametrize(
    "path",
    [
        "/api",
        "/heal",
        "/api/{item_id:int}",
        "/api/{item_id}",
    ],
)
def test_extension_routes_that_cannot_enter_a_public_namespace_are_allowed(path):
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    router = APIRouter()
    router.add_api_route(path, endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("private:install"):
        registry.routers((router,))

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    assert any(getattr(route, "path", None) == path for route in app.routes)


def test_unknown_convertor_near_a_public_namespace_fails_closed(monkeypatch):
    from fastapi import APIRouter, FastAPI
    from starlette.convertors import CONVERTOR_TYPES, Convertor

    from deerflow.extensions.gateway import include_contributed_routers

    class UppercaseConvertor(Convertor[str]):
        regex = "[A-Z]+"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    monkeypatch.setitem(CONVERTOR_TYPES, "uppercase", UppercaseConvertor())
    router = APIRouter()
    router.add_api_route(
        "/api/{value:uppercase}",
        lambda: {"ok": True},
        methods=["GET"],
    )
    registry = ExtensionRegistry()
    with registry.attributed_to("custom-public:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(
        FastAPI(docs_url=None, redoc_url=None, openapi_url=None),
        registry.build(),
    )

    assert [diagnostic.source for diagnostic in diagnostics] == ["custom-public:install"]
    assert "public namespace" in diagnostics[0].message


def test_private_custom_convertor_with_named_backreference_is_allowed(monkeypatch):
    from fastapi import APIRouter, FastAPI
    from starlette.convertors import CONVERTOR_TYPES, Convertor

    from deerflow.extensions.gateway import include_contributed_routers

    class DoubledLetterConvertor(Convertor[str]):
        regex = r"(?P<char>[A-Z])(?P=char)"

        def convert(self, value: str) -> str:
            return value

        def to_string(self, value: str) -> str:
            return value

    monkeypatch.setitem(CONVERTOR_TYPES, "doubled", DoubledLetterConvertor())
    router = APIRouter()
    router.add_api_route(
        "/private/{value:doubled}",
        lambda value: {"value": value},
        methods=["GET"],
    )
    registry = ExtensionRegistry()
    with registry.attributed_to("custom-private:install"):
        registry.routers((router,))

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    assert any(getattr(route, "path", None) == "/private/{value:doubled}" for route in app.routes)


def test_extension_public_paths_track_auth_middleware_public_paths():
    from app.gateway.auth_middleware import (
        _PUBLIC_EXACT_PATHS,
        _PUBLIC_PATH_PREFIXES,
        _is_public,
    )
    from deerflow.extensions.gateway import (
        _HOST_PUBLIC_EXACT_PATHS,
        _HOST_PUBLIC_PATH_PREFIXES,
    )

    assert _HOST_PUBLIC_PATH_PREFIXES == _PUBLIC_PATH_PREFIXES
    assert _HOST_PUBLIC_EXACT_PATHS == _PUBLIC_EXACT_PATHS
    assert all(_is_public(f"{path}//") for path in _HOST_PUBLIC_EXACT_PATHS)


@pytest.mark.parametrize(
    ("host_path", "host_method", "candidate_path", "candidate_method"),
    [
        (
            "/api/v1/auth/login/local",
            "POST",
            "/api/v1/auth/login/local",
            "GET",
        ),
        (
            "/api/v1/auth/login/local",
            "POST",
            "/api/v1/auth/login/local/",
            "GET",
        ),
        (
            "/api/v1/auth/login/local",
            "POST",
            "/api/v1/auth/login/local//",
            "GET",
        ),
        ("/api/v1/auth/me", "GET", "/api/v1/auth/me", "POST"),
        ("/api/v1/auth/me", "GET", "/api/v1/auth/me/", "POST"),
    ],
)
def test_extension_routes_cannot_claim_reserved_exact_paths_with_a_disjoint_method(
    host_path,
    host_method,
    candidate_path,
    candidate_method,
):
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    app = FastAPI()
    app.add_api_route(host_path, endpoint, methods=[host_method])
    router = APIRouter()
    router.add_api_route(candidate_path, endpoint, methods=[candidate_method])
    registry = ExtensionRegistry()
    with registry.attributed_to("public-exact:install"):
        registry.routers((router,))

    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["public-exact:install"]
    assert "reserved" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == candidate_path and getattr(route, "methods", set()) == {candidate_method} for route in app.routes)


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/auth/me",
        "/api/v1/auth/me/",
        "/api/v1/auth/me//",
    ],
)
def test_extension_csrf_reserved_exact_paths_track_csrf_exemption(
    monkeypatch,
    path,
):
    from starlette.requests import Request

    from app.gateway import csrf_middleware
    from deerflow.extensions.gateway import (
        _CSRF_STATE_CHANGING_METHODS,
        _HOST_CSRF_EXEMPT_EXACT_PATHS,
        _HOST_PUBLIC_EXACT_PATHS,
    )

    monkeypatch.setattr(csrf_middleware, "is_auth_disabled", lambda: False)
    request = Request(
        {
            "type": "http",
            "method": "POST",
            "scheme": "http",
            "path": path,
            "raw_path": path.encode(),
            "query_string": b"",
            "headers": [],
            "server": ("testserver", 80),
        }
    )

    assert path.rstrip("/") in _HOST_CSRF_EXEMPT_EXACT_PATHS
    assert _HOST_CSRF_EXEMPT_EXACT_PATHS == csrf_middleware._CSRF_EXEMPT_EXACT_PATHS
    assert _CSRF_STATE_CHANGING_METHODS == csrf_middleware._CSRF_STATE_CHANGING_METHODS
    assert csrf_middleware._AUTH_EXEMPT_PATHS <= _HOST_PUBLIC_EXACT_PATHS
    assert csrf_middleware.should_check_csrf(request) is False


def test_safe_method_at_csrf_exempt_exact_path_is_not_reserved():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    router = APIRouter()
    router.add_api_route(
        "/api/v1/auth/me",
        lambda: {"ok": True},
        methods=["GET"],
    )
    registry = ExtensionRegistry()
    with registry.attributed_to("safe-csrf:install"):
        registry.routers((router,))

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    diagnostics = include_contributed_routers(app, registry.build())

    assert diagnostics == []
    assert any(getattr(route, "path", None) == "/api/v1/auth/me" for route in app.routes)


def test_router_with_one_conflict_is_rejected_atomically_and_names_first_owner():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    first = APIRouter()
    first.add_api_route("/shared", endpoint, methods=["GET"])
    second = APIRouter()
    second.add_api_route("/would-have-been-reachable", endpoint, methods=["GET"])
    second.add_api_route("/shared", endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("first:install"):
        registry.routers((first,))
    with registry.attributed_to("second:install"):
        registry.routers((second,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())
    paths = [getattr(route, "path", None) for route in app.routes]

    assert "/shared" in paths
    assert "/would-have-been-reachable" not in paths
    assert len(diagnostics) == 1
    assert diagnostics[0].source == "second:install"
    assert "first:install" in diagnostics[0].message


def test_router_is_rejected_when_its_own_earlier_route_shadows_a_later_one():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    router = APIRouter()
    router.add_api_route("/same/{value}", endpoint, methods=["GET"])
    router.add_api_route("/same/fixed", endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("self-shadow:install"):
        registry.routers((router,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["self-shadow:install"]
    assert "self-shadow:install" in diagnostics[0].message
    assert not any(getattr(route, "path", "").startswith("/same/") for route in app.routes)


def test_contributed_mount_is_rejected_but_does_not_starve_later_router():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    bad = APIRouter()
    bad.mount("/nested", FastAPI())
    good = APIRouter()

    @good.get("/extension-good")
    async def extension_good():
        return {"ok": True}

    registry = ExtensionRegistry()
    with registry.attributed_to("bad:install"):
        registry.routers((bad,))
    with registry.attributed_to("good:install"):
        registry.routers((good,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["bad:install"]
    assert "Mount" in diagnostics[0].message
    assert any(getattr(route, "path", None) == "/extension-good" for route in app.routes)


def test_router_with_unsupported_route_item_is_rejected_atomically():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    router = APIRouter()
    router.add_api_route("/otherwise-valid", endpoint, methods=["GET"])
    router.routes.append(object())
    registry = ExtensionRegistry()
    with registry.attributed_to("unsupported:install"):
        registry.routers((router,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["unsupported:install"]
    assert "unsupported route" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == "/otherwise-valid" for route in app.routes)


def test_include_router_failure_rolls_back_partial_routes_before_continuing():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    broken = APIRouter()
    broken.add_api_route("/partial", endpoint, methods=["GET"])
    broken.add_api_route("/explodes", endpoint, methods=["GET"])
    broken.routes[-1].endpoint = None
    later = APIRouter()
    later.add_api_route("/partial", endpoint, methods=["GET"])

    registry = ExtensionRegistry()
    with registry.attributed_to("broken:install"):
        registry.routers((broken,))
    with registry.attributed_to("later:install"):
        registry.routers((later,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())
    partial_routes = [route for route in app.routes if getattr(route, "path", None) == "/partial"]

    assert [diagnostic.source for diagnostic in diagnostics] == ["broken:install"]
    assert len(partial_routes) == 1


def test_router_lifecycle_hooks_are_rejected_in_favor_of_extension_service():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    async def endpoint():
        return {"ok": True}

    async def startup_hook():
        raise RuntimeError("must never be installed")

    bad = APIRouter()
    bad.add_api_route("/has-lifecycle", endpoint, methods=["GET"])
    bad.add_event_handler("startup", startup_hook)
    good = APIRouter()
    good.add_api_route("/after-lifecycle", endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("lifecycle:install"):
        registry.routers((bad,))
    with registry.attributed_to("good:install"):
        registry.routers((good,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())
    paths = [getattr(route, "path", None) for route in app.routes]

    assert [diagnostic.source for diagnostic in diagnostics] == ["lifecycle:install"]
    assert "ExtensionService" in diagnostics[0].message
    assert "/has-lifecycle" not in paths
    assert "/after-lifecycle" in paths
    assert startup_hook not in app.router.on_startup


def test_router_custom_lifespan_is_rejected_in_favor_of_extension_service():
    from fastapi import APIRouter, FastAPI

    from deerflow.extensions.gateway import include_contributed_routers

    @asynccontextmanager
    async def lifespan(_app):
        yield

    async def endpoint():
        return {"ok": True}

    router = APIRouter(lifespan=lifespan)
    router.add_api_route("/custom-lifespan", endpoint, methods=["GET"])
    registry = ExtensionRegistry()
    with registry.attributed_to("lifespan:install"):
        registry.routers((router,))

    app = FastAPI()
    diagnostics = include_contributed_routers(app, registry.build())

    assert [diagnostic.source for diagnostic in diagnostics] == ["lifespan:install"]
    assert "ExtensionService" in diagnostics[0].message
    assert not any(getattr(route, "path", None) == "/custom-lifespan" for route in app.routes)