Merge 323dddcb43 into 658f50663d

litellm/__init__.py

@@ -45,8 +45,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,
@@ -153,6 +155,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "smtp_email",
     "deepeval",
     "s3_v2",
+    "pointfive",
     "aws_sqs",
     "vector_store_pre_call_hook",
     "dotprompt",
@@ -435,6 +438,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

litellm/integrations/callback_configs.json

@@ -365,6 +365,27 @@
     },
     "description": "OpenTelemetry Logging Integration"
   },
+  {
+    "id": "pointfive",
+    "displayName": "PointFive",
+    "logo": "pointfive.png",
+    "supports_key_team_logging": false,
+    "dynamic_params": {
+      "POINTFIVE_API_KEY": {
+        "type": "password",
+        "ui_name": "API Key",
+        "description": "PointFive API key, used to request an upload url for each batch of logs",
+        "required": true
+      },
+      "POINTFIVE_API_URL": {
+        "type": "text",
+        "ui_name": "API URL",
+        "description": "PointFive API endpoint. Leave blank to use https://api.pointfive.co/query",
+        "required": false
+      }
+    },
+    "description": "PointFive Logging Integration"
+  },
   {
     "id": "s3",
     "displayName": "S3",

litellm/integrations/pointfive/__init__.py

@@ -0,0 +1,5 @@
+"""PointFive logging integration for LiteLLM."""
+
+from litellm.integrations.pointfive.logger import PointFiveLogger
+
+__all__ = ("PointFiveLogger",)

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)

litellm/integrations/pointfive/payload.py

@@ -0,0 +1,53 @@
+"""Turns buffered log records into the gzipped NDJSON objects that get uploaded."""
+
+import gzip
+from collections.abc import Iterator, Mapping, Sequence
+from itertools import accumulate, groupby, islice
+from typing import Final
+
+from litellm.litellm_core_utils.asyncify import asyncify
+from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
+
+_NEWLINE_BYTES: Final = 1
+
+
+def serialize_records(records: Sequence[Mapping[str, object]]) -> tuple[str, ...]:
+    """Serialize each record to one JSON line."""
+    return tuple(safe_dumps(record) for record in records)
+
+
+def _encoded_size(line: str) -> int:
+    return len(line.encode("utf-8")) + _NEWLINE_BYTES
+
+
+def _object_indices(sizes: Sequence[int], max_bytes: int) -> Iterator[int]:
+    """Number each line with the object it belongs to, opening a new one on overflow."""
+
+    def advance(state: tuple[int, int], size: int) -> tuple[int, int]:
+        index, used = state
+        return (index + 1, size) if used and used + size > max_bytes else (index, used + size)
+
+    return (index for index, _ in islice(accumulate(sizes, advance, initial=(0, 0)), 1, None))
+
+
+def chunk_lines(lines: Sequence[str], max_bytes: int) -> tuple[tuple[str, ...], ...]:
+    """
+    Group serialized lines into objects of at most ``max_bytes`` uncompressed.
+
+    A line above the bound on its own still becomes its own object. A record cannot be
+    split, and holding it back would stall every record queued behind it.
+    """
+    sizes: Final = tuple(_encoded_size(line) for line in lines)
+    numbered: Final = zip(_object_indices(sizes, max_bytes), lines, strict=True)
+    return tuple(tuple(line for _, line in group) for _, group in groupby(numbered, lambda pair: pair[0]))
+
+
+async def encode_lines(lines: Sequence[str]) -> bytes:
+    """
+    Join lines as NDJSON and gzip them off the event loop.
+
+    An object can be several megabytes, and compressing that inline would block the
+    proxy for as long as it takes.
+    """
+    compress: Final = asyncify(gzip.compress)
+    return await compress("\n".join(lines).encode("utf-8"))

litellm/integrations/pointfive/upload_client.py

