Providing a custom lifespan overrides FastA2A’s default lifespan (and skips critical aenter)

[joshua-brumpton-uniper]

Summary

In FastA2A, passing a lifespan at construction replaces the built-in _default_lifespan, which means the essential initialization (entering task_manager and thus the broker) is skipped.

Expected behavior

The FastA2A default (compulsory) lifespan should always execute (to enter task_manager/broker).

If the user provides a lifespan, it should be composed with the default, not replace it.

Actual behavior

Providing any lifespan causes Starlette to use that instead of _default_lifespan.

As a result, task_manager (and broker) are not entered unless the user re-implements that logic themselves.

Minimal reproduction

from contextlib import asynccontextmanager

# Simplified FastA2A excerpt
class FastA2A(Starlette):
    def __init__(self, *, storage, broker, lifespan=None, **kwargs):
        if lifespan is None:
            lifespan = _default_lifespan  # enters task_manager/broker
        super().__init__(lifespan=lifespan, **kwargs)

@asynccontextmanager
async def _default_lifespan(app: "FastA2A"):
    async with app.task_manager:  # also enters broker
        yield

# --- User code: wants extra setup during startup
@asynccontextmanager
async def user_lifespan(app):
    # custom setup/teardown
    yield

# This currently REPLACES _default_lifespan entirely:
app = FastA2A(storage=..., broker=..., lifespan=user_lifespan)

# => app.task_manager / broker are never entered by default

Root cause

Starlette’s constructor accepts a single lifespan callable. FastA2A currently passes either the default or the user’s, never both, so the default gets lost when a custom lifespan is provided.

Proposed fix:

Compose lifespans and make default compulsory

Introduce a small composer that always wraps the user’s lifespan (if any) inside the default “compulsory” lifespan:

from contextlib import asynccontextmanager
from typing import AsyncIterator, Callable, Optional

LifespanFn = Callable[["FastA2A"], AsyncIterator[None]]

@asynccontextmanager
async def compulsory_lifespan(app: "FastA2A"):
    async with app.task_manager:  # enters broker too
        yield

def compose_lifespans(outer: LifespanFn, inner: Optional[LifespanFn]) -> LifespanFn:
    @asynccontextmanager
    async def _composed(app: "FastA2A"):
        async with outer(app):
            if inner is None:
                yield
            else:
                async with inner(app):
                    yield
    return _composed

class FastA2A(Starlette):
    def __init__(self, *, lifespan: LifespanFn | None = None, **kwargs):
        # Always wrap with the compulsory/default lifespan
        base = compulsory_lifespan  # formerly _default_lifespan
        composed = compose_lifespans(base, lifespan)
        super().__init__(lifespan=composed, **kwargs)