Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 16 additions & 7 deletions src/tabpfn_client/sagemaker/estimator.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ def __init__(
s3_prefix: str = "async-io",
async_poll_interval_s: float = 2.0,
async_timeout_s: float = 60 * 60,
invocation_timeout_s: Optional[int] = None,
):
self.endpoint_name = endpoint_name
self.region_name = region_name
Expand All @@ -104,6 +105,11 @@ def __init__(
self.s3_prefix = s3_prefix
self.async_poll_interval_s = async_poll_interval_s
self.async_timeout_s = async_timeout_s
# Server-side cap (seconds, 1-3600) the endpoint may spend on one async
# request before it is failed. Left as the service default when None;
# raise it for long-running requests. Distinct from `async_timeout_s`,
# which only bounds how long the client polls for the result.
self.invocation_timeout_s = invocation_timeout_s

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

It is highly recommended to validate invocation_timeout_s in the constructor. SageMaker's InvocationTimeoutSeconds must be an integer between 1 and 3600.

Additionally, if invocation_timeout_s is greater than async_timeout_s, the client-side polling will time out before the server-side invocation can complete, which is a configuration mismatch.

Validating this early prevents uploading orphaned input payloads to S3 (which happens in _invoke_async before the SageMaker call is made).

Suggested change
self.invocation_timeout_s = invocation_timeout_s
self.invocation_timeout_s = invocation_timeout_s
if invocation_timeout_s is not None:
if not (1 <= invocation_timeout_s <= 3600):
raise ValueError("invocation_timeout_s must be between 1 and 3600 seconds.")
if async_timeout_s < invocation_timeout_s:
raise ValueError(
f"async_timeout_s ({async_timeout_s}) must be greater than or equal to "
f"invocation_timeout_s ({invocation_timeout_s}) to prevent client-side timeout "
f"before the server-side invocation completes."
)


@property
def _thinking_active(self) -> bool:
Expand Down Expand Up @@ -263,13 +269,16 @@ def _invoke_async(self, body_bytes: bytes) -> Dict[str, Any]:
Body=body_bytes,
ContentType="application/json",
)
resp = self._runtime_client().invoke_endpoint_async(
EndpointName=self.endpoint_name,
InputLocation=f"s3://{self.s3_bucket}/{input_key}",
ContentType="application/json",
Accept="application/json",
InferenceId=inference_id,
)
invoke_kwargs: Dict[str, Any] = {
"EndpointName": self.endpoint_name,
"InputLocation": f"s3://{self.s3_bucket}/{input_key}",
"ContentType": "application/json",
"Accept": "application/json",
"InferenceId": inference_id,
}
if self.invocation_timeout_s is not None:
invoke_kwargs["InvocationTimeoutSeconds"] = int(self.invocation_timeout_s)
resp = self._runtime_client().invoke_endpoint_async(**invoke_kwargs)
out_bucket, out_key = resp["OutputLocation"].removeprefix("s3://").split("/", 1)
fail_location = resp.get("FailureLocation")
fail_bucket, fail_key = (None, None)
Expand Down
30 changes: 30 additions & 0 deletions tests/unit/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
# Copyright (c) Prior Labs GmbH 2026.
# Licensed under the Apache License, Version 2.0
"""Isolate process-global client state between unit tests.

Several tests mutate the module-global ``options._opts`` (e.g. via
``set_access_token`` / ``ServiceClient.authorize``) or the ``TABPFN_TOKEN``
environment variable. Without isolation that state leaks into later tests and
makes ordering-dependent assertions (such as ``test_reload_opts``) fail. Snapshot
both before each test and restore afterwards so tests cannot pollute one another.
"""

import os

import pytest

import tabpfn_client.options as options


@pytest.fixture(autouse=True)
def _restore_global_client_state(): # pyright: ignore[reportUnusedFunction]
saved_opts = options._opts.model_copy(deep=True)
saved_token = os.environ.get("TABPFN_TOKEN")
try:
yield
finally:
options._opts = saved_opts
if saved_token is None:
os.environ.pop("TABPFN_TOKEN", None)
else:
os.environ["TABPFN_TOKEN"] = saved_token
58 changes: 58 additions & 0 deletions tests/unit/test_sagemaker_invocation_timeout.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
# Copyright (c) Prior Labs GmbH 2026.
# Licensed under the Apache License, Version 2.0
"""Unit tests for the async `InvocationTimeoutSeconds` pass-through on the
SageMaker estimators. boto3 is mocked, so these run without the extra."""

import json
import unittest
from io import BytesIO
from unittest.mock import MagicMock, patch

import numpy as np

from tabpfn_client.sagemaker import TabPFNClassifier


def _fake_s3_client() -> MagicMock:
s3 = MagicMock()
# `_invoke_async` reads `s3.exceptions.NoSuchKey` as an exception class.
s3.exceptions.NoSuchKey = type("NoSuchKey", (Exception,), {})
s3.get_object.return_value = {
"Body": BytesIO(json.dumps({"prediction": [[0, 1]], "metadata": {}}).encode())
}
return s3


class TestSagemakerInvocationTimeout(unittest.TestCase):
def _invoke_async_kwargs(self, invocation_timeout_s):
clf = TabPFNClassifier(
endpoint_name="ep",
region_name="us-east-1",
use_async=True,
s3_bucket="bucket",
async_poll_interval_s=0.0,
invocation_timeout_s=invocation_timeout_s,
)
runtime = MagicMock()
runtime.invoke_endpoint_async.return_value = {
"OutputLocation": "s3://bucket/out.json"
}
with (
patch.object(clf, "_runtime_client", return_value=runtime),
patch.object(clf, "_s3_client", return_value=_fake_s3_client()),
):
clf.fit(np.zeros((3, 2)), np.array([0, 1, 0]))
clf.predict(np.zeros((2, 2)))
return runtime.invoke_endpoint_async.call_args.kwargs

def test_timeout_passed_through(self):
kwargs = self._invoke_async_kwargs(1234)
self.assertEqual(kwargs["InvocationTimeoutSeconds"], 1234)

def test_timeout_omitted_by_default(self):
kwargs = self._invoke_async_kwargs(None)
self.assertNotIn("InvocationTimeoutSeconds", kwargs)
Comment on lines +52 to +54

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

It would be beneficial to add unit tests to verify that ValueError is raised when invocation_timeout_s is out of bounds (e.g., < 1 or > 3600) or when it is greater than async_timeout_s.

    def test_timeout_omitted_by_default(self):
        kwargs = self._invoke_async_kwargs(None)
        self.assertNotIn("InvocationTimeoutSeconds", kwargs)

    def test_invalid_timeout_raises_value_error(self):
        with self.assertRaises(ValueError):
            TabPFNClassifier(
                endpoint_name="ep",
                use_async=True,
                s3_bucket="bucket",
                invocation_timeout_s=0,
            )
        with self.assertRaises(ValueError):
            TabPFNClassifier(
                endpoint_name="ep",
                use_async=True,
                s3_bucket="bucket",
                invocation_timeout_s=3601,
            )

    def test_timeout_mismatch_raises_value_error(self):
        with self.assertRaises(ValueError):
            TabPFNClassifier(
                endpoint_name="ep",
                use_async=True,
                s3_bucket="bucket",
                async_timeout_s=100,
                invocation_timeout_s=200,
            )



if __name__ == "__main__":
unittest.main()
Loading