@@ -0,0 +1,211 @@
+"""
+Uploads one batch to PointFive through a presigned URL.
+
+The proxy holds no cloud credentials. For every batch it asks the PointFive API for a
+single-use presigned URL and PUTs the bytes there, so the same plugin runs unchanged on
+AWS, GCP, Azure or on-prem. The server picks the object key, so the proxy never chooses
+where its data lands.
+"""
+
+import asyncio
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Final
+
+import httpx
+from pydantic import BaseModel, Field, ValidationError
+
+from litellm._logging import verbose_logger
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.types.integrations.pointfive import (
+    RETRYABLE_UPLOAD_STATUS_CODES,
+    PointFiveUploadFailure,
+    PointFiveUploadTarget,
+)
+
+UPLOAD_URL_QUERY: Final = (
+    "query UploadUrl($kind: UploadKind!, $byteCount: Int!) "
+    "{ uploadUrl(kind: $kind, byteCount: $byteCount) { uploadUrl objectKey } }"
+)
+PING_QUERY: Final = "query IntegrationPing($kind: UploadKind!) { integrationPing(kind: $kind) }"
+UPLOAD_KIND: Final = "LITELLM"
+
+
+class _PresignVariables(BaseModel):
+    kind: str = UPLOAD_KIND
+    byte_count: int = Field(serialization_alias="byteCount")
+
+
+class _PresignRequest(BaseModel):
+    variables: _PresignVariables
+    query: str = UPLOAD_URL_QUERY
+
+
+class _PingVariables(BaseModel):
+    kind: str = UPLOAD_KIND
+
+
+class _PingRequest(BaseModel):
+    variables: _PingVariables = _PingVariables()
+    query: str = PING_QUERY
+
+
+class _TargetPayload(BaseModel):
+    upload_url: str = Field(alias="uploadUrl")
+    object_key: str = Field(alias="objectKey")
+
+
+class _DataPayload(BaseModel):
+    upload_url: _TargetPayload = Field(alias="uploadUrl")
+
+
+class _ErrorPayload(BaseModel):
+    message: str = ""
+
+
+class _ErrorsPayload(BaseModel):
+    """Just the errors array, so it reads any operation's response, whatever its data shape."""
+
+    errors: tuple[_ErrorPayload, ...] | None = None
+
+
+class _EnvelopePayload(BaseModel):
+    data: _DataPayload | None = None
+
+
+class PointFiveUploadError(Exception):
+    """A batch could not be uploaded and the failure is worth retrying."""
+
+
+def _failure_for(status_code: int, detail: str) -> PointFiveUploadFailure:
+    return PointFiveUploadFailure(detail, retryable=status_code in RETRYABLE_UPLOAD_STATUS_CODES)
+
+
+def _graphql_error(body: str) -> PointFiveUploadFailure | None:
+    """GraphQL reports failures inside a 200, so the body has to be read either way."""
+    try:
+        envelope: Final = _ErrorsPayload.model_validate_json(body)
+    except ValidationError:
+        return PointFiveUploadFailure("pointfive api returned an unreadable body", retryable=False)
+    if not envelope.errors:
+        return None
+    joined: Final = ", ".join(error.message for error in envelope.errors)
+    return PointFiveUploadFailure(f"pointfive api rejected the request: {joined}", retryable=False)
+
+
+def _parse_target(body: str) -> PointFiveUploadTarget | PointFiveUploadFailure:
+    """Read the presigned target out of a GraphQL response body."""
+    error: Final = _graphql_error(body)
+    if error is not None:
+        return error
+
+    try:
+        envelope: Final = _EnvelopePayload.model_validate_json(body)
+    except ValidationError:
+        return PointFiveUploadFailure("pointfive api returned an unreadable body", retryable=False)
+    if envelope.data is None:
+        return PointFiveUploadFailure("pointfive api returned no upload url", retryable=False)
+
+    target: Final = envelope.data.upload_url
+    return PointFiveUploadTarget(upload_url=target.upload_url, object_key=target.object_key)
+
+
+class PointFiveUploadClient:
+    """Presigns and uploads one batch at a time."""
+
+    def __init__(
+        self,
+        api_key: str,
+        api_url: str,
+        http_client: AsyncHTTPHandler,
+        max_retries: int,
+        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    ) -> None:
+        self.api_key: Final = api_key
+        self.api_url: Final = api_url
+        self.http_client: Final = http_client
+        self.max_retries: Final = max_retries
+        self.sleep: Final = sleep
+
+    async def upload(self, body: bytes) -> str | PointFiveUploadFailure:
+        """
+        Upload one gzipped batch, returning the object key it landed at.
+
+        Every attempt presigns again, so a retry never reuses a URL that has expired or
+        has already been consumed.
+        """
+        for attempt in range(self.max_retries):
+            match await self._upload_once(body):
+                case PointFiveUploadFailure(retryable=True) as failure:
+                    if attempt + 1 >= self.max_retries:
+                        return PointFiveUploadFailure(
+                            f"{failure.detail}, gave up after {self.max_retries} attempts", retryable=True
+                        )
+                    await self.sleep(float(1 << attempt))
+                case outcome:
+                    return outcome
+        return PointFiveUploadFailure("max_upload_retries must be at least 1", retryable=False)
+
+    async def _upload_once(self, body: bytes) -> str | PointFiveUploadFailure:
+        target: Final = await self._presign(len(body))
+        if isinstance(target, PointFiveUploadFailure):
+            return target
+
+        rejection: Final = await self._put(target, body)
+        if rejection is not None:
+            return rejection
+
+        verbose_logger.debug("pointfive: uploaded %s gzipped bytes to %s", len(body), target.object_key)
+        return target.object_key
+
+    async def ping(self) -> PointFiveUploadFailure | None:
+        """Report that the proxy is alive when it has nothing to upload."""
+        body: Final = await self._query(_PingRequest().model_dump(by_alias=True))
+        if isinstance(body, PointFiveUploadFailure):
+            return body
+        return _graphql_error(body)
+
+    async def _presign(self, byte_count: int) -> PointFiveUploadTarget | PointFiveUploadFailure:
+        """Ask the PointFive API for a presigned URL sized to this batch."""
+        body: Final = await self._query(
+            _PresignRequest(variables=_PresignVariables(byte_count=byte_count)).model_dump(by_alias=True)
+        )
+        if isinstance(body, PointFiveUploadFailure):
+            return body
+        return _parse_target(body)
+
+    async def _query(self, request: Mapping[str, object]) -> str | PointFiveUploadFailure:
+        """POST one GraphQL request to the PointFive API and return its raw body."""
+        try:
+            response: Final = await self.http_client.post(
+                self.api_url,
+                json=dict(request),  # mutable-ok: AsyncHTTPHandler.post types json as dict
+                headers={  # mutable-ok: AsyncHTTPHandler.post types headers as dict
+                    "Authorization": f"Bearer {self.api_key}",
+                    "Content-Type": "application/json",
+                },
+            )
+        except httpx.HTTPStatusError as e:
+            return _failure_for(e.response.status_code, f"pointfive api returned {e.response.status_code}")
+        except Exception as e:  # noqa: BLE001  # a transport fault is worth another attempt
+            return PointFiveUploadFailure(f"pointfive api unreachable: {type(e).__name__}", retryable=True)
+
+        if response is None:
+            return PointFiveUploadFailure("pointfive api returned no response", retryable=True)
+        return response.text
+
+    async def _put(self, target: PointFiveUploadTarget, body: bytes) -> PointFiveUploadFailure | None:
+        """PUT the batch to the presigned URL, which carries its own authorization."""
+        try:
+            await self.http_client.put(
+                target.upload_url,
+                data=body,
+                headers={  # mutable-ok: AsyncHTTPHandler.put types headers as dict
+                    "Content-Type": "application/x-ndjson",
+                    "Content-Encoding": "gzip",
+                },
+            )
+        except httpx.HTTPStatusError as e:
+            return _failure_for(e.response.status_code, f"presigned upload returned {e.response.status_code}")
+        except Exception as e:  # noqa: BLE001  # a transport fault is worth another attempt
+            return PointFiveUploadFailure(f"presigned upload unreachable: {type(e).__name__}", retryable=True)
+        return None

