Merge pull request #38509 from yinonkahta-p5/litellm_pointfive_logger

feat(pointfive): add the pointfive logging integration

litellm/__init__.py

@@ -45,9 +45,11 @@ 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.litellm_core_utils.core_helpers import drop_params_env_flag
+from litellm.types.integrations.pointfive import PointFiveInitParams
 from litellm._logging import (
     set_verbose,
     _turn_on_debug,
@@ -154,6 +156,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "smtp_email",
     "deepeval",
     "s3_v2",
+    "pointfive",
     "aws_sqs",
     "vector_store_pre_call_hook",
     "dotprompt",
@@ -439,6 +442,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

@@ -378,6 +378,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/api/v1/ingestion",
+        "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,304 @@
+"""
+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.litellm_core_utils.redact_messages import (
+    redacted_standard_logging_payload,
+    should_redact_message_logging,
+)
+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 a config value that may name a secret, in any shape the secret manager accepts.
+
+    A reference that resolves to nothing stays unresolved rather than falling back to its own
+    text, so an unset ``os.environ/NAME`` reports a missing key instead of being sent as one.
+    """
+    if value is None:
+        return None
+    resolved: Final = get_secret_str(value)
+    if resolved:
+        return resolved
+    return None if value.startswith(_ENV_REFERENCE_PREFIX) else 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),
+        )
+        self._flushing: bool = False
+        self._batch_flush_task: asyncio.Task[None] | None = None
+        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())
+
+    def _start_batch_flush_task(self) -> None:
+        """
+        Upload a full batch in the background, so no request waits on PointFive.
+
+        Awaiting it here put the upload, its retries and their backoff on the caller's
+        path, and a hung api held a response open for as long as the attempts took.
+        """
+        if self._batch_flush_task is not None and not self._batch_flush_task.done():
+            return
+        try:
+            loop: Final = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        self._batch_flush_task = loop.create_task(self.flush_queue(skip_if_flushing=True))
+
+    def _flush_task_is_alive(self) -> bool:
+        """A task whose loop has been closed never runs again, yet never reports itself done."""
+        task: Final = self._periodic_flush_task
+        return task is not None and not task.done() and not task.get_loop().is_closed()
+
+    async def periodic_flush(self) -> None:
+        """
+        Report in straight away, then flush on the interval as usual.
+
+        The inherited loop sleeps first, so a proxy that has just loaded the callback says
+        nothing for a whole interval, five minutes by default. PointFive shows the integration
+        as still waiting for its first call for all that time, which reads as a broken setup
+        rather than an idle one. An empty queue makes this first cycle a ping, so a proxy with
+        no traffic yet announces itself without uploading an object that holds no records.
+        """
+        await self.flush_queue(skip_if_flushing=True)
+        await super().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 not self._flush_task_is_alive():
+                self._periodic_flush_task = self._start_periodic_flush_task()
+
+            record: Final = self._record_for(kwargs)
+            if record is None:
+                verbose_logger.debug("pointfive: event carried no standard_logging_object, skipping")
+                return
+
+            self.log_queue.append(record)
+            self._drop_overflow()
+            if len(self.log_queue) >= self.batch_size:
+                self._start_batch_flush_task()
+        except Exception:  # noqa: BLE001  # logging must never break the request path
+            verbose_logger.exception("pointfive: failed to queue an event")
+
+    def _record_for(self, kwargs: Mapping[str, object]) -> Mapping[str, object] | None:
+        """
+        The record to buffer, redacted the way the framework would have redacted it.
+
+        A success reaches a callback already redacted, an async failure does not, so both
+        the excluded-field list and this callback's own setting are applied here, then the
+        global, per-request and header settings that only the framework's predicate knows.
+        """
+        details: Final = self.redact_standard_logging_payload_from_model_call_details(
+            dict(kwargs)  # mutable-ok: both framework helpers take the call details as a dict
+        )
+        payload: Final = details.get("standard_logging_object")
+        if not isinstance(payload, dict):
+            return None
+        if should_redact_message_logging(details):
+            return redacted_standard_logging_payload(payload)
+        return payload
+
+    def _drop_overflow(self) -> None:
+        """
+        Hold the queue to its cap as records arrive, not only after a flush has failed.
+
+        Never while a flush is running: it holds a snapshot taken by length, and trimming
+        the front underneath it would make the post-flush drain remove records that arrived
+        during the upload and were never sent. The next arrival after the flush trims.
+        """
+        if self._flushing:
+            return
+        overflow: Final = len(self.log_queue) - self.max_queue_size
+        if overflow <= 0:
+            return
+        del self.log_queue[:overflow]
+        verbose_logger.warning("pointfive: queue over %s records, dropped %s oldest", self.max_queue_size, overflow)
+
+    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, and the loop's opening cycle, pass. 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 rest of the batch
+        for the next flush; the records already shipped or already refused leave the queue
+        first, so a retry re-sends at most the object that failed. A rejection the server
+        will refuse again drops that object, since holding it would block every record
+        queued behind it.
+        """
+        pending: Final = tuple(self.log_queue)
+        if not pending:
+            return
+
+        client: Final = self.upload_client
+        chunks: Final = chunk_lines(serialize_records(pending), self.max_batch_bytes)
+        for index, chunk in enumerate(chunks):
+            outcome = await client.upload(await encode_lines(chunk))
+            if not isinstance(outcome, PointFiveUploadFailure):
+                continue
+            if outcome.retryable:
+                del self.log_queue[: sum(len(shipped) for shipped in chunks[:index])]
+                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,194 @@
+"""
+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
+from types import MappingProxyType
+from typing import Final
+
+import httpx
+from pydantic import BaseModel, Field, ValidationError
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.types.integrations.pointfive import (
+    RETRYABLE_UPLOAD_STATUS_CODES,
+    PointFiveUploadFailure,
+    PointFiveUploadTarget,
+)
+
+UPLOAD_KIND: Final = "LITELLM"
+UPLOAD_URL_PATH: Final = "/upload-url"
+PING_PATH: Final = "/ping"
+PUT_HEADERS: Final = MappingProxyType({"Content-Type": "application/x-ndjson", "Content-Encoding": "gzip"})
+
+
+class _PresignRequest(BaseModel):
+    kind: str = UPLOAD_KIND
+    byte_count: int = Field(serialization_alias="byteCount")
+
+
+class _PingRequest(BaseModel):
+    kind: str = UPLOAD_KIND
+
+
+class _TargetPayload(BaseModel):
+    upload_url: str = Field(alias="uploadUrl")
+    object_key: str = Field(alias="objectKey")
+
+
+class _ErrorPayload(BaseModel):
+    error: str = ""
+
+
+class PointFiveUploadError(Exception):
+    """A batch could not be uploaded and the failure is worth retrying."""
+
+
+def _failure_for(response: httpx.Response, what: str) -> PointFiveUploadFailure:
+    detail: Final = f"{what} returned {response.status_code}"
+    reason: Final = _refusal_reason(response.text)
+    return PointFiveUploadFailure(
+        f"{detail}, {reason}" if reason else detail,
+        retryable=response.status_code in RETRYABLE_UPLOAD_STATUS_CODES,
+    )
+
+
+def _refusal_reason(body: str) -> str:
+    try:
+        return _ErrorPayload.model_validate_json(body).error
+    except ValidationError:
+        return ""
+
+
+def _parse_target(body: str) -> PointFiveUploadTarget | PointFiveUploadFailure:
+    try:
+        target: Final = _TargetPayload.model_validate_json(body)
+    except ValidationError:
+        return PointFiveUploadFailure("pointfive api returned an unreadable body", retryable=False)
+    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,
+        validate_upload_url: Callable[[str], tuple[str, str]] = validate_url,
+    ) -> None:
+        self.api_key: Final = api_key
+        self.api_url: Final = api_url.rstrip("/")
+        self.http_client: Final = http_client
+        self.max_retries: Final = max_retries
+        self.sleep: Final = sleep
+        self.validate_upload_url: Final = validate_upload_url
+
+    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._post(PING_PATH, _PingRequest())
+        if isinstance(body, PointFiveUploadFailure):
+            return body
+        return None
+
+    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._post(UPLOAD_URL_PATH, _PresignRequest(byte_count=byte_count))
+        if isinstance(body, PointFiveUploadFailure):
+            return body
+        return _parse_target(body)
+
+    async def _post(self, path: str, request: BaseModel) -> str | PointFiveUploadFailure:
+        """POST one JSON request to the PointFive ingestion API and return its raw body."""
+        try:
+            response: Final = await self.http_client.post(
+                self.api_url + path,
+                json=request.model_dump(by_alias=True),
+                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, "pointfive api")
+        except Exception as e:  # noqa: BLE001  # a transport fault is worth another attempt
+            return PointFiveUploadFailure(f"pointfive api unreachable: {type(e).__name__}", 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.
+
+        The server chose that URL, so it is treated like any other externally supplied
+        destination: the host is checked against blocked networks before connecting, and
+        a redirect is refused rather than followed. A presigned URL never legitimately
+        redirects, and following one would let a compromised endpoint point the proxy at
+        an internal service.
+        """
+        destination: Final = self._destination(target.upload_url)
+        if isinstance(destination, PointFiveUploadFailure):
+            return destination
+        url, host = destination
+        headers: Final = dict(PUT_HEADERS, Host=host) if host else dict(PUT_HEADERS)  # mutable-ok: put wants dict
+        try:
+            await self.http_client.put(url, data=body, headers=headers, follow_redirects=False)
+        except httpx.HTTPStatusError as e:
+            if e.response.is_redirect:
+                return PointFiveUploadFailure(
+                    f"presigned upload redirected with {e.response.status_code}, refusing to follow", retryable=False
+                )
+            return _failure_for(e.response, "presigned upload")
+        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
+
+    def _destination(self, upload_url: str) -> tuple[str, str | None] | PointFiveUploadFailure:
+        if not getattr(litellm, "user_url_validation", True):
+            return upload_url, None
+        try:
+            return self.validate_upload_url(upload_url)
+        except SSRFError as e:
+            return PointFiveUploadFailure(f"presigned upload url refused: {e}", retryable=False)

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

@@ -182,6 +182,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
@@ -4376,6 +4377,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):
@@ -5064,6 +5073,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/litellm_core_utils/redact_messages.py

@@ -162,6 +162,19 @@ def _redact_responses_api_output_dict(output_items, redacted_str: str):
             output_item["arguments"] = redacted_str
 
 
+def redacted_standard_logging_payload(payload: Mapping[str, object]) -> Mapping[str, object]:
+    """
+    Return a copy of a ``StandardLoggingPayload`` with its messages and response redacted.
+
+    The success path redacts through ``perform_redaction`` before a callback ever sees the
+    payload, but the failure path does not, so a callback that batches both has to redact
+    the ones it is handed.
+    """
+    redacted: Final = copy.deepcopy(dict(payload))  # mutable-ok: redacted in place below
+    _redact_standard_logging_object({"standard_logging_object": redacted})  # mutable-ok: the callee's shape
+    return redacted
+
+
 def _redact_standard_logging_object(model_call_details: dict):
     """Redact messages and response inside standard_logging_object if present."""
     standard_logging_object: Final = model_call_details.get("standard_logging_object")

litellm/llms/custom_httpx/http_handler.py

@@ -751,7 +751,9 @@ class AsyncHTTPHandler:
         timeout: float | httpx.Timeout | None = None,
         stream: bool = False,
         content: _RequestContent | None = None,
+        follow_redirects: bool | None = None,
     ):
+        _follow_redirects: Final = follow_redirects if follow_redirects is not None else USE_CLIENT_DEFAULT
         try:
             if timeout is None:
                 timeout = self.timeout
@@ -769,22 +771,30 @@ class AsyncHTTPHandler:
                 timeout=timeout,
                 content=request_content,
             )
-            response: Final = await self.client.send(req)
+            response: Final = await self.client.send(req, follow_redirects=_follow_redirects)
             response.raise_for_status()
             return response
         except (httpx.RemoteProtocolError, httpx.ConnectError):
             # Retry the request with a new session if there is a connection error
             new_client: Final = self.create_client(timeout=timeout, event_hooks=self.event_hooks)
             try:
-                return await self.single_connection_post_request(
-                    url=url,
-                    client=new_client,
-                    data=data,
+                retry_data, retry_content = _prepare_request_data_and_content(data, content)
+                retry: Final = new_client.build_request(
+                    "PUT",
+                    url,
+                    data=retry_data,
                     json=json,
                     params=params,
                     headers=headers,
-                    stream=stream,
+                    timeout=timeout,
+                    content=retry_content,
                 )
+                retried: Final = await new_client.send(retry, stream=stream, follow_redirects=_follow_redirects)
+                try:
+                    retried.raise_for_status()
+                except httpx.HTTPStatusError as retried_error:
+                    await _raise_masked_async_error(retried_error, stream)
+                return retried
             finally:
                 await new_client.aclose()
         except httpx.TimeoutException as e:

litellm/proxy/_types.py

@@ -3726,6 +3726,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

@@ -209,6 +209,7 @@ services = (
         "arize",
         "galileo",
         "newrelic",
+        "pointfive",
         "sqs",
     ]
     | str
@@ -296,6 +297,7 @@ async def health_services_endpoint(
             "arize",
             "galileo",
             "newrelic",
+            "pointfive",
             "sqs",
         ]:
             raise HTTPException(
@@ -320,7 +322,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",
@@ -412,6 +414,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,45 @@
+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/api/v1/ingestion"
+
+
+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. ``batch_size`` also bounds how much a busy
+    proxy holds in memory between flushes, so it stays modest. ``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=1_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,759 @@
+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 DEFAULT_API_URL, 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)
+
+    await _settle(logger)
+    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)
+
+    await _settle(logger)
+    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)
+
+    await _settle(logger)
+    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)
+
+    await _settle(logger)
+    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)
+
+    await _settle(logger)
+    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)
+
+    await _settle(logger)
+    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 == 1_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/api/v1/ingestion"
+
+
+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/api/v1/ingestion")
+
+    logger = PointFiveLogger()
+
+    assert logger.upload_client.api_url == "https://api.staging.pointfive.co/api/v1/ingestion"
+
+
+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/api/v1/ingestion")
+
+    logger = PointFiveLogger(
+        params=PointFiveInitParams(api_key="p5tu_from_config", api_url="https://from-config.example/api/v1/ingestion")
+    )
+
+    assert logger.upload_client.api_key == "p5tu_from_config"
+    assert logger.upload_client.api_url == "https://from-config.example/api/v1/ingestion"
+
+
+def test_a_missing_api_key_fails_at_startup_not_at_the_first_flush(monkeypatch):
+    monkeypatch.delenv("POINTFIVE_API_KEY", raising=False)
+
+    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)
+
+    await _settle(logger)
+    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/api/v1/ingestion")
+    logger = PointFiveLogger(params=PointFiveInitParams())
+
+    assert logger.upload_client.api_key == "p5tu_first"
+    assert logger.upload_client.api_url == "https://first.example.invalid/api/v1/ingestion"
+
+    monkeypatch.setenv("POINTFIVE_API_KEY", "p5tu_second")
+    monkeypatch.setenv("POINTFIVE_API_URL", "https://second.example.invalid/api/v1/ingestion")
+
+    assert logger.upload_client.api_key == "p5tu_second"
+    assert logger.upload_client.api_url == "https://second.example.invalid/api/v1/ingestion"
+
+
+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()
+    # No periodic task: this test drives the flushes itself, and the loop's opening cycle
+    # would otherwise ship the queue it seeds below.
+    logger = PointFiveLogger(
+        params=PointFiveInitParams(batch_size=2),
+        upload_client=upload_client,
+        start_periodic_flush=False,
+    )
+    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)]
+
+
+async def _settle(logger) -> None:
+    """
+    Wait out the flush a full batch schedules, the way the proxy's loop would.
+
+    The upload runs off the request path now, and either the batch task or the periodic
+    loop can be the one carrying it, so this waits for whichever is in flight to finish.
+    """
+    for _ in range(200):
+        await asyncio.sleep(0.001)
+        task = logger._batch_flush_task
+        if task is not None and not task.done():
+            await task
+        if not logger._flushing:
+            return
+    raise AssertionError("the flush never finished")
+
+
+async def _until(done: Callable[[], bool], ticks: int = 400) -> None:
+    """Wait for a condition the flush path reaches only after gzip finishes on a worker thread."""
+    for _ in range(ticks):
+        if done():
+            return
+        await asyncio.sleep(0.005)
+    raise AssertionError("condition never became true")
+
+
+@pytest.mark.asyncio
+async def test_a_new_logger_announces_itself_without_waiting_for_the_interval():
+    """
+    Configuring the callback must make the integration connect, with no traffic and no test click.
+
+    The inherited loop sleeps a whole interval before its first flush, which left a freshly
+    configured proxy silent for five minutes and the integration looking unconfigured.
+    """
+    upload_client = FakeUploadClient()
+    logger = _logger(upload_client)
+
+    await asyncio.sleep(0)  # let the flush task reach its first cycle
+
+    assert upload_client.pings == 1
+    assert upload_client.bodies == []
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_the_first_cycle_ships_records_rather_than_announcing():
+    """Announcing is only for an empty queue: records already waiting must go out as an upload."""
+    upload_client = FakeUploadClient()
+    logger = PointFiveLogger(
+        params=PointFiveInitParams(batch_size=100),
+        upload_client=upload_client,
+        start_periodic_flush=False,
+    )
+    logger.log_queue.append(_event("queued-before-start")["standard_logging_object"])
+
+    logger._periodic_flush_task = logger._start_periodic_flush_task()
+    await _until(lambda: bool(upload_client.bodies))
+
+    assert upload_client.pings == 0
+    assert [record["id"] for record in upload_client.records()] == ["queued-before-start"]
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_failed_request_ships_redacted_when_message_logging_is_off():
+    """
+    Failure events skip the framework's redaction, so the callback has to redact what it buffers.
+
+    Without this the prompt of every failed request reaches PointFive in full, even though
+    the integration was configured not to send message content.
+    """
+    upload_client = FakeUploadClient()
+    logger = _logger(upload_client, batch_size=1, turn_off_message_logging=True)
+    event = _event("failed-request")
+    event["standard_logging_object"]["messages"] = [{"role": "user", "content": "my secret prompt"}]
+    event["standard_logging_object"]["response"] = "the secret answer"
+
+    await logger.async_log_failure_event(event, None, None, None)
+    await _settle(logger)
+
+    shipped = upload_client.records()[0]
+    assert "my secret prompt" not in json.dumps(shipped)
+    assert "the secret answer" not in json.dumps(shipped)
+    assert event["standard_logging_object"]["messages"][0]["content"] == "my secret prompt"
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_dead_loop_does_not_strand_the_flusher():
+    """A task whose loop was closed never runs and never reports done, so it must be replaced."""
+    logger = PointFiveLogger(params=PointFiveInitParams(), upload_client=FakeUploadClient(), start_periodic_flush=False)
+    stranded_loop = asyncio.new_event_loop()
+    forever = asyncio.sleep(3600)
+    logger._periodic_flush_task = stranded_loop.create_task(forever)
+    stranded_loop.close()
+    forever.close()
+
+    await logger.async_log_success_event(_event("after-loop-close"), None, None, None)
+
+    assert logger._periodic_flush_task is not None
+    assert logger._periodic_flush_task.get_loop() is asyncio.get_running_loop()
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_retry_does_not_resend_objects_that_already_landed():
+    """
+    A failure part way through a multi-object flush used to hand the whole batch back.
+
+    Every record already shipped, and every record already refused for good, went out
+    again on the next flush, so PointFive received duplicates of both.
+    """
+    upload_client = FakeUploadClient(outcomes=[OBJECT_KEY, PointFiveUploadFailure("service busy", retryable=True)])
+    logger = PointFiveLogger(
+        params=PointFiveInitParams(max_batch_bytes=1),  # one record per object
+        upload_client=upload_client,
+        start_periodic_flush=False,
+    )
+    logger.log_queue.extend(_event(request_id)["standard_logging_object"] for request_id in ("first", "second"))
+
+    with pytest.raises(PointFiveUploadError):
+        await logger.async_send_batch()
+
+    assert [record["id"] for record in logger.log_queue] == ["second"]
+
+
+@pytest.mark.asyncio
+async def test_the_queue_stops_growing_at_its_cap_without_waiting_for_a_failure():
+    """The base class trims only after a failed send, so a proxy that keeps flushing never trims."""
+    logger = _logger(FakeUploadClient(), batch_size=10_000)
+    logger.max_queue_size = 3
+
+    for request_id in ("a", "b", "c", "d", "e"):
+        await logger.async_log_success_event(_event(request_id), None, None, None)
+
+    assert [record["id"] for record in logger.log_queue] == ["c", "d", "e"]
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_failed_request_honours_the_global_redaction_setting(monkeypatch):
+    """
+    Redaction can be turned on globally or per request, not only on this callback.
+
+    The async failure path hands the payload over untouched, so a tenant could trigger a
+    provider failure and ship prompts that the operator had already asked to be redacted.
+    """
+    import litellm
+
+    monkeypatch.setattr(litellm, "turn_off_message_logging", True)
+    upload_client = FakeUploadClient()
+    logger = _logger(upload_client, batch_size=1)
+    event = _event("globally-redacted")
+    event["standard_logging_object"]["messages"] = [{"role": "user", "content": "my secret prompt"}]
+
+    await logger.async_log_failure_event(event, None, None, None)
+
+    await _settle(logger)
+    assert "my secret prompt" not in json.dumps(upload_client.records()[0])
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_excluded_fields_are_dropped_from_a_failed_request(monkeypatch):
+    """standard_logging_payload_excluded_fields drops a field entirely; failures skipped it too."""
+    import litellm
+
+    monkeypatch.setattr(litellm, "standard_logging_payload_excluded_fields", ["messages"])
+    upload_client = FakeUploadClient()
+    logger = _logger(upload_client, batch_size=1, turn_off_message_logging=True)
+    event = _event("field-excluded")
+    event["standard_logging_object"]["messages"] = [{"role": "user", "content": "my secret prompt"}]
+
+    await logger.async_log_failure_event(event, None, None, None)
+    await _settle(logger)
+
+    shipped = upload_client.records()[0]
+    assert "messages" not in shipped
+    assert shipped["id"] == "field-excluded"
+    logger._periodic_flush_task.cancel()
+
+
+def _held_upload(upload_client: FakeUploadClient, release: asyncio.Event) -> None:
+    finish = upload_client.upload
+
+    async def held(body: bytes):
+        await release.wait()
+        return await finish(body)
+
+    upload_client.upload = held
+
+
+@pytest.mark.asyncio
+async def test_a_full_batch_does_not_hold_the_request():
+    """
+    The upload belongs off the request path.
+
+    Awaiting it inline meant a hung PointFive api held the caller's response open for as
+    long as the attempts and their backoff took.
+    """
+    upload_client = FakeUploadClient()
+    release = asyncio.Event()
+    _held_upload(upload_client, release)
+    logger = _logger(upload_client, batch_size=1)
+
+    await asyncio.wait_for(logger.async_log_success_event(_event("first"), None, None, None), timeout=2)
+
+    assert upload_client.bodies == []
+    release.set()
+    await _settle(logger)
+    assert [record["id"] for record in upload_client.records()] == ["first"]
+    logger._periodic_flush_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_records_arriving_during_a_flush_survive_the_queue_cap():
+    """
+    The flush drains by count, so trimming the front underneath it loses records.
+
+    Records that arrived while the upload was in flight would be deleted by that drain
+    without ever being sent.
+    """
+    upload_client = FakeUploadClient()
+    release = asyncio.Event()
+    _held_upload(upload_client, release)
+    logger = PointFiveLogger(
+        params=PointFiveInitParams(batch_size=2),
+        upload_client=upload_client,
+        start_periodic_flush=False,
+    )
+    logger.max_queue_size = 2
+    logger.log_queue.extend(_event(request_id)["standard_logging_object"] for request_id in ("a", "b"))
+
+    flushing = asyncio.create_task(logger.flush_queue())
+    await asyncio.sleep(0.01)
+    for request_id in ("c", "d", "e"):
+        await logger.async_log_success_event(_event(request_id), None, None, None)
+    release.set()
+    await flushing
+
+    assert [record["id"] for record in upload_client.records()] == ["a", "b"]
+    assert [record["id"] for record in logger.log_queue] == ["c", "d", "e"]
+    logger._periodic_flush_task.cancel()
+
+
+def test_an_unset_env_reference_is_never_used_as_the_key(monkeypatch):
+    """
+    A config that names a missing variable has no key, and must say so.
+
+    Falling back to the reference text sent the literal "os.environ/NAME" as the bearer
+    token, so the callback started and every upload was rejected for the wrong reason.
+    """
+    monkeypatch.delenv("POINTFIVE_API_KEY", raising=False)
+    monkeypatch.delenv("POINTFIVE_MISSING_KEY", raising=False)
+
+    with pytest.raises(ValueError, match="requires an api key"):
+        PointFiveLogger(
+            params=PointFiveInitParams(api_key="os.environ/POINTFIVE_MISSING_KEY"),
+            start_periodic_flush=False,
+        )
+
+
+def test_an_unset_url_reference_falls_back_to_the_public_endpoint(monkeypatch):
+    """An unresolved url reference must not become the destination the proxy uploads to."""
+    from litellm.integrations.pointfive.logger import _resolved_api_url
+
+    monkeypatch.delenv("POINTFIVE_API_URL", raising=False)
+    monkeypatch.delenv("POINTFIVE_MISSING_URL", raising=False)
+
+    assert _resolved_api_url(PointFiveInitParams(api_url="os.environ/POINTFIVE_MISSING_URL")) == DEFAULT_API_URL

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,378 @@
+import json
+from collections.abc import Sequence
+
+import httpx
+import pytest
+
+import litellm
+from litellm.integrations.pointfive.upload_client import PointFiveUploadClient
+from litellm.litellm_core_utils.url_utils import validate_url
+from litellm.types.integrations.pointfive import PointFiveUploadFailure
+
+API_URL = "https://api.pointfive.co/api/v1/ingestion"
+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, {"uploadUrl": UPLOAD_URL, "objectKey": OBJECT_KEY, "expiresAt": "2026-08-25T14:35:00Z"}
+    )
+
+
+def _response(status_code: int, payload: object) -> httpx.Response:
+    return httpx.Response(status_code, text=json.dumps(payload))
+
+
+def _refused(status_code: int, error: str) -> httpx.Response:
+    """The body PointFive sends with every refusal."""
+    return _response(status_code, {"success": False, "error": error})
+
+
+def _accepted() -> httpx.Response:
+    return httpx.Response(200, text="")
+
+
+def _no_content() -> httpx.Response:
+    return httpx.Response(204, 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, follow_redirects=None, **_):
+        self.put_calls.append(
+            {"url": url, "data": data, "headers": headers or {}, "follow_redirects": follow_redirects}
+        )
+        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 >= 300:
+        request = httpx.Request("POST", url)
+        raise httpx.HTTPStatusError(
+            "boom",
+            request=request,
+            response=httpx.Response(result.status_code, text=result.text, headers=result.headers),
+        )
+    return result
+
+
+async def _no_backoff(_seconds: float) -> None:
+    return None
+
+
+def _trusting_validator(url: str) -> tuple[str, str]:
+    """Stands in for validate_url so the fixture hosts need no DNS; the SSRF tests use the real one."""
+    return url, httpx.URL(url).host
+
+
+def _client(
+    http_client: FakeHTTPClient,
+    max_retries: int = 3,
+    api_url: str = API_URL,
+    validate_upload_url=_trusting_validator,
+) -> PointFiveUploadClient:
+    return PointFiveUploadClient(
+        api_key="p5tu_testkey",
+        api_url=api_url,
+        http_client=http_client,
+        max_retries=max_retries,
+        sleep=_no_backoff,
+        validate_upload_url=validate_upload_url,
+    )
+
+
+def _presigned_for(upload_url: str) -> httpx.Response:
+    return _response(200, {"uploadUrl": upload_url, "objectKey": OBJECT_KEY, "expiresAt": "2026-08-25T14:35:00Z"})
+
+
+@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["url"] == "https://api.pointfive.co/api/v1/ingestion/upload-url"
+    assert call["headers"]["Authorization"] == "Bearer p5tu_testkey"
+    assert call["json"] == {"kind": "LITELLM", "byteCount": len(BODY)}
+
+
+@pytest.mark.asyncio
+async def test_a_trailing_slash_on_the_api_url_is_tolerated():
+    """A pasted URL often ends in a slash; it must not produce a double slash in the path."""
+    http_client = FakeHTTPClient()
+
+    await _client(http_client, api_url=API_URL + "/").upload(BODY)
+
+    assert http_client.presign_calls[0]["url"] == "https://api.pointfive.co/api/v1/ingestion/upload-url"
+
+
+@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_upload_pins_the_host_and_never_follows_a_redirect():
+    http_client = FakeHTTPClient()
+
+    await _client(http_client).upload(BODY)
+
+    call = http_client.put_calls[0]
+    assert call["headers"]["Host"] == "uploads.example.invalid"
+    assert call["follow_redirects"] is False
+
+
+@pytest.mark.asyncio
+async def test_a_redirected_upload_is_refused_rather_than_followed():
+    """A presigned URL never redirects legitimately; following one is how a bad endpoint reaches inside."""
+    http_client = FakeHTTPClient(put=[httpx.Response(301, headers={"location": "http://169.254.169.254/"})])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure(
+        "presigned upload redirected with 301, refusing to follow", retryable=False
+    )
+    assert len(http_client.put_calls) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "upload_url",
+    [
+        "http://169.254.169.254/latest/meta-data",
+        "https://10.0.0.7/internal/bucket/object",
+        "http://127.0.0.1:9000/bucket/object",
+    ],
+)
+async def test_an_upload_url_inside_the_network_is_refused_before_any_bytes_leave(upload_url):
+    http_client = FakeHTTPClient(presign=[_presigned_for(upload_url)])
+
+    outcome = await _client(http_client, validate_upload_url=validate_url).upload(BODY)
+
+    assert isinstance(outcome, PointFiveUploadFailure)
+    assert not outcome.retryable
+    assert outcome.detail.startswith("presigned upload url refused: ")
+    assert http_client.put_calls == []
+
+
+@pytest.mark.asyncio
+async def test_an_operator_can_switch_destination_validation_off(monkeypatch):
+    """litellm.user_url_validation is the proxy-wide switch every SSRF guard honours."""
+    monkeypatch.setattr(litellm, "user_url_validation", False)
+    http_client = FakeHTTPClient(presign=[_presigned_for("http://10.0.0.7/bucket/object")])
+
+    outcome = await _client(http_client, validate_upload_url=validate_url).upload(BODY)
+
+    assert outcome == OBJECT_KEY
+    assert http_client.put_calls[0]["url"] == "http://10.0.0.7/bucket/object"
+    assert "Host" 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_the_reason_for_a_refusal_is_surfaced():
+    """A 403 means the key no longer maps to an integration; the operator needs to read why."""
+    http_client = FakeHTTPClient(presign=[_refused(403, "no integration accepts uploads from this api key")])
+
+    outcome = await _client(http_client).upload(BODY)
+
+    assert outcome == PointFiveUploadFailure(
+        "pointfive api returned 403, no integration accepts uploads from this api key", 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_too_many_requests_is_retried():
+    http_client = FakeHTTPClient(presign=[httpx.Response(429), _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_malformed_api_body_is_not_retried():
+    http_client = FakeHTTPClient(presign=[_response(200, {"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_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_ping_reports_a_live_shipper():
+    http_client = FakeHTTPClient(presign=(_no_content(),))
+
+    failure = await _client(http_client).ping()
+
+    assert failure is None
+    assert http_client.presign_calls[0]["url"] == "https://api.pointfive.co/api/v1/ingestion/ping"
+    assert http_client.presign_calls[0]["json"] == {"kind": "LITELLM"}
+
+
+@pytest.mark.asyncio
+async def test_ping_surfaces_a_revoked_key():
+    http_client = FakeHTTPClient(presign=(_refused(403, "no integration accepts uploads from this api key"),))
+
+    failure = await _client(http_client).ping()
+
+    assert failure is not None
+    assert not failure.retryable
+    assert "no integration accepts uploads from this api key" 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_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/llms/custom_httpx/test_http_handler.py

@@ -1547,3 +1547,68 @@ def test_sync_force_ipv4_https_proxy_mount_uses_handler_ca_bundle(
         handler.close()
 
     assert response.text == "ok-tls"
+
+
+@pytest.mark.asyncio
+async def test_put_can_refuse_to_follow_a_redirect():
+    """The client follows redirects by default; a caller uploading to a URL it did not choose must be able to opt out."""
+    hops: list[str] = []  # mutable-ok: the fake transport records the paths it was asked for
+
+    async def mock_handler(request: httpx.Request) -> httpx.Response:
+        hops.append(request.url.path)
+        if request.url.path == "/first":
+            return httpx.Response(302, request=request, headers={"location": "/second"})
+        return httpx.Response(200, request=request)
+
+    handler = AsyncHTTPHandler()
+    await handler.client.aclose()
+    handler.client = httpx.AsyncClient(transport=httpx.MockTransport(mock_handler), follow_redirects=True)
+    try:
+        followed = await handler.put("https://uploads.example/first", data=b"x")
+        assert followed.status_code == 200
+        assert hops == ["/first", "/second"]
+
+        hops.clear()
+        with pytest.raises(MaskedHTTPStatusError) as refused:
+            await handler.put("https://uploads.example/first", data=b"x", follow_redirects=False)
+        assert refused.value.status_code == 302
+        assert hops == ["/first"]
+    finally:
+        await handler.close()
+
+
+@pytest.mark.asyncio
+async def test_a_retried_put_stays_a_put_and_still_refuses_redirects():
+    """
+    The connection-error retry used to resend as POST through a client that follows redirects.
+
+    Storage answers a POST to a presigned PUT url with 403 or 405, so the batch looked
+    permanently rejected, and the redirect refusal the caller asked for was silently lost.
+    """
+    attempts: list[tuple[str, str]] = []  # mutable-ok: the fake transports record what they were asked for
+
+    async def refusing_transport(request: httpx.Request) -> httpx.Response:
+        attempts.append((request.method, request.url.path))
+        raise httpx.ConnectError("connection reset", request=request)
+
+    async def retry_transport(request: httpx.Request) -> httpx.Response:
+        attempts.append((request.method, request.url.path))
+        if request.url.path == "/first":
+            return httpx.Response(302, request=request, headers={"location": "/second"})
+        return httpx.Response(200, request=request)
+
+    class HandlerWithFakeRetryClient(AsyncHTTPHandler):
+        def create_client(self, *args, **kwargs) -> httpx.AsyncClient:
+            return httpx.AsyncClient(transport=httpx.MockTransport(retry_transport), follow_redirects=True)
+
+    handler = HandlerWithFakeRetryClient()
+    await handler.client.aclose()
+    handler.client = httpx.AsyncClient(transport=httpx.MockTransport(refusing_transport))
+    try:
+        with pytest.raises(MaskedHTTPStatusError) as refused:
+            await handler.put("https://uploads.example/first", data=b"x", follow_redirects=False)
+
+        assert refused.value.status_code == 302
+        assert attempts == [("PUT", "/first"), ("PUT", "/first")]
+    finally:
+        await handler.client.aclose()

tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py

@@ -1058,6 +1058,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():
     """
@@ -3131,3 +3160,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;
@@ -162,6 +163,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

@@ -49412,7 +49412,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;