Expand ASGI adapter lifecycle coverage: close on failed startup, tests, and docs

docs/frameworks.md

@@ -69,8 +69,12 @@ fallback is intentional.
 - On `lifespan.shutdown.complete`, the adapter awaits `bus.close()` when the owned
   bus exposes a synchronous or asynchronous close method.
 - Cleanup completes before shutdown success is forwarded to the ASGI server.
-- Startup failures do not claim that shutdown cleanup occurred; applications should
-  manage resources created outside the adapter in their own lifespan handler.
+- A failed startup (either `lifespan.startup.failed` or an exception escaping the
+  lifespan application) also closes an adapter-owned bus. Caller-owned buses remain
+  untouched in every failure path.
+- Cleanup is idempotent: repeated startup/shutdown cycles against the same middleware
+  instance do not close a bus more than once. Create a new application/middleware
+  instance to obtain a fresh adapter-owned bus after shutdown.
 
 The local `EventBus` has no resources to close. The generic close behavior exists
 for application bus subclasses that coordinate transports, stores, or plugins.
@@ -81,3 +85,18 @@ The bus itself is application-scoped. Request state isolates access paths, not b
 registrations. Use separate application instances or a custom `bus_factory` when
 tests or tenants require distinct registration state. `state_key=` supports
 coexistence with another request-state convention.
+
+Each middleware instance creates its own bus, so concurrently running application
+instances are isolated. This is process-local isolation only: pre-fork and
+multi-worker deployments create one adapter-owned bus per worker. Eventful does not
+coordinate registrations or delivery between workers; inject a caller-managed
+transport-backed bus when cross-process behavior is required. Do not share one
+adapter-owned middleware instance between event loops.
+
+## Supported versions
+
+The adapters support FastAPI 0.110 or newer and Starlette 0.37 or newer on Eventful's
+supported Python versions. The compatibility floor is exercised through the public
+ASGI and dependency APIs; CI's normal dependency resolution also exercises current
+framework releases. Major-version compatibility is not promised until separately
+validated.

docs/work-register.md