litellm/litellm_core_utils/custom_logger_registry.py

@@ -43,6 +43,7 @@ from litellm.integrations.newrelic import NewRelicLogger
 from litellm.integrations.openmeter import OpenMeterLogger
 from litellm.integrations.opentelemetry import OpenTelemetry
 from litellm.integrations.opik.opik import OpikLogger
+from litellm.integrations.pointfive import PointFiveLogger
 from litellm.integrations.posthog import PostHogLogger
 from litellm.integrations.prometheus import PrometheusLogger
 from litellm.integrations.s3_v2 import S3Logger
@@ -95,6 +96,7 @@ class CustomLoggerRegistry:
         "agentops": AgentOps,
         "deepeval": DeepEvalLogger,
         "s3_v2": S3Logger,
+        "pointfive": PointFiveLogger,
         "aws_sqs": SQSLogger,
         "dynamic_rate_limiter": _PROXY_DynamicRateLimitHandler,
         "dynamic_rate_limiter_v3": _PROXY_DynamicRateLimitHandlerV3,

litellm/litellm_core_utils/litellm_logging.py

@@ -178,6 +178,7 @@ from ..integrations.lunary import LunaryLogger
 from ..integrations.newrelic import NewRelicLogger
 from ..integrations.openmeter import OpenMeterLogger
 from ..integrations.opik.opik import OpikLogger
+from ..integrations.pointfive import PointFiveLogger
 from ..integrations.posthog import PostHogLogger
 from ..integrations.prompt_layer import PromptLayerLogger
 from ..integrations.s3 import S3Logger
@@ -4281,6 +4282,14 @@ def _init_custom_logger_compatible_class(
             _s3_v2_logger: Final = S3V2Logger()
             _in_memory_loggers.append(_s3_v2_logger)
             return _s3_v2_logger
+        elif logging_integration == "pointfive":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, PointFiveLogger):
+                    return callback
+
+            _pointfive_logger: Final = PointFiveLogger()
+            _in_memory_loggers.append(_pointfive_logger)
+            return _pointfive_logger
         elif logging_integration == "aws_sqs":
             for callback in _in_memory_loggers:
                 if isinstance(callback, SQSLogger):
@@ -4917,6 +4926,10 @@ def get_custom_logger_compatible_class(
             for callback in _in_memory_loggers:
                 if isinstance(callback, S3V2Logger):
                     return callback
+        elif logging_integration == "pointfive":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, PointFiveLogger):
+                    return callback
         elif logging_integration == "aws_sqs":
             for callback in _in_memory_loggers:
                 if isinstance(callback, SQSLogger):

litellm/proxy/_types.py

@@ -3618,6 +3618,15 @@ class AllCallbacks(LiteLLMPydanticObjectBase):
         ],
     )
 
