fix context none in eventstream

Author: Iwaniukooo11

src/a2a/server/cluster/database_event_stream.py

@@ -122,20 +122,20 @@ async def _seq_at_or_before_version(
         session: AsyncSession,
         task_id: str,
         after: TaskVersion,
-        owner: str | None,
+        owner: str,
     ) -> int:
         """Seq to start polling from so nothing after `after` is missed."""
         if after.is_missing:
             return 0
-        conditions = [
-            self._event_model.task_id == task_id,
-            self._event_model.task_version <= _as_int(after),
-        ]
-        if owner is not None:
-            conditions.append(self._event_model.owner == owner)
         stmt = (
             select(self._event_model.seq)
-            .where(and_(*conditions))
+            .where(
+                and_(
+                    self._event_model.task_id == task_id,
+                    self._event_model.owner == owner,
+                    self._event_model.task_version <= _as_int(after),
+                )
+            )
             .order_by(self._event_model.seq.desc())
             .limit(1)
         )
@@ -151,7 +151,9 @@ async def subscribe(  # type: ignore[override]
     ) -> AsyncGenerator[VersionedEvent, None]:
         """Polls the log for events of `task_id` newer than `after`."""
         await self._ensure_initialized()
-        owner = self.owner_resolver(context) if context is not None else None
+        if context is None:
+            context = ServerCallContext()
+        owner = self.owner_resolver(context)
         async with self._session_maker() as session:
             cursor = await self._seq_at_or_before_version(
                 session, task_id, after, owner
@@ -160,19 +162,19 @@ async def subscribe(  # type: ignore[override]
         while True:
             try:
                 async with self._session_maker() as session:
-                    conditions = [
-                        self._event_model.task_id == task_id,
-                        self._event_model.seq > cursor,
-                    ]
-                    if owner is not None:
-                        conditions.append(self._event_model.owner == owner)
                     stmt = (
                         select(
                             self._event_model.seq,
                             self._event_model.task_version,
                             self._event_model.event_data,
                         )
-                        .where(and_(*conditions))
+                        .where(
+                            and_(
+                                self._event_model.task_id == task_id,
+                                self._event_model.owner == owner,
+                                self._event_model.seq > cursor,
+                            )
+                        )
                         .order_by(self._event_model.seq.asc())
                     )
                     rows = (await session.execute(stmt)).all()

tests/server/cluster/test_database_event_stream.py

@@ -362,7 +362,7 @@ async def test_owner_context_succeeds_on_all_methods(db_url: str) -> None:
 @pytest.mark.asyncio
 @pytest.mark.timeout(10)
 async def test_other_owner_context_does_not_receive_events(db_url: str) -> None:
-    """A different owner does not see events belonging to another owner."""
+    """A different owner (or no context) does not see events belonging to another owner."""
     task_id = 'task-other-owner'
     store_engine = create_async_engine(db_url)
     stream_engine = create_async_engine(db_url)
@@ -394,10 +394,11 @@ async def test_other_owner_context_does_not_receive_events(db_url: str) -> None:
         )
         other = ServerCallContext(user=SampleUser('someone_else'))
 
-        agen = stream.subscribe(task_id, after=v1, context=other)
-        with pytest.raises(asyncio.TimeoutError):
-            await asyncio.wait_for(agen.__anext__(), timeout=0.15)
-        await agen.aclose()
+        for ctx in (other, None):
+            agen = stream.subscribe(task_id, after=v1, context=ctx)
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(agen.__anext__(), timeout=0.15)
+            await agen.aclose()
     finally:
         await store_engine.dispose()
         await stream_engine.dispose()