📄 event_service_base.py (第 94-143 行)

async def search_events(
    self,
    conversation_id: UUID,
    kind__eq: EventKind | None = None,
    timestamp__gte: datetime | None = None,
    timestamp__lt: datetime | None = None,
    sort_order: EventSortOrder = EventSortOrder.TIMESTAMP,
    page_id: str | None = None,
    limit: int = 100,
) -> EventPage:
    """Search events matching the given filters."""
    loop = asyncio.get_running_loop()
    prefix = await self.get_conversation_path(conversation_id)
    paths = await loop.run_in_executor(None, self._search_paths, prefix)

    events = await self._load_events_from_paths(paths)
    # 时间过滤：datetime 转为 ISO 字符串与 event.timestamp 比较
    timestamp_gte_str = timestamp__gte.isoformat() if timestamp__gte else None
    timestamp_lt_str = timestamp__lt.isoformat() if timestamp__lt else None

    items = []
    for event in events:
        if not event:
            continue
        if kind__eq and event.kind != kind__eq:
            continue
        if timestamp_gte_str and event.timestamp < timestamp_gte_str:
            continue
        if timestamp_lt_str and event.timestamp >= timestamp_lt_str:
            continue
        items.append(event)

    if sort_order:
        items.sort(
            key=lambda e: e.timestamp,
            reverse=(sort_order == EventSortOrder.TIMESTAMP_DESC),
        )

    # 分页逻辑
    start_offset = 0
    next_page_id = None
    if page_id:
        start_offset = int(page_id)
        items = items[start_offset:]
    if len(items) > limit:
        next_page_id = str(start_offset + limit)
        items = items[:limit]

    return EventPage(items=items, next_page_id=next_page_id)

📄 filesystem_event_service.py (第 17-43 行)

@dataclass
class FilesystemEventService(EventServiceBase):
    """Event service based on file system"""

    limit: int = 500

    def _load_event(self, path: Path) -> Event | None:
        try:
            content = path.read_text()
            content = Event.model_validate_json(content)
            return content
        except Exception:
            if path.exists():
                _logger.exception('Error reading event', stack_info=True)
            return None

    def _store_event(self, path: Path, event: Event):
        path.parent.mkdir(parents=True, exist_ok=True)
        content = event.model_dump_json(indent=2)
        path.write_text(content)

    def _search_paths(self, prefix: Path, page_id: str | None = None) -> list[Path]:
        search_path = f'{prefix}/*'
        files = glob.glob(str(search_path))
        paths = [Path(file) for file in files]
        return paths

5.2 AwsEventService — AWS S3

@dataclass
class AwsEventService(EventServiceBase):
    """AWS S3-based implementation of EventService.
    Uses role-based authentication, so no explicit credentials are needed."""

    s3_client: Any
    bucket_name: str

    def _load_event(self, path: Path) -> Event | None:
        try:
            response = self.s3_client.get_object(
                Bucket=self.bucket_name, Key=str(path)
            )
            with response['Body'] as stream:
                json_data = stream.read().decode('utf-8')
            event = Event.model_validate_json(json_data)
            return event
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchKey':
                return None
            _logger.exception(f'Error reading event from {path}', stack_info=True)
            return None

    def _search_paths(self, prefix: Path, page_id: str | None = None) -> list[Path]:
        kwargs = {'Bucket': self.bucket_name, 'Prefix': str(prefix)}
        if page_id:
            kwargs['ContinuationToken'] = page_id
        response = self.s3_client.list_objects_v2(**kwargs)
        contents = response.get('Contents', [])
        paths = [Path(obj['Key']) for obj in contents]
        return paths

关键点：使用 IAM Role 认证，无需显式密钥。S3 Key 路径与文件系统路径完全一致，实现了存储透明切换。

@lru_cache(maxsize=1)
def _get_shared_storage_client() -> Client:
    """Return a process-wide shared GCS client.
    google.cloud.storage.Client is thread-safe and manages its own urllib3
    connection pool.  Creating one per request leads to pool exhaustion under
    load ("Connection pool is full, discarding connection")."""
    return storage.Client()

@dataclass
class GoogleCloudEventService(EventServiceBase):
    """Google Cloud Storage-based implementation of EventService."""

    bucket: Bucket

    def _load_event(self, path: Path) -> Event | None:
        blob: Blob = self.bucket.blob(str(path))
        try:
            with blob.open('r') as f:
                json_data = f.read()
            event = Event.model_validate_json(json_data)
            return event
        except NotFound:
            return None

    def _search_paths(self, prefix: Path, page_id: str | None = None) -> list[Path]:
        blobs: Iterator[Blob] = self.bucket.list_blobs(
            page_token=page_id, prefix=str(prefix)
        )
        paths = list(Path(blob.name) for blob in blobs)
        return paths

class StorageProvider(str, Enum):
    """Storage provider types for event and shared event storage."""
    AWS = 'aws'
    GCP = 'gcp'
    FILESYSTEM = 'filesystem'

def get_storage_provider() -> StorageProvider:
    provider = os.environ.get('SHARED_EVENT_STORAGE_PROVIDER', '').lower()
    if not provider:
        provider = os.environ.get('FILE_STORE', '').lower()

    if provider in ('aws', 's3'):
        return StorageProvider.AWS
    elif provider in ('gcp', 'google_cloud'):
        return StorageProvider.GCP
    else:
        return StorageProvider.FILESYSTEM