+    pointfive: CallbackOnUI = CallbackOnUI(
+        litellm_callback_name="pointfive",
+        ui_callback_name="PointFive",
+        litellm_callback_params=[  # mutable-ok: the registry field is typed list
+            "POINTFIVE_API_KEY",
+            "POINTFIVE_API_URL",
+        ],
+    )
+
 
 class SpendLogsRouterMetadata(TypedDict):
     """

litellm/proxy/health_endpoints/_health_endpoints.py

@@ -182,6 +182,7 @@ services = (
         "arize",
         "galileo",
         "newrelic",
+        "pointfive",
         "sqs",
     ]
     | str
@@ -269,6 +270,7 @@ async def health_services_endpoint(
             "arize",
             "galileo",
             "newrelic",
+            "pointfive",
             "sqs",
         ]:
             raise HTTPException(
@@ -293,7 +295,7 @@ async def health_services_endpoint(
             service == "openmeter"
             or service == "braintrust"
             or service == "generic_api"
-            or (service_in_success_callbacks and service != "langfuse")
+            or (service_in_success_callbacks and service not in ("langfuse", "pointfive"))
         ):
             _ = await litellm.acompletion(
                 model="openai/litellm-mock-response-model",
@@ -385,6 +387,27 @@ async def health_services_endpoint(
                 ),
             }
 
+        elif service == "pointfive":
+            if not _is_proxy_admin(user_api_key_dict):
+                non_admin_detail: Final[_ServiceTestErrorDetail] = {
+                    "error": "Only proxy admins can trigger the PointFive liveness ping."
+                }
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=non_admin_detail)
+            from litellm.integrations.pointfive import PointFiveLogger
+
+            try:
+                pointfive_logger: Final = PointFiveLogger(start_periodic_flush=False)
+            except ValueError as missing_key:
+                # No key configured is the answer the operator asked for, not a server error.
+                no_key: Final[_ServiceTestSuccessResponse] = {"status": "unhealthy", "message": str(missing_key)}
+                return no_key
+            response = await pointfive_logger.async_health_check()
+            pointfive_health: Final[_ServiceTestSuccessResponse] = {
+                "status": response["status"],
+                "message": (response["error_message"] if response["status"] == "unhealthy" else "PointFive is healthy")
+                or "PointFive is healthy",
+            }
+            return pointfive_health
         if service == "webhook":
             user_info: Final = CallInfo(
                 token=user_api_key_dict.token or "",

litellm/types/integrations/pointfive.py

@@ -0,0 +1,44 @@
+from dataclasses import dataclass
+from typing import Final
+
+from pydantic import Field
+
+from litellm.types.integrations.custom_logger import StandardCustomLoggerInitParams
+
+RETRYABLE_UPLOAD_STATUS_CODES: Final = frozenset({429, 500, 502, 503, 504})
+
+DEFAULT_API_URL: Final = "https://api.pointfive.co/query"
+
+
+class PointFiveInitParams(StandardCustomLoggerInitParams):
+    """
+    Params for initializing a PointFive logger on litellm.
+
+    Defaults trade freshness for fewer, larger uploads: every flush becomes one object, so
+    the interval is minutes rather than seconds. ``max_batch_bytes`` bounds how much a
+    single object may hold, which matters most when message logging is left on, since an
+    unredacted payload is orders of magnitude larger than a redacted one.
+    """
+
+    api_key: str | None = None
+    api_url: str | None = None
+    batch_size: int = Field(default=10_000, gt=0)
+    flush_interval: int = Field(default=300, gt=0)
+    max_batch_bytes: int = Field(default=8 * 1024 * 1024, gt=0)
+    max_upload_retries: int = Field(default=3, ge=1)
+
+
+@dataclass(frozen=True, slots=True)
+class PointFiveUploadTarget:
+    """A single-use presigned destination for one batch, issued by the PointFive API."""
+
+    upload_url: str
+    object_key: str
+
+
+@dataclass(frozen=True, slots=True)
+class PointFiveUploadFailure:
+    """Why a batch could not be uploaded, and whether a later attempt could still succeed."""
+
+    detail: str
+    retryable: bool

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)]

tests/test_litellm/integrations/pointfive/test_payload.py

@@ -0,0 +1,80 @@
+import gzip
+import json
+
+import pytest
+
+from litellm.integrations.pointfive.payload import chunk_lines, encode_lines, serialize_records
+
+UNBOUNDED = 10_000_000
+
+
+def test_each_record_becomes_one_json_line():
+    lines = serialize_records([{"id": "a"}, {"id": "b"}, {"id": "c"}])
+
+    assert len(lines) == 3
+    assert [json.loads(line)["id"] for line in lines] == ["a", "b", "c"]
+
+
+def test_non_serializable_values_do_not_raise():
+    """An odd payload must not kill the flush."""
+    lines = serialize_records([{"id": "a", "when": object()}])
+
+    assert json.loads(lines[0])["id"] == "a"
+
+
+def test_records_that_fit_stay_in_one_object():
+    lines = serialize_records([{"id": f"r{i}"} for i in range(50)])
+
+    assert chunk_lines(lines, UNBOUNDED) == (lines,)
+
+
+def test_objects_are_capped_by_uncompressed_size():
+    lines = serialize_records([{"id": f"r{i}", "blob": "x" * 100} for i in range(10)])
+    line_bytes = len(lines[0].encode("utf-8")) + 1
+
+    chunks = chunk_lines(lines, line_bytes * 3)
+
+    assert [len(chunk) for chunk in chunks] == [3, 3, 3, 1]
+
+
+def test_oversized_single_record_is_sent_alone_not_stalled():
+    """A record too big for the cap must still go out, or it blocks everything behind it."""
+    lines = serialize_records([{"id": "small"}, {"id": "huge", "blob": "x" * 5000}, {"id": "small2"}])
+
+    chunks = chunk_lines(lines, 200)
+
+    assert sum(len(chunk) for chunk in chunks) == 3
+    huge = [chunk for chunk in chunks if any("huge" in line for line in chunk)]
+    assert len(huge) == 1
+    assert len(huge[0]) == 1
+
+
+def test_no_records_produces_no_objects():
+    assert chunk_lines((), UNBOUNDED) == ()
+
+
+def test_every_record_appears_exactly_once():
+    lines = serialize_records([{"id": f"r{i}"} for i in range(37)])
+
+    chunks = chunk_lines(lines, len(lines[0]) * 4)
+
+    assert [line for chunk in chunks for line in chunk] == list(lines)
+
+
+@pytest.mark.asyncio
+async def test_encode_lines_round_trips_through_gzip():
+    lines = serialize_records([{"id": f"r{i}"} for i in range(5)])
+
+    encoded = await encode_lines(lines)
+
+    assert encoded[:2] == b"\x1f\x8b"
+    assert gzip.decompress(encoded).decode("utf-8") == "\n".join(lines)
+
+
+@pytest.mark.asyncio
+async def test_encode_lines_compresses_repetitive_records():
+    lines = serialize_records([{"id": f"r{i}", "model": "gpt-4o", "cost": 0.01} for i in range(200)])
+
+    encoded = await encode_lines(lines)
+
+    assert len(encoded) < len(gzip.decompress(encoded)) / 2

tests/test_litellm/integrations/pointfive/test_upload_client.py

@@ -0,0 +1,285 @@
+import json
+from collections.abc import Sequence
+
+import httpx
+import pytest
+
+from litellm.integrations.pointfive.upload_client import PointFiveUploadClient
+from litellm.types.integrations.pointfive import PointFiveUploadFailure
+
+API_URL = "https://api.pointfive.co/query"
+UPLOAD_URL = "https://uploads.example.invalid/some/object.ndjson.gz?signature=sig"
+OBJECT_KEY = "some/object.ndjson.gz"
+BODY = b"gzipped-bytes"
+
+
+def _presigned(status_code: int = 200) -> httpx.Response:
+    return _response(status_code, {"data": {"uploadUrl": {"uploadUrl": UPLOAD_URL, "objectKey": OBJECT_KEY}}})
+
+
+def _response(status_code: int, payload: object) -> httpx.Response:
+    return httpx.Response(status_code, text=json.dumps(payload))
+
+
+def _accepted() -> httpx.Response:
+    return httpx.Response(200, text="")
+
+
+class FakeHTTPClient:
+    """
+    Stands in for AsyncHTTPHandler, including its habit of raising on error statuses.
+
+    Scripted results are consumed in order, and the last one repeats, so a test that
+    cares about a single behaviour passes a single result.
+    """
+
+    def __init__(
+        self,
+        presign: Sequence[httpx.Response | Exception] | None = None,
+        put: Sequence[httpx.Response | Exception] | None = None,
+    ) -> None:
+        self.presign = list(presign) if presign else [_presigned()]  # mutable-ok: results are consumed by popping
+        self.put_results = list(put) if put else [_accepted()]  # mutable-ok: results are consumed by popping
+        self.presign_calls: list[dict] = []
+        self.put_calls: list[dict] = []
+
+    async def post(self, url, json=None, headers=None, **_):
+        self.presign_calls.append({"url": url, "json": json, "headers": headers or {}})
+        return _next_result(self.presign, url)
+
+    async def put(self, url, data=None, headers=None, **_):
+        self.put_calls.append({"url": url, "data": data, "headers": headers or {}})
+        return _next_result(self.put_results, url)
+
+
+def _next_result(results: list, url: str) -> httpx.Response:
+    result = results.pop(0) if len(results) > 1 else results[0]
+    if isinstance(result, Exception):
+        raise result
+    if result.status_code >= 400:
+        request = httpx.Request("POST", url)
+        raise httpx.HTTPStatusError("boom", request=request, response=httpx.Response(result.status_code, text="denied"))
+    return result
+
+
+async def _no_backoff(_seconds: float) -> None:
+    return None
+
+
+def _client(http_client: FakeHTTPClient, max_retries: int = 3) -> PointFiveUploadClient:
+    return PointFiveUploadClient(
+        api_key="p5tu_testkey",
+        api_url=API_URL,
+        http_client=http_client,
+        max_retries=max_retries,
+        sleep=_no_backoff,
+    )
+
+
+@pytest.mark.asyncio
+async def test_uploads_the_body_to_the_url_the_api_returned():
+    http_client = FakeHTTPClient()
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == OBJECT_KEY
+    assert http_client.put_calls[0]["url"] == UPLOAD_URL
+    assert http_client.put_calls[0]["data"] == BODY
+
+
+@pytest.mark.asyncio
+async def test_presign_request_is_authenticated_and_sized():
+    http_client = FakeHTTPClient()
+
+    await _client(http_client).upload(BODY)
+
+    call = http_client.presign_calls[0]
+    assert call["headers"]["Authorization"] == "Bearer p5tu_testkey"
+    assert call["json"]["variables"] == {"kind": "LITELLM", "byteCount": len(BODY)}
+
+
+@pytest.mark.asyncio
+async def test_no_bearer_token_is_sent_to_the_presigned_url():
+    """The URL carries its own authorization, so the api key must not travel with it."""
+    http_client = FakeHTTPClient()
+
+    await _client(http_client).upload(BODY)
+
+    assert "Authorization" not in http_client.put_calls[0]["headers"]
+
+
+@pytest.mark.asyncio
+async def test_the_object_is_declared_as_gzipped_ndjson():
+    http_client = FakeHTTPClient()
+
+    await _client(http_client).upload(BODY)
+
+    assert http_client.put_calls[0]["headers"]["Content-Encoding"] == "gzip"
+    assert http_client.put_calls[0]["headers"]["Content-Type"] == "application/x-ndjson"
+
+
+@pytest.mark.asyncio
+async def test_each_retry_presigns_again():
+    """A retry must never reuse a URL that was consumed or has expired."""
+    http_client = FakeHTTPClient(put=[httpx.Response(503), _accepted()])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == OBJECT_KEY
+    assert len(http_client.presign_calls) == 2
+    assert len(http_client.put_calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_retryable_upload_failure_gives_up_after_max_retries():
+    http_client = FakeHTTPClient(put=[httpx.Response(503)])
+
+    outcome = await _client(http_client, max_retries=2).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("presigned upload returned 503, gave up after 2 attempts", retryable=True)
+    assert len(http_client.put_calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_rejected_upload_is_not_retried():
+    http_client = FakeHTTPClient(put=[httpx.Response(403)])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("presigned upload returned 403", retryable=False)
+    assert len(http_client.put_calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_bad_api_key_is_not_retried():
+    http_client = FakeHTTPClient(presign=[httpx.Response(401)])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("pointfive api returned 401", retryable=False)
+    assert http_client.put_calls == []
+
+
+@pytest.mark.asyncio
+async def test_api_server_error_is_retried():
+    http_client = FakeHTTPClient(presign=[httpx.Response(503), _presigned()])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == OBJECT_KEY
+    assert len(http_client.presign_calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_unreachable_api_is_retried_then_reported_as_retryable():
+    http_client = FakeHTTPClient(presign=(ConnectionError("down"),))
+
+    outcome = await _client(http_client, max_retries=2).upload(BODY)
+
+    assert isinstance(outcome, PointFiveUploadFailure)
+    assert outcome.retryable
+    assert "unreachable" in outcome.detail
+    assert len(http_client.presign_calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_graphql_errors_are_not_retried():
+    http_client = FakeHTTPClient(presign=[_response(200, {"errors": [{"message": "forbidden"}]})])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("pointfive api rejected the request: forbidden", retryable=False)
+    assert http_client.put_calls == []
+
+
+@pytest.mark.asyncio
+async def test_malformed_api_body_is_not_retried():
+    http_client = FakeHTTPClient(presign=[_response(200, {"data": {"uploadUrl": {"objectKey": "k"}}})])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("pointfive api returned an unreadable body", retryable=False)
+    assert http_client.put_calls == []
+
+
+@pytest.mark.asyncio
+async def test_ping_reports_a_live_shipper():
+    http_client = FakeHTTPClient(presign=(_response(200, {"data": {"integrationPing": True}}),))
+
+    failure = await _client(http_client).ping()
+
+    assert failure is None
+    assert http_client.presign_calls[0]["json"]["variables"] == {"kind": "LITELLM"}
+    assert "integrationPing" in http_client.presign_calls[0]["json"]["query"]
+
+
+@pytest.mark.asyncio
+async def test_ping_surfaces_a_graphql_error():
+    """GraphQL answers 200 with an errors array, so the body has to be read."""
+    http_client = FakeHTTPClient(presign=(_response(200, {"errors": [{"message": "revoked"}]}),))
+
+    failure = await _client(http_client).ping()
+
+    assert failure is not None
+    assert "revoked" in failure.detail
+
+
+@pytest.mark.asyncio
+async def test_ping_surfaces_an_unreachable_api():
+    http_client = FakeHTTPClient(presign=(ConnectionError("down"),))
+
+    failure = await _client(http_client).ping()
+
+    assert failure is not None
+    assert failure.retryable
+
+
+@pytest.mark.asyncio
+async def test_a_body_that_is_not_json_is_reported_as_unreadable():
+    http_client = FakeHTTPClient(presign=(httpx.Response(200, text="<html>gateway</html>"),))
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("pointfive api returned an unreadable body", retryable=False)
+    assert http_client.put_calls == []
+
+
+@pytest.mark.asyncio
+async def test_a_response_carrying_neither_data_nor_errors_is_reported():
+    http_client = FakeHTTPClient(presign=(_response(200, {"data": None}),))
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("pointfive api returned no upload url", retryable=False)
+
+
+@pytest.mark.asyncio
+async def test_no_response_at_all_is_worth_retrying():
+    class SilentHTTPClient(FakeHTTPClient):
+        async def post(self, url, json=None, headers=None, **_):
+            return None
+
+    outcome = await _client(SilentHTTPClient()).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure(
+        "pointfive api returned no response, gave up after 3 attempts", retryable=True
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_transport_fault_on_the_upload_itself_is_retryable():
+    http_client = FakeHTTPClient(put=(ConnectionError("reset"),))
+
+    outcome = await _client(http_client, max_retries=1).upload(BODY)
+
+    assert isinstance(outcome, PointFiveUploadFailure)
+    assert outcome.retryable
+    assert "presigned upload unreachable" in outcome.detail
+
+
+@pytest.mark.asyncio
+async def test_a_client_that_may_not_try_at_all_says_so():
+    """max_upload_retries is validated as >= 1, so this guards the loop against a future zero."""
+    outcome = await _client(FakeHTTPClient(), max_retries=0).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure("max_upload_retries must be at least 1", retryable=False)

tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py

@@ -1056,6 +1056,35 @@ async def test_health_services_endpoint_galileo(status, error_message):
         mock_instance.async_health_check.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "status,error_message",
+    [
+        ("healthy", ""),
+        ("unhealthy", "PointFive authentication failed"),
+    ],
+)
+async def test_health_services_endpoint_pointfive(monkeypatch, status, error_message):
+    import litellm.integrations.pointfive as pointfive_package
+
+    mock_instance = MagicMock()
+    mock_instance.async_health_check = AsyncMock(return_value={"status": status, "error_message": error_message})
+    logger_class = MagicMock(return_value=mock_instance)
+    monkeypatch.setattr(pointfive_package, "PointFiveLogger", logger_class)
+
+    result = await health_services_endpoint(user_api_key_dict=_pointfive_admin(), service="pointfive")
+
+    if status == "healthy":
+        assert result["status"] == "healthy"
+        assert result["message"] == "PointFive is healthy"
+    else:
+        assert result["status"] == "unhealthy"
+        assert result["message"] == error_message
+    mock_instance.async_health_check.assert_awaited_once()
+    # A check that left the periodic flush running would leak a flusher per press of the ui test button.
+    logger_class.assert_called_once_with(start_periodic_flush=False)
+
+
 @pytest.mark.asyncio
 async def test_health_services_endpoint_datadog_llm_observability():
     """
