File size: 5,667 Bytes
b729ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Regression tests for #83220: oversized approvals.timeout must never
overflow platform wait primitives (macOS time_t OverflowError).
The clamp lives at the single config-read site (_get_approval_timeout), so
every consumer — CLI prompt thread.join, gateway poll deadline, human-wait
ceiling, and the tool_executor authorization gate — is covered at once.
"""
from __future__ import annotations
import threading
from unittest.mock import patch
from agent.deadline import MAX_SAFE_TIMEOUT_S
def _with_configured_timeout(value):
return patch(
"tools.approval_context._get_approval_config",
return_value={"timeout": value},
)
class TestApprovalTimeoutOverflowClamp:
def test_normal_value_passes_through(self):
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout(300):
assert _get_approval_timeout() == 300
def test_oversized_value_clamped(self):
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout(10**18):
assert _get_approval_timeout() == int(MAX_SAFE_TIMEOUT_S)
def test_invalid_value_falls_back_to_default(self):
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout("soon"):
assert _get_approval_timeout() == 300
def test_oversized_float_value_clamped(self):
# YAML `1e18` arrives as a float, not an int — different int() path
# than the string/int forms; the clamp must cover it too.
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout(1e18):
assert _get_approval_timeout() == int(MAX_SAFE_TIMEOUT_S)
def test_clamp_engagement_logs_warning(self, caplog):
# Capping silently changes behavior for every consumer; operators
# must see it happen.
import tools.approval as approval_mod
from tools import approval_context
with _with_configured_timeout(10**18):
with caplog.at_level("WARNING", logger=approval_mod.__name__):
approval_context._get_approval_timeout()
assert "exceeds the platform-safe maximum" in caplog.text
def test_deadline_import_failure_fails_closed(self, monkeypatch):
# If agent.deadline ever fails to import, the clamp must fail CLOSED
# (a finite safe cap) — returning the raw value would re-open the
# exact time_t overflow this fix exists to prevent.
import builtins
from tools.approval_context import _get_approval_timeout
real_import = builtins.__import__
def _blocked(name, *args, **kwargs):
if name == "agent.deadline" or name.startswith("agent.deadline."):
raise ImportError("simulated packaging failure")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", _blocked)
with _with_configured_timeout(10**18):
value = _get_approval_timeout()
assert value == 365 * 24 * 3600
# Still platform-safe for the crashing primitive.
lock = threading.Lock()
assert lock.acquire(timeout=value)
lock.release()
def test_clamped_value_safe_for_lock_acquire(self):
# The exact primitive that crashed in #83220: Lock.acquire on macOS
# converts the relative timeout to an absolute time_t timestamp.
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout(10**18):
timeout = _get_approval_timeout()
lock = threading.Lock()
assert lock.acquire(timeout=timeout) # would raise OverflowError unclamped
lock.release()
def test_clamped_value_safe_for_thread_join(self):
# Sibling crash site: the CLI prompt fallback joins the input thread
# with the configured timeout (tools/approval.py get_input path).
from tools.approval_context import _get_approval_timeout
with _with_configured_timeout(10**18):
timeout = _get_approval_timeout()
t = threading.Thread(target=lambda: None)
t.start()
t.join(timeout=timeout) # would raise OverflowError unclamped
assert not t.is_alive()
def test_human_wait_ceiling_inherits_clamp(self):
from tools.approval_human_wait import HUMAN_WAIT_MARGIN_S, human_wait_ceiling
with _with_configured_timeout(10**18):
ceiling = human_wait_ceiling()
assert ceiling == float(int(MAX_SAFE_TIMEOUT_S)) + HUMAN_WAIT_MARGIN_S
lock = threading.Lock()
assert lock.acquire(timeout=ceiling)
lock.release()
def test_authorization_gate_timeout_safe_and_extends_with_config(self):
# The gate bound must (a) be platform-safe with an oversized config
# and (b) still EXTEND beyond the 360s fallback when approvals.timeout
# is legitimately larger — clamping it down to the fallback would
# break serialization while a real prompt is still answerable (#79719).
from agent.tool_executor import (
_AUTHORIZATION_GATE_LOCK_TIMEOUT_S,
_authorization_gate_lock_timeout,
)
with _with_configured_timeout(10**18):
bound = _authorization_gate_lock_timeout()
lock = threading.Lock()
assert lock.acquire(timeout=bound)
lock.release()
with _with_configured_timeout(3600):
bound = _authorization_gate_lock_timeout()
assert bound > _AUTHORIZATION_GATE_LOCK_TIMEOUT_S
assert bound == 3600 + 60.0 # approvals.timeout + HUMAN_WAIT_MARGIN_S
|