@@ -4,7 +4,7 @@ All intentional incompleteness must use an annotation listed in `docs/annotation
 
 | Module | Status | Deferred work |
 | --- | --- | --- |
-| `eventful.adapters` | provisional ASGI lifecycle | Validate multi-worker ownership and framework-version compatibility (Issue: Forebase/Eventful#1). |
+| `eventful.adapters` | validated in-process ASGI lifecycle | Multi-worker delivery remains deployment-managed; major framework versions require separate validation (Issue: Forebase/Eventful#1). |
 | `eventful.transports.redis` | provisional Pub/Sub | Validate operational reconnect and load behavior; Redis Streams durability remains deferred (Issue: Forebase/Eventful#2). |
 | `eventful.persistence.postgres_persistence` | provisional durable store | Validate migration upgrades, retention, replication, and operational load behavior (Issue: Forebase/Eventful#3). |
 | `eventful.contracts` | provisional, reference-validated | Validate async lifecycle, delivery, and durability semantics against real Redis/PostgreSQL integrations (Issue: Forebase/Eventful#4). |

src/eventful/adapters/starlette.py

@@ -50,7 +50,7 @@ def __init__(
         self._close_lock = asyncio.Lock()
 
     async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
-        """Attach state and wait for application shutdown before closing resources."""
+        """Attach state and close owned resources when a lifespan terminates."""
         if scope["type"] in {"http", "websocket"}:
             scope.setdefault("state", {})[self.state_key] = self.bus
 
@@ -59,12 +59,21 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None
             return
 
         async def lifespan_send(message: dict[str, Any]) -> None:
-            """Close owned resources before reporting successful shutdown."""
-            if message["type"] == "lifespan.shutdown.complete":
+            """Close resources before reporting shutdown or failed startup."""
+            if message["type"] in {
+                "lifespan.startup.failed",
+                "lifespan.shutdown.complete",
+            }:
                 await self.close()
             await send(message)
 
-        await self.app(scope, receive, lifespan_send)
+        try:
+            await self.app(scope, receive, lifespan_send)
+        except BaseException:
+            # A lifespan exception may prevent the application from sending either
+            # terminal message. Do not strand a bus that this adapter created.
+            await self.close()
+            raise
 
     async def close(self) -> None:
         """Close an owned/opted-in bus once when it exposes `close()`."""

tests/test_adapters.py

@@ -8,6 +8,9 @@
 from typing import Any
 
 from fastapi import Depends, FastAPI, Request
+from starlette.applications import Starlette
+from starlette.requests import Request as StarletteRequest
+from starlette.responses import JSONResponse
 
 from eventful import EventBus
 from eventful.adapters.fastapi import event_bus_dependency, install_eventful
@@ -21,10 +24,12 @@ def __init__(self) -> None:
         """Create an open local bus."""
         super().__init__()
         self.closed = False
+        self.close_calls = 0
 
     async def close(self) -> None:
         """Mark the test bus closed."""
         self.closed = True
+        self.close_calls += 1
 
 
 async def terminal_app(scope: dict[str, Any], receive: Any, send: Any) -> None:
@@ -139,6 +144,68 @@ def test_lifespan_respects_external_and_explicit_ownership() -> None:
     assert owned.closed is True
 
 
+def test_repeated_lifespan_shutdown_closes_owned_bus_once() -> None:
+    """Make repeated server lifespan cycles safe and cleanup idempotent."""
+    bus = CloseableBus()
+    middleware = EventfulMiddleware(terminal_app, bus_factory=lambda: bus)
+
+    assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete"
+    assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete"
+    assert bus.close_calls == 1
+
+
+def test_concurrent_application_instances_have_distinct_owned_buses() -> None:
+    """Never share implicitly created buses between application instances."""
+    first = EventfulMiddleware(terminal_app, bus_factory=CloseableBus)
+    second = EventfulMiddleware(terminal_app, bus_factory=CloseableBus)
+
+    async def exercise() -> None:
+        await asyncio.gather(invoke_lifespan(first), invoke_lifespan(second))
+
+    asyncio.run(exercise())
+    assert first.bus is not second.bus
+    assert first.bus.closed is True
+    assert second.bus.closed is True
+
+
+def test_owned_bus_is_cleaned_up_after_startup_failure() -> None:
+    """Release adapter resources when startup fails or raises."""
+    async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None:
+        await receive()
+        await send({"type": "lifespan.startup.failed", "message": "nope"})
+
+    failed_bus = CloseableBus()
+    failed = EventfulMiddleware(failed_app, bus_factory=lambda: failed_bus)
+    assert asyncio.run(invoke_lifespan(failed)) == ["lifespan.startup.failed"]
+    assert failed_bus.close_calls == 1
+
+    async def raising_app(scope: dict[str, Any], receive: Any, send: Any) -> None:
+        await receive()
+        raise RuntimeError("startup exploded")
+
+    raised_bus = CloseableBus()
+    raised = EventfulMiddleware(raising_app, bus_factory=lambda: raised_bus)
+    try:
+        asyncio.run(invoke_lifespan(raised))
+    except RuntimeError as exc:
+        assert str(exc) == "startup exploded"
+    else:
+        raise AssertionError("startup exception should propagate")
+    assert raised_bus.close_calls == 1
+
+
+def test_startup_failure_preserves_application_owned_bus() -> None:
+    """Do not clean up an injected bus merely because application startup fails."""
+    async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None:
+        await receive()
+        await send({"type": "lifespan.startup.failed"})
+
+    bus = CloseableBus()
+    middleware = EventfulMiddleware(failed_app, bus=bus)
+    asyncio.run(invoke_lifespan(middleware))
+    assert bus.close_calls == 0
+
+
 def test_fastapi_installs_middleware_and_dependency_uses_request_state() -> None:
     """Use FastAPI's middleware registry and Request annotation contract."""
     app = FastAPI()
@@ -157,6 +224,19 @@ def bus_endpoint(resolved: EventBus = Depends(dependency)):
     assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True})
 
 
+def test_supported_starlette_application_uses_request_state() -> None:
+    """Exercise the public middleware API on the supported Starlette release."""
+    app = Starlette()
+    bus = CloseableBus()
+    app.add_middleware(EventfulMiddleware, bus=bus)
+
+    async def endpoint(request: StarletteRequest) -> JSONResponse:
+        return JSONResponse({"same": request_event_bus(request) is bus})
+
+    app.add_route("/bus", endpoint)
+    assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True})
+
+
 def test_request_event_bus_requires_middleware_state() -> None:
     """Avoid silently leaking the process-global bus into unconfigured requests."""
     request = SimpleNamespace(state=SimpleNamespace())