@@ -2928,3 +2957,60 @@ def test_test_model_connection_accepts_image_edit_mode(monkeypatch):
 
     assert response.status_code == 200, response.text
     assert response.json()["status"] == "success"
+
+
+def _pointfive_admin() -> UserAPIKeyAuth:
+    return UserAPIKeyAuth(token="admin-token", user_id="admin-user", user_role=LitellmUserRoles.PROXY_ADMIN)
+
+
+@pytest.mark.asyncio
+async def test_health_services_endpoint_pointfive_without_a_key_is_unhealthy_not_a_server_error(monkeypatch):
+    """
+    The logger refuses to start without an api key.
+
+    That refusal is the answer the operator asked for, so it has to come back as an
+    unhealthy result rather than a 500 from the endpoint.
+    """
+    import litellm.integrations.pointfive as pointfive_package
+
+    def refuse(**_):
+        raise ValueError("pointfive logging requires an api key. Set POINTFIVE_API_KEY")
+
+    monkeypatch.setattr(pointfive_package, "PointFiveLogger", refuse)
+
+    result = await health_services_endpoint(user_api_key_dict=_pointfive_admin(), service="pointfive")
+
+    assert result["status"] == "unhealthy"
+    assert "requires an api key" in result["message"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "role",
+    [
+        LitellmUserRoles.INTERNAL_USER,
+        LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
+        LitellmUserRoles.TEAM,
+        LitellmUserRoles.CUSTOMER,
+    ],
+)
+async def test_health_services_endpoint_pointfive_blocks_non_admin(monkeypatch, role):
+    """
+    The ping travels on the proxy-wide PointFive credential and stamps liveness at PointFive.
+
+    A tenant key must not be able to keep an integration looking alive, or read back
+    account-level authentication failures through it.
+    """
+    import litellm.integrations.pointfive as pointfive_package
+    from litellm.proxy._types import ProxyException
+
+    logger_class = MagicMock()
+    monkeypatch.setattr(pointfive_package, "PointFiveLogger", logger_class)
+
+    with pytest.raises(ProxyException) as raised:
+        await health_services_endpoint(
+            user_api_key_dict=UserAPIKeyAuth(token="t", user_id="u", user_role=role), service="pointfive"
+        )
+
+    assert str(raised.value.code) == "403"
+    logger_class.assert_not_called()

