fix(server): create the cluster tables lazily in VersionedDatabaseTaskStore get and delete (#1357)

# Description

Thank you for opening a Pull Request!
Before submitting your PR, there are a few things you can do to make
sure it goes smoothly:

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [x] Ensure the tests and linter pass (Run `bash scripts/format.sh`
from the repository root to format)
- [x] Appropriate docs were updated (if necessary)

Fixes #1356 🦕

## Problem

`VersionedDatabaseTaskStore` creates its tables lazily, like
`DatabaseTaskStore`, but only `save()` creates the cluster tables.
`get()` and `delete()` initialize the underlying `DatabaseTaskStore`
only, so when the store is used without calling `initialize()` and the
first operation is a `get()` or a `delete()`, the query on
`task_versions` fails:

```python
store = VersionedDatabaseTaskStore(engine=create_async_engine('sqlite+aiosqlite:///:memory:'))

await store.get('missing', ServerCallContext())
# sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: task_versions
```

## Fix

`get()` and `delete()` (`src/a2a/server/cluster/database_task_store.py`)
now call `self._ensure_cluster_tables()` after initializing the
underlying store, the same way `save()` does. `delete()` also
initializes the underlying store before creating the cluster tables, so
the tables are created in the same order as in `save()`.

`_ensure_cluster_tables()` returns right away once the tables are ready,
so stores that call `initialize()` are not affected.

## Test

Tests were added to the existing suite at
`tests/server/cluster/test_database_versioned_task_store.py`, using a
new `uninitialized_store` fixture that builds the store on an empty
database with `create_table=True` and without calling `initialize()`:

- `test_get_without_initialize_creates_tables`: `get()` of a missing
task returns `None`.
- `test_delete_without_initialize_creates_tables`: `delete()` of a
missing task does not fail, and a later `get()` returns `None`.

Both tests fail on `main` with `no such table: task_versions` and pass
with the fix. No existing test was modified or removed. Happy to adapt
the test approach based on maintainer feedback.

Co-authored-by: mykytanetipa <mykytanetipa@google.com>

Author: cchinchilla-dev

src/a2a/server/cluster/database_task_store.py

@@ -258,6 +258,7 @@ async def get(
         commits) rather than failing the read.
         """
         await self._db._ensure_initialized()  # noqa: SLF001
+        await self._ensure_cluster_tables()
         owner = self._db.owner_resolver(context)
         attempts = 0
         while True:
@@ -355,6 +356,8 @@ async def list(
 
     async def delete(self, task_id: str, context: ServerCallContext) -> None:
         """Deletes a task and its version row via the underlying store."""
+        await self._db._ensure_initialized()  # noqa: SLF001
+        await self._ensure_cluster_tables()
         owner = self._db.owner_resolver(context)
         await self._db.delete(task_id, context)
         async with self._db.async_session_maker.begin() as session:

tests/server/cluster/test_database_versioned_task_store.py

@@ -136,6 +136,24 @@ async def versioned_store(
         await engine.dispose()
 
 
+@pytest_asyncio.fixture(params=DB_CONFIGS)
+async def uninitialized_store(
+    request,
+) -> AsyncGenerator[VersionedDatabaseTaskStore, None]:
+    """A store that creates its tables lazily, without `initialize()`."""
+    db_url, dialect_name = request.param
+    if db_url is None:
+        pytest.skip(f'DSN for {dialect_name} not set in environment variables.')
+    engine = create_async_engine(db_url)
+    _enforce_sqlite_foreign_keys(engine)
+    try:
+        yield VersionedDatabaseTaskStore(engine=engine, create_table=True)
+    finally:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+        await engine.dispose()
+
+
 @pytest.mark.asyncio
 async def test_is_a_versioned_task_store(
     versioned_store: VersionedDatabaseTaskStore,
@@ -654,6 +672,21 @@ async def test_delete_removes_version_row(
     assert stored is not None
 
 
+@pytest.mark.asyncio
+async def test_get_without_initialize_creates_tables(
+    uninitialized_store: VersionedDatabaseTaskStore,
+) -> None:
+    assert await uninitialized_store.get('task-abc', TEST_CONTEXT) is None
+
+
+@pytest.mark.asyncio
+async def test_delete_without_initialize_creates_tables(
+    uninitialized_store: VersionedDatabaseTaskStore,
+) -> None:
+    await uninitialized_store.delete('task-abc', TEST_CONTEXT)
+    assert await uninitialized_store.get('task-abc', TEST_CONTEXT) is None
+
+
 @pytest.mark.asyncio
 async def test_versions_count_writes(
     versioned_store: VersionedDatabaseTaskStore,