From 35e470b2e77d5829f79aa043ff26671d4b1a87fa Mon Sep 17 00:00:00 2001 From: Yinon Kahta Date: Thu, 27 Aug 2026 13:05:58 +0300 Subject: [PATCH] feat(pointfive): add batching logger callback --- litellm/__init__.py | 4 + litellm/integrations/pointfive/__init__.py | 5 + litellm/integrations/pointfive/logger.py | 221 ++++++++ .../integrations/pointfive/test_logger.py | 472 ++++++++++++++++++ 4 files changed, 702 insertions(+) create mode 100644 litellm/integrations/pointfive/__init__.py create mode 100644 litellm/integrations/pointfive/logger.py create mode 100644 tests/test_litellm/integrations/pointfive/test_logger.py diff --git a/litellm/__init__.py b/litellm/__init__.py index c83e72a78b4..9656f2893a6 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -42,8 +42,10 @@ from typing import ( TYPE_CHECKING, Union, ) +from collections.abc import Mapping from litellm.types.integrations.datadog import DatadogInitParams from litellm.types.integrations.newrelic import NewRelicInitParams +from litellm.types.integrations.pointfive import PointFiveInitParams from litellm._logging import ( set_verbose, _turn_on_debug, @@ -150,6 +152,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "smtp_email", "deepeval", "s3_v2", + "pointfive", "aws_sqs", "vector_store_pre_call_hook", "dotprompt", @@ -428,6 +431,7 @@ s3_audit_callback_params: Optional[Dict] = None datadog_llm_observability_params: Optional[Union[DatadogLLMObsInitParams, Dict]] = None datadog_params: Optional[Union[DatadogInitParams, Dict]] = None newrelic_params: Optional[Union[NewRelicInitParams, Dict]] = None +pointfive_params: Optional[Union[PointFiveInitParams, Mapping[str, object]]] = None aws_sqs_callback_params: Optional[Dict] = None generic_logger_headers: Optional[Dict] = None default_key_generate_params: Optional[Dict] = None diff --git a/litellm/integrations/pointfive/__init__.py b/litellm/integrations/pointfive/__init__.py new file mode 100644 index 00000000000..1f3ca3c65c7 --- /dev/null +++ b/litellm/integrations/pointfive/__init__.py @@ -0,0 +1,5 @@ +"""PointFive logging integration for LiteLLM.""" + +from litellm.integrations.pointfive.logger import PointFiveLogger + +__all__ = ("PointFiveLogger",) diff --git a/litellm/integrations/pointfive/logger.py b/litellm/integrations/pointfive/logger.py new file mode 100644 index 00000000000..548777c0416 --- /dev/null +++ b/litellm/integrations/pointfive/logger.py @@ -0,0 +1,221 @@ +""" +PointFive logging integration. + +Buffers ``StandardLoggingPayload`` records and ships each flush as one gzipped +newline-delimited JSON object, rather than one object per request. Uploads go through a +presigned URL issued by the PointFive API, so the proxy needs no cloud credentials and +runs unchanged wherever it is hosted. +""" + +import asyncio +from collections.abc import Mapping +from datetime import datetime +from typing import Final + +import litellm +from litellm._logging import verbose_logger +from litellm.integrations.custom_batch_logger import CustomBatchLogger +from litellm.integrations.pointfive.payload import chunk_lines, encode_lines, serialize_records +from litellm.integrations.pointfive.upload_client import PointFiveUploadClient, PointFiveUploadError +from litellm.llms.custom_httpx.http_handler import get_async_httpx_client, httpxSpecialProvider +from litellm.secret_managers.main import get_secret_str +from litellm.types.integrations.base_health_check import IntegrationHealthCheckStatus +from litellm.types.integrations.pointfive import DEFAULT_API_URL, PointFiveInitParams, PointFiveUploadFailure + +_ENV_REFERENCE_PREFIX: Final = "os.environ/" + + +def _resolved_secret(value: str | None) -> str | None: + """Resolve an ``os.environ/VAR`` config reference, the way config.yaml spells secrets.""" + if value is None or not value.startswith(_ENV_REFERENCE_PREFIX): + return value + return get_secret_str(value) + + +def _configured_params() -> PointFiveInitParams: + """Read ``litellm.pointfive_params``, validating a raw config dict on the way through.""" + configured: Final = litellm.pointfive_params + if isinstance(configured, PointFiveInitParams): + return configured + if isinstance(configured, Mapping): + return PointFiveInitParams.model_validate(configured) + return PointFiveInitParams() + + +def _resolved_api_key(params: PointFiveInitParams) -> str | None: + """Prefer the configured key, falling back to the environment the proxy UI writes.""" + return _resolved_secret(params.api_key) or get_secret_str("POINTFIVE_API_KEY") + + +def _resolved_api_url(params: PointFiveInitParams) -> str: + """Prefer the configured url, then the environment, then the public endpoint.""" + return _resolved_secret(params.api_url) or get_secret_str("POINTFIVE_API_URL") or DEFAULT_API_URL + + +def _upload_client_for(params: PointFiveInitParams) -> PointFiveUploadClient: + """ + Build an upload client for the key and url configured right now. + + Resolved per call rather than kept: the proxy ui writes new values into the + environment of a running proxy, and reading them once would need a restart to take + effect. ``get_async_httpx_client`` is cached, so this reuses the same connections. + """ + api_key: Final = _resolved_api_key(params) + if not api_key: + raise ValueError( + "pointfive logging requires an api key. Set POINTFIVE_API_KEY, or " + "litellm_settings.pointfive_params.api_key in config.yaml" + ) + return PointFiveUploadClient( + api_key=api_key, + api_url=_resolved_api_url(params), + http_client=get_async_httpx_client(llm_provider=httpxSpecialProvider.LoggingCallback), + max_retries=params.max_upload_retries, + ) + + +class PointFiveLogger(CustomBatchLogger): + """Batching callback that ships LiteLLM request logs to PointFive.""" + + preserve_events_added_during_flush = True + + def __init__( + self, + params: PointFiveInitParams | None = None, + upload_client: PointFiveUploadClient | None = None, + start_periodic_flush: bool = True, + ) -> None: + resolved: Final = params if params is not None else _configured_params() + self.max_batch_bytes: Final = resolved.max_batch_bytes + self.params: Final = resolved + self.given_upload_client: Final = upload_client + if upload_client is None: + _upload_client_for(resolved) # refuse to start without a key, rather than at the first flush + super().__init__( + flush_lock=asyncio.Lock(), + batch_size=resolved.batch_size, + flush_interval=resolved.flush_interval, + turn_off_message_logging=bool(resolved.turn_off_message_logging), + ) + # A caller that only wants one answer out of this logger, such as a health check, + # passes start_periodic_flush=False so it leaves no flush task behind. + self._flushing: bool = False + self._periodic_flush_task: asyncio.Task[None] | None = ( + self._start_periodic_flush_task() if start_periodic_flush else None + ) + + @property + def upload_client(self) -> PointFiveUploadClient: + """The client for the currently configured key and url, so a ui edit needs no restart.""" + if self.given_upload_client is not None: + return self.given_upload_client + return _upload_client_for(self.params) + + def _start_periodic_flush_task(self) -> asyncio.Task[None] | None: + """Start the periodic flush only once an event loop is actually running.""" + try: + loop: Final = asyncio.get_running_loop() + except RuntimeError: + return None + return loop.create_task(self.periodic_flush()) + + async def async_log_success_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: datetime, + end_time: datetime, + ) -> None: + await self._enqueue(kwargs) + + async def async_log_failure_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: datetime, + end_time: datetime, + ) -> None: + await self._enqueue(kwargs) + + async def _enqueue(self, kwargs: Mapping[str, object]) -> None: + """Buffer one record, flushing early once the batch threshold is reached.""" + try: + if self._periodic_flush_task is None or self._periodic_flush_task.done(): + self._periodic_flush_task = self._start_periodic_flush_task() + + payload: Final = kwargs.get("standard_logging_object") + if not isinstance(payload, dict): + verbose_logger.debug("pointfive: event carried no standard_logging_object, skipping") + return + + self.log_queue.append(payload) + if len(self.log_queue) >= self.batch_size: + await self.flush_queue(skip_if_flushing=True) + except Exception: # noqa: BLE001 # logging must never break the request path + verbose_logger.exception("pointfive: failed to queue an event") + + async def flush_queue(self, skip_if_flushing: bool = False) -> None: + """ + Flush as usual, or report liveness when there is nothing to send. + + ``CustomBatchLogger`` skips an empty queue entirely, so without this an idle proxy + would look identical to a dead one. + + ``skip_if_flushing`` is what a full batch passes. Uploading one takes seconds, and + every event arriving meanwhile crosses the threshold too, so each would queue on the + flush lock and then ship the handful of records left behind it. That turns one burst + into a stream of tiny objects, which is what batching exists to avoid. The running + flush already carries what is queued, and the interval catches whatever it missed. + """ + if not self.log_queue: + await self._ping() + return + if skip_if_flushing and self._flushing: + return + + self._flushing = True + try: + await super().flush_queue() + finally: + self._flushing = False + + async def async_health_check(self) -> IntegrationHealthCheckStatus: + """Answer the proxy ui test button by asking the api whether it accepts this key.""" + try: + failure: Final = await self.upload_client.ping() + except ValueError as missing_key: + return IntegrationHealthCheckStatus(status="unhealthy", error_message=str(missing_key)) + if failure is not None: + return IntegrationHealthCheckStatus(status="unhealthy", error_message=failure.detail) + return IntegrationHealthCheckStatus(status="healthy", error_message=None) + + async def _ping(self) -> None: + """Report liveness, never failing the flush over it.""" + try: + failure: Final = await self.upload_client.ping() + except ValueError as missing_key: + verbose_logger.warning("pointfive: liveness ping skipped, %s", missing_key) + return + if failure is not None: + verbose_logger.warning("pointfive: liveness ping failed, %s", failure.detail) + + async def async_send_batch(self) -> None: + """ + Upload everything queued, split into objects of at most ``max_batch_bytes``. + + A retryable failure propagates so ``CustomBatchLogger`` keeps the whole batch for + the next flush, which can re-send objects that already landed, making delivery + at-least-once. A rejection the server will refuse again drops the object instead, + since holding it would block every record queued behind it. + """ + pending: Final = tuple(self.log_queue) + if not pending: + return + + for chunk in chunk_lines(serialize_records(pending), self.max_batch_bytes): + outcome = await self.upload_client.upload(await encode_lines(chunk)) + if not isinstance(outcome, PointFiveUploadFailure): + continue + if outcome.retryable: + raise PointFiveUploadError(outcome.detail) + verbose_logger.error("pointfive: dropping %s records, %s", len(chunk), outcome.detail) diff --git a/tests/test_litellm/integrations/pointfive/test_logger.py b/tests/test_litellm/integrations/pointfive/test_logger.py new file mode 100644 index 00000000000..5979b3af146 --- /dev/null +++ b/tests/test_litellm/integrations/pointfive/test_logger.py @@ -0,0 +1,472 @@ +import asyncio +import gzip +import json +import logging +from collections.abc import Callable + +import pytest + +from litellm.integrations.pointfive.logger import PointFiveLogger +from litellm.integrations.pointfive.upload_client import PointFiveUploadError +from litellm.types.integrations.pointfive import PointFiveInitParams, PointFiveUploadFailure + +OBJECT_KEY = "some/object.ndjson.gz" + + +class FakeUploadClient: + """Records the objects a flush produced, so tests can read what would have shipped.""" + + def __init__( + self, + outcomes: list[str | PointFiveUploadFailure] | None = None, + ping_failure: PointFiveUploadFailure | None = None, + ) -> None: + self.outcomes = outcomes or [OBJECT_KEY] + self.bodies: list[bytes] = [] + self.on_upload: Callable[[], None] | None = None + self.ping_failure = ping_failure + self.pings = 0 + + async def ping(self) -> PointFiveUploadFailure | None: + self.pings += 1 + return self.ping_failure + + async def upload(self, body: bytes) -> str | PointFiveUploadFailure: + if self.on_upload is not None: + self.on_upload() + self.bodies.append(body) + return self.outcomes.pop(0) if len(self.outcomes) > 1 else self.outcomes[0] + + def records(self) -> list[dict]: + return [json.loads(line) for body in self.bodies for line in gzip.decompress(body).decode().splitlines()] + + +def _logger(upload_client: FakeUploadClient, **params) -> PointFiveLogger: + return PointFiveLogger(params=PointFiveInitParams(**params), upload_client=upload_client) + + +def _event(request_id: str, size: int = 0) -> dict: + return {"standard_logging_object": {"id": request_id, "model": "gpt-4o", "blob": "x" * size}} + + +@pytest.mark.asyncio +async def test_a_flush_ships_one_object_holding_every_buffered_record(): + """One object per flush is the whole point: s3_v2 sends one per request.""" + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=3) + + for request_id in ("a", "b", "c"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + assert len(upload_client.bodies) == 1 + assert [record["id"] for record in upload_client.records()] == ["a", "b", "c"] + + +@pytest.mark.asyncio +async def test_records_are_held_until_the_batch_is_full(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=3) + + await logger.async_log_success_event(_event("a"), None, None, None) + + assert upload_client.bodies == [] + assert len(logger.log_queue) == 1 + + +@pytest.mark.asyncio +async def test_failed_requests_are_logged_too(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=1) + + await logger.async_log_failure_event(_event("failed"), None, None, None) + + assert [record["id"] for record in upload_client.records()] == ["failed"] + + +@pytest.mark.asyncio +async def test_an_event_without_a_standard_payload_is_skipped(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=1) + + await logger.async_log_success_event({"kwargs": "but no payload"}, None, None, None) + + assert upload_client.bodies == [] + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_a_batch_over_the_byte_cap_ships_as_several_objects(): + """Record count cannot bound an object: an unredacted payload dwarfs a redacted one.""" + cap = 600 + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=4, max_batch_bytes=cap) + + for request_id in ("a", "b", "c", "d"): + await logger.async_log_success_event(_event(request_id, size=200), None, None, None) + + assert len(upload_client.bodies) > 1 + assert [record["id"] for record in upload_client.records()] == ["a", "b", "c", "d"] + assert all(len(gzip.decompress(body)) <= cap for body in upload_client.bodies) + + +@pytest.mark.asyncio +async def test_a_retryable_failure_keeps_the_batch_for_the_next_flush(): + upload_client = FakeUploadClient([PointFiveUploadFailure("upload target is down", retryable=True)]) + logger = _logger(upload_client, batch_size=2) + + for request_id in ("a", "b"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + assert [record["id"] for record in logger.log_queue] == ["a", "b"] + + +@pytest.mark.asyncio +async def test_a_retryable_failure_surfaces_so_the_base_logger_can_preserve_it(): + upload_client = FakeUploadClient([PointFiveUploadFailure("upload target is down", retryable=True)]) + logger = _logger(upload_client, batch_size=99) + logger.log_queue.append(_event("a")["standard_logging_object"]) + + with pytest.raises(PointFiveUploadError, match="upload target is down"): + await logger.async_send_batch() + + +@pytest.mark.asyncio +async def test_a_rejected_batch_is_dropped_rather_than_blocking_the_queue(): + """Retrying a rejection forever would stall every record queued behind it.""" + upload_client = FakeUploadClient([PointFiveUploadFailure("object too large", retryable=False)]) + logger = _logger(upload_client, batch_size=2) + + for request_id in ("a", "b"): + await logger.async_log_success_event(_event(request_id), None, None, None) + + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_records_already_queued_ship_with_the_event_that_triggers_the_flush(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=1) + logger.log_queue.append(_event("mid-flight")["standard_logging_object"]) + + await logger.async_log_success_event(_event("a"), None, None, None) + + assert [record["id"] for record in upload_client.records()] == ["mid-flight", "a"] + assert logger.log_queue == [] + + +@pytest.mark.asyncio +async def test_a_record_that_arrives_mid_flush_is_kept_for_the_next_one(): + """The queue is drained by count, so a record appended mid-upload must survive.""" + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=1) + upload_client.on_upload = lambda: logger.log_queue.append(_event("late")["standard_logging_object"]) + + await logger.async_log_success_event(_event("first"), None, None, None) + + assert [record["id"] for record in upload_client.records()] == ["first"] + assert [record["id"] for record in logger.log_queue] == ["late"] + + +def test_defaults_favour_fewer_larger_uploads_over_freshness(): + upload_client = FakeUploadClient() + + logger = _logger(upload_client) + + assert logger.batch_size == 10_000 + assert logger.flush_interval == 300 + assert logger.max_batch_bytes == 8 * 1024 * 1024 + + +def test_the_default_api_url_is_the_pointfive_ingress(monkeypatch): + """api.pointfive.co is the host the ingress serves; .com does not resolve to it.""" + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_env") + + logger = PointFiveLogger() + + assert logger.upload_client.api_url == "https://api.pointfive.co/query" + + +def test_the_api_key_can_come_from_the_environment(monkeypatch): + """The proxy ui configures a callback by writing environment variables.""" + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_from_env") + + logger = PointFiveLogger() + + assert logger.upload_client.api_key == "p5tu_from_env" + + +def test_the_api_url_can_come_from_the_environment(monkeypatch): + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_env") + monkeypatch.setenv("POINTFIVE_API_URL", "https://api.staging.pointfive.co/query") + + logger = PointFiveLogger() + + assert logger.upload_client.api_url == "https://api.staging.pointfive.co/query" + + +def test_config_yaml_wins_over_the_environment(monkeypatch): + """A value set in config.yaml is explicit, so it outranks whatever the ui left behind.""" + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_from_env") + monkeypatch.setenv("POINTFIVE_API_URL", "https://from-env.example/query") + + logger = PointFiveLogger( + params=PointFiveInitParams(api_key="p5tu_from_config", api_url="https://from-config.example/query") + ) + + assert logger.upload_client.api_key == "p5tu_from_config" + assert logger.upload_client.api_url == "https://from-config.example/query" + + +def test_a_missing_api_key_fails_at_startup_not_at_the_first_flush(): + with pytest.raises(ValueError, match="api key"): + PointFiveLogger(params=PointFiveInitParams()) + + +def test_an_api_key_can_be_an_environment_reference(monkeypatch): + """config.yaml spells secrets as `os.environ/NAME`, so the plugin must resolve one.""" + monkeypatch.setenv("POINTFIVE_TEST_KEY", "p5tu_from_env") + + logger = PointFiveLogger(params=PointFiveInitParams(api_key="os.environ/POINTFIVE_TEST_KEY")) + + assert logger.upload_client.api_key == "p5tu_from_env" + + +def test_params_are_read_from_litellm_settings(monkeypatch): + import litellm + + monkeypatch.setattr(litellm, "pointfive_params", {"api_key": "p5tu_configured", "batch_size": 7}) + + logger = PointFiveLogger() + + assert logger.upload_client.api_key == "p5tu_configured" + assert logger.batch_size == 7 + + +def test_an_out_of_range_setting_is_rejected(): + with pytest.raises(ValueError, match="batch_size"): + PointFiveInitParams(api_key="p5tu_k", batch_size=0) + + +@pytest.mark.asyncio +async def test_an_idle_flush_reports_liveness_instead_of_uploading(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=99) + + await logger.flush_queue() + + assert upload_client.pings == 1 + assert upload_client.bodies == [] + + +@pytest.mark.asyncio +async def test_a_flush_with_records_uploads_and_does_not_ping(): + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=1) + + await logger.async_log_success_event(_event("a"), None, None, None) + + assert upload_client.pings == 0 + assert len(upload_client.bodies) == 1 + + +@pytest.mark.asyncio +async def test_a_failed_ping_does_not_raise(): + """Liveness is bookkeeping; a proxy must not see errors from it.""" + upload_client = FakeUploadClient(ping_failure=PointFiveUploadFailure("api down", retryable=True)) + logger = _logger(upload_client, batch_size=99) + + await logger.flush_queue() + + assert upload_client.pings == 1 + + +@pytest.mark.asyncio +async def test_health_check_is_healthy_when_the_api_accepts_the_key(): + upload_client = FakeUploadClient() + + assert await _logger(upload_client).async_health_check() == {"status": "healthy", "error_message": None} + assert upload_client.pings == 1 + + +@pytest.mark.asyncio +async def test_health_check_reports_why_the_api_refused(): + """The ui test button shows this message, so a rejected key has to say so rather than pass.""" + upload_client = FakeUploadClient(ping_failure=PointFiveUploadFailure("key was revoked", retryable=False)) + + outcome = await _logger(upload_client).async_health_check() + + assert outcome == {"status": "unhealthy", "error_message": "key was revoked"} + + +def test_the_client_follows_a_key_and_url_changed_after_startup(monkeypatch): + """ + The proxy ui writes new values into a running proxy's environment. + + Reading them once at construction would leave the logger talking to the old endpoint + until someone restarted the proxy. + """ + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_first") + monkeypatch.setenv("POINTFIVE_API_URL", "https://first.example.invalid/query") + logger = PointFiveLogger(params=PointFiveInitParams()) + + assert logger.upload_client.api_key == "p5tu_first" + assert logger.upload_client.api_url == "https://first.example.invalid/query" + + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_second") + monkeypatch.setenv("POINTFIVE_API_URL", "https://second.example.invalid/query") + + assert logger.upload_client.api_key == "p5tu_second" + assert logger.upload_client.api_url == "https://second.example.invalid/query" + + +def test_a_configured_key_still_wins_over_the_environment(monkeypatch): + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_from_env") + logger = PointFiveLogger(params=PointFiveInitParams(api_key="p5tu_from_config")) + + assert logger.upload_client.api_key == "p5tu_from_config" + + +@pytest.mark.asyncio +async def test_health_check_says_so_when_the_key_was_removed(monkeypatch): + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_present") + logger = PointFiveLogger(params=PointFiveInitParams()) + monkeypatch.delenv("POINTFIVE_API_KEY") + + outcome = await logger.async_health_check() + + assert outcome["status"] == "unhealthy" + assert "requires an api key" in (outcome["error_message"] or "") + + +def _pending_flush_tasks() -> tuple[asyncio.Task, ...]: + return tuple(task for task in asyncio.all_tasks() if "periodic_flush" in str(task.get_coro())) + + +@pytest.mark.asyncio +async def test_a_one_shot_logger_leaves_no_flush_task_behind(): + """ + A health check builds a logger for a single answer and drops it. + + Without this, every check would leave a flusher running that keeps pinging for the + lifetime of the proxy. + """ + before = _pending_flush_tasks() + + logger = PointFiveLogger(params=PointFiveInitParams(), upload_client=FakeUploadClient(), start_periodic_flush=False) + + assert logger._periodic_flush_task is None + assert _pending_flush_tasks() == before + + +@pytest.mark.asyncio +async def test_the_logger_flushes_periodically_by_default(): + logger = PointFiveLogger(params=PointFiveInitParams(), upload_client=FakeUploadClient()) + + assert logger._periodic_flush_task is not None + logger._periodic_flush_task.cancel() + + +@pytest.mark.asyncio +async def test_params_already_built_are_used_as_they_are(monkeypatch): + """config.yaml is validated once into a params object; a second validation would be wasted.""" + import litellm + + monkeypatch.setattr(litellm, "pointfive_params", PointFiveInitParams(max_batch_bytes=4096)) + + logger = PointFiveLogger(upload_client=FakeUploadClient()) + + assert logger.max_batch_bytes == 4096 + + +@pytest.mark.asyncio +async def test_a_dead_flush_task_is_restarted_by_the_next_event(): + """A cancelled or crashed flusher would otherwise leave the queue growing forever.""" + logger = _logger(FakeUploadClient()) + logger._periodic_flush_task.cancel() + await asyncio.sleep(0) # let the cancellation land, so the task reports itself done + + await logger.async_log_success_event(_event("after-cancel"), None, None, None) + + assert logger._periodic_flush_task is not None + assert not logger._periodic_flush_task.done() + logger._periodic_flush_task.cancel() + + +@pytest.mark.asyncio +async def test_a_failure_while_queueing_never_breaks_the_request(): + """Logging sits on the request path, so a fault here must not surface to the caller.""" + + class ExplodingQueue(list): + def append(self, _item): + raise RuntimeError("queue is broken") + + upload_client = FakeUploadClient() + logger = _logger(upload_client) + logger.log_queue = ExplodingQueue() + + await logger.async_log_success_event(_event("boom"), None, None, None) + + logger.log_queue = [] + await logger.async_log_success_event(_event("after-the-fault"), None, None, None) + assert [record["id"] for record in logger.log_queue] == ["after-the-fault"] + + +@pytest.mark.asyncio +async def test_a_flush_with_nothing_queued_uploads_nothing(): + upload_client = FakeUploadClient() + logger = _logger(upload_client) + + await logger.async_send_batch() + + assert upload_client.bodies == [] + + +@pytest.mark.asyncio +async def test_the_idle_ping_is_skipped_when_the_key_was_removed(monkeypatch, caplog): + """A key pulled mid-flight must not turn the periodic flush into an exception.""" + monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_present") + logger = PointFiveLogger(params=PointFiveInitParams()) + monkeypatch.delenv("POINTFIVE_API_KEY") + + with caplog.at_level(logging.WARNING): + await logger.flush_queue() + + assert "liveness ping skipped" in caplog.text + logger._periodic_flush_task.cancel() + + +@pytest.mark.asyncio +async def test_a_full_batch_stands_down_while_a_flush_is_already_running(): + """ + Under load every event landing mid-upload also crosses the batch threshold. + + Letting each one flush turns a single burst into a stream of tiny objects, which is + what batching exists to avoid, so a full batch defers to the flush already running. + """ + upload_client = FakeUploadClient() + logger = _logger(upload_client, batch_size=2) + release = asyncio.Event() + finish_upload = upload_client.upload + + async def held_upload(body: bytes): + await release.wait() + return await finish_upload(body) + + upload_client.upload = held_upload + logger.log_queue.extend(_event(f"first-{index}")["standard_logging_object"] for index in range(2)) + + flushing = asyncio.create_task(logger.flush_queue()) + await asyncio.sleep(0) + + # Bounded: without the guard these block on the flush lock the held upload owns. + for index in range(6): + await asyncio.wait_for(logger.async_log_success_event(_event(f"mid-{index}"), None, None, None), timeout=2) + + assert upload_client.bodies == [] + + release.set() + await flushing + + assert len(upload_client.bodies) == 1 + assert [record["id"] for record in upload_client.records()] == ["first-0", "first-1"] + assert [record["id"] for record in logger.log_queue] == [f"mid-{index}" for index in range(6)]