tests/test_litellm/proxy/test_pointfive_dashboard_config.py

@@ -0,0 +1,47 @@
+import json
+from pathlib import Path
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+
+
+def _dashboard_configs() -> tuple[dict, ...]:
+    path = Path(litellm.__file__).parent / "integrations" / "callback_configs.json"
+    return tuple(json.loads(path.read_text()))
+
+
+def _pointfive_config() -> dict:
+    return next(config for config in _dashboard_configs() if config["id"] == "pointfive")
+
+
+def test_pointfive_appears_in_the_dashboard_callback_dropdown():
+    """The dropdown is served from callback_configs.json, so an entry only in the dashboard source is invisible."""
+    entry = _pointfive_config()
+
+    assert entry["displayName"] == "PointFive"
+    assert entry["supports_key_team_logging"] is False
+    assert entry["dynamic_params"]["POINTFIVE_API_KEY"]["required"] is True
+    assert entry["dynamic_params"]["POINTFIVE_API_KEY"]["type"] == "password"
+    assert entry["dynamic_params"]["POINTFIVE_API_URL"]["required"] is False
+
+
+def test_the_dropdown_logo_asset_exists():
+    """A logo the dashboard cannot resolve degrades silently to a letter tile."""
+    logo = _pointfive_config()["logo"]
+    repo_root = Path(litellm.__file__).parent.parent
+    asset = repo_root / "ui" / "litellm-dashboard" / "public" / "assets" / "logos" / logo
+
+    assert logo == "pointfive.png"
+    assert asset.is_file()
+
+
+def test_the_dropdown_fields_are_the_env_vars_the_logger_reads():
+    """
+    The field names are the environment variables verbatim.
+
+    The proxy would uppercase them either way, but naming them as stored means the edit
+    form finds the saved values and prefills them instead of showing blanks.
+    """
+    fields = tuple(_pointfive_config()["dynamic_params"])
+
+    assert fields == tuple(CustomLogger.get_callback_env_vars("pointfive"))