if config.event is None:
    provider = get_storage_provider()

    if provider == StorageProvider.AWS:
        bucket_name = os.environ.get('FILE_STORE_PATH')
        if not bucket_name:
            raise ValueError(
                'FILE_STORE_PATH environment variable is required for S3 storage'
            )
        config.event = AwsEventServiceInjector(bucket_name=bucket_name)
    elif provider == StorageProvider.GCP:
        bucket_name = os.environ.get('FILE_STORE_PATH')
        if not bucket_name:
            raise ValueError(
                'FILE_STORE_PATH environment variable is required for Google Cloud storage'
            )
        config.event = GoogleCloudEventServiceInjector(bucket_name=bucket_name)
    else:
        config.event = FilesystemEventServiceInjector()

class EventCallbackStatus(Enum):
    ACTIVE = 'ACTIVE'
    DISABLED = 'DISABLED'
    COMPLETED = 'COMPLETED'
    ERROR = 'ERROR'

class EventCallbackProcessor(DiscriminatedUnionMixin, ABC):
    event_kind: ClassVar[EventKind] = 'MessageEvent'

    @abstractmethod
    async def __call__(
        self,
        conversation_id: UUID,
        callback: EventCallback,
        event: Event,
    ) -> EventCallbackResult | None:
        """Process an event."""

class EventCallback(CreateEventCallbackRequest):
    id: OpenHandsUUID = Field(default_factory=uuid4)
    status: EventCallbackStatus = Field(default=EventCallbackStatus.ACTIVE)
    created_at: datetime = Field(default_factory=utc_now)
    updated_at: datetime = Field(default_factory=utc_now)

@dataclass
class SQLEventCallbackService(EventCallbackService):
    """SQL implementation of EventCallbackService.

    The service does **not** hold a long-lived ``AsyncSession``. Each public
    method opens a short-lived session via ``self.async_session_maker`` for the
    duration of its own DB work. This means:

    * The pool connection is only checked out while the method runs an
      actual SQL statement — it is released as soon as the method returns,
      so callers never accidentally pin a connection across slow logic
      (e.g. webhook deliveries, ``asyncio.gather`` of callback processors).
    * Methods are independent: a failure in one (e.g. a flush error) cannot
      leave another with a session in an unknown transactional state.
    * The service itself is no longer an async-context-manager. There is
      no per-service ``__aenter__``/``__aexit__`` lifecycle to get wrong.
    """

    async_session_maker: async_sessionmaker

async def execute_callbacks(self, conversation_id: UUID, event: Event) -> None:
    """Run all active callbacks for the event and persist their results."""
    callbacks = await self.get_active_callbacks(conversation_id, event)
    if not callbacks:
        return
    outcomes = await asyncio.gather(
        *[invoke_callback(cb, conversation_id, event) for cb in callbacks],
        return_exceptions=True,
    )
    normalised: list[StoredEventCallbackResult | None] = []
    for callback, outcome in zip(callbacks, outcomes, strict=False):
        if isinstance(outcome, BaseException):
            _logger.exception(
                f'Exception in callback {callback.id}', stack_info=True
            )
            normalised.append(
                StoredEventCallbackResult(
                    status=EventCallbackResultStatus.ERROR,
                    event_callback_id=callback.id,
                    event_id=event.id,
                    conversation_id=conversation_id,
                    detail=str(outcome),
                )
            )
        else:
            normalised.append(outcome)
    await self.persist_callback_results(callbacks, normalised)

# Delay between attempts to poll title
_POLL_DELAY_S = 3
# Number of attempts to poll title
_NUM_POLL_ATTEMPTS = 4
# Avoid starting one slow title poll per webhook event.
_CONVERSATIONS_BEING_POLLED: set[UUID] = set()

async def _poll_for_title(
    httpx_client: httpx.AsyncClient,
    url: str,
    session_api_key: str | None,
) -> str | None:
    """Poll the agent server for the conversation title."""
    for _ in range(_NUM_POLL_ATTEMPTS):
        await asyncio.sleep(_POLL_DELAY_S)
        try:
            headers = {'X-Session-API-Key': session_api_key} if session_api_key else {}
            response = await httpx_client.get(url, headers=headers)
            response.raise_for_status()
        except httpx.HTTPError as exc:
            _logger.warning('Title poll failed for conversation %s: %s', url, exc)
        else:
            title = response.json().get('title')
            if title:
                return title
    return None

router = APIRouter(
    prefix='/conversation/{conversation_id}/events',
    tags=['Events'],
    dependencies=get_dependencies(),
)
event_service_dependency = depends_event_service()

@router.get('/search')
async def search_events(
    conversation_id: str,
    kind__eq: EventKind | None = None,
    timestamp__gte: datetime | None = None,
    timestamp__lt: datetime | None = None,
    sort_order: EventSortOrder = EventSortOrder.TIMESTAMP,
    page_id: str | None = None,
    limit: int = 100,
    event_service: EventService = event_service_dependency,
) -> EventPage:
    """Search / List events."""
    return await event_service.search_events(...)

@router.get('/count')
async def count_events(...) -> int:
    """Count events matching the given filters."""

@router.get('')
async def batch_get_events(
    conversation_id: str,
    id: list[str],
    event_service: EventService = event_service_dependency,
) -> list[Event | None]:
    """Get a batch of events given their ids."""
    if len(id) > 100:
        raise HTTPException(status_code=400, ...)
    event_ids = [UUID(id_) for id_ in id]
    return await event_service.batch_get_events(UUID(conversation_id), event_ids)