Skip to content
Open
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
24 changes: 15 additions & 9 deletions tests/tools/test_mcp_oauth_cold_load_expiry.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,10 +137,16 @@ def test_get_tokens_uses_expires_at_for_remaining_ttl(
# Should be slightly less than 3600 after the 50ms sleep.
assert 3500 < reloaded.expires_in <= 3600

def test_get_tokens_returns_zero_ttl_for_expired_token(
def test_get_tokens_returns_negative_ttl_for_expired_token(
self, tmp_path, monkeypatch
):
"""An already-expired token reloaded from disk must report expires_in=0."""
"""An expired token must remain expired after the SDK rebases its TTL.

``expires_in=0`` is not sufficient on Windows: the SDK calculates the
expiry as ``time.time() + 0`` and its inclusive validity check can see
that timestamp as still current for one coarse clock tick. A negative
TTL makes the preemptive refresh path deterministic.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from tools.mcp_oauth import HermesTokenStorage, _get_token_dir

Expand All @@ -162,9 +168,9 @@ def test_get_tokens_returns_zero_ttl_for_expired_token(
storage = HermesTokenStorage("srv")
reloaded = asyncio.run(storage.get_tokens())
assert reloaded is not None
assert reloaded.expires_in == 0, (
"Expired token must reload with expires_in=0 so the SDK's "
"is_token_valid() returns False and preemptive refresh fires."
assert reloaded.expires_in < 0, (
"Expired token must reload with a negative TTL so the SDK's "
"inclusive is_token_valid() check cannot briefly accept it."
)

def test_get_tokens_legacy_file_without_expires_at_is_loadable(
Expand All @@ -174,8 +180,8 @@ def test_get_tokens_legacy_file_without_expires_at_is_loadable(

Pre-existing token files have ``expires_in`` but no ``expires_at``.
Fix A falls back to the file's mtime as a best-effort wall-clock
proxy: a file whose (mtime + expires_in) is in the past clamps
expires_in to zero so the SDK refreshes on next request. A fresh
proxy: a file whose (mtime + expires_in) is in the past receives a
negative expires_in so the SDK refreshes on next request. A fresh
legacy-format file (mtime = now) keeps most of its TTL.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
Expand Down Expand Up @@ -204,9 +210,9 @@ def test_get_tokens_legacy_file_without_expires_at_is_loadable(
storage = HermesTokenStorage("srv")
reloaded = asyncio.run(storage.get_tokens())
assert reloaded is not None
assert reloaded.expires_in == 0, (
assert reloaded.expires_in < 0, (
"Legacy file whose mtime + expires_in is in the past must report "
"expires_in=0 so the SDK refreshes on next request."
"a negative TTL so the SDK refreshes on next request."
)


Expand Down
9 changes: 6 additions & 3 deletions tools/mcp_oauth.py
Original file line number Diff line number Diff line change
Expand Up @@ -315,14 +315,17 @@ def _rebase_expires_in(self, data: dict) -> None:
"""Rewrite ``expires_in`` to seconds remaining from the stored absolute ``expires_at`` (not an
SDK field, so stripped): a relative value reloaded after restart would make ``is_token_valid()``
True for tokens that expired while down. Legacy files without it use the file mtime, clamped
to zero (self-heals on the next ``set_tokens``)."""
to a negative TTL once expired (self-heals on the next ``set_tokens``). The negative sentinel
matters on coarse clocks: the SDK treats ``time.time() + 0`` as valid on an inclusive boundary."""
absolute_expiry = data.pop("expires_at", None)
if absolute_expiry is not None:
data["expires_in"] = int(max(absolute_expiry - time.time(), 0))
remaining = absolute_expiry - time.time()
data["expires_in"] = int(remaining) if remaining >= 1 else -1
elif data.get("expires_in") is not None:
with contextlib.suppress(OSError, TypeError, ValueError):
implied_expiry = self._tokens_path().stat().st_mtime + int(data["expires_in"])
data["expires_in"] = int(max(implied_expiry - time.time(), 0))
remaining = implied_expiry - time.time()
data["expires_in"] = int(remaining) if remaining >= 1 else -1

async def get_tokens(self) -> "OAuthToken | None":
return self._load_model(self._tokens_path(), "OAuthToken", "tokens", self._rebase_expires_in)
Expand Down