tests/test_litellm/proxy/test_pointfive_ui_callback.py

@@ -0,0 +1,15 @@
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.proxy._types import AllCallbacks
+
+
+def test_pointfive_is_offered_in_the_ui_callback_registry():
+    """The proxy ui builds its form from this registry, so an absent entry is an absent form."""
+    entry = AllCallbacks().pointfive
+
+    assert entry.litellm_callback_name == "pointfive"
+    assert entry.ui_callback_name == "PointFive"
+
+
+def test_the_ui_offers_the_two_settings_the_plugin_reads():
+    """get_callback_env_vars is what the ui renders; it must match what the logger looks up."""
+    assert tuple(CustomLogger.get_callback_env_vars("pointfive")) == ("POINTFIVE_API_KEY", "POINTFIVE_API_URL")

ui/litellm-dashboard/src/components/callback_info_helpers.tsx

@@ -9,6 +9,7 @@ import langsmithLogo from "../../public/assets/logos/langsmith.png";
 import newrelicLogo from "../../public/assets/logos/newrelic.png";
 import openmeterLogo from "../../public/assets/logos/openmeter.png";
 import otelLogo from "../../public/assets/logos/otel.png";
+import pointfiveLogo from "../../public/assets/logos/pointfive.png";
 
 interface CallbackConfig {
   id: string;
@@ -161,6 +162,17 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
     },
     description: "OpenTelemetry Logging Integration",
   },
+  {
+    id: "pointfive",
+    displayName: "PointFive",
+    logo: pointfiveLogo.src,
+    supports_key_team_logging: false,
+    dynamic_params: {
+      POINTFIVE_API_KEY: "password",
+      POINTFIVE_API_URL: "text",
+    },
+    description: "PointFive Logging Integration",
+  },
   {
     id: "s3",
     displayName: "S3",

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -48762,7 +48762,7 @@ export interface operations {
         parameters: {
             query: {
                 /** @description Specify the service being hit. */
-                service: ("slack_budget_alerts" | "langfuse" | "langfuse_otel" | "slack" | "ms_teams" | "openmeter" | "webhook" | "email" | "braintrust" | "datadog" | "datadog_llm_observability" | "generic_api" | "arize" | "galileo" | "newrelic" | "sqs") | string;
+                service: ("slack_budget_alerts" | "langfuse" | "langfuse_otel" | "slack" | "ms_teams" | "openmeter" | "webhook" | "email" | "braintrust" | "datadog" | "datadog_llm_observability" | "generic_api" | "arize" | "galileo" | "newrelic" | "pointfive" | "sqs") | string;
             };
             header?: never;
             path?: never;