fix(nodes): per-thread Google transport and GET-only retries for transport faults

nodes/src/nodes/tool_google_workspace/google_client.py

@@ -39,8 +39,10 @@
 - scope diagnostics (:func:`token_scope_report`) shared by ``validateConfig``,
   ``check_connection``, and ``build_service`` so the three checks cannot drift;
 - request execution with exponential backoff on 429/5xx and on rate-limit
-  403s (parsed from the structured error body), while permission 403s and
-  other errors fail fast with an agent-readable message.
+  403s (parsed from the structured error body), plus the same backoff for
+  status-less transport faults on GET requests only, all on a per-thread
+  transport (httplib2 is not thread safe under parallel agent tool calls);
+  permission 403s and other errors fail fast with an agent-readable message.
 
 Per-service subpackages keep only their tool functions and response cleaners,
 and bind these functions via ``functools.partial`` (see e.g. ``sheets/client.py``).
@@ -52,6 +54,7 @@
 import binascii
 import json
 import os
+import threading
 import time as _time
 from dataclasses import dataclass, field
 from typing import Any
@@ -399,23 +402,68 @@ def build_service(svc: GoogleService, auth_type: str, cfg: dict, scopes: list[st
     return build(svc.api, svc.version, credentials=creds, cache_discovery=False)
 
 
+_thread_transport = threading.local()
+
+
+def _request_http(request: Any):
+    """Per-thread transport for a request built on a shared service handle.
+
+    googleapiclient services carry one httplib2 transport and every request
+    built from a service inherits it. httplib2 is not thread safe: concurrent
+    ``execute()`` calls on the same handle can interleave on one TLS
+    connection, which surfaces as ``[SSL] record layer failure`` when an agent
+    executor runs a wave of tool calls in parallel threads. Rebuild the
+    authorized transport once per thread, keyed by the credential object, so
+    parallel calls never share a connection. Returns None (keep the request's
+    own transport) when there is nothing to rebuild from.
+    """
+    shared = getattr(request, 'http', None)
+    creds = getattr(shared, 'credentials', None)
+    if creds is None:
+        return None
+    try:
+        import google_auth_httplib2
+        import httplib2
+    except ImportError:
+        return None
+    cache = getattr(_thread_transport, 'by_creds', None)
+    if cache is None:
+        cache = {}
+        _thread_transport.by_creds = cache
+    http = cache.get(id(creds))
+    if http is None:
+        http = google_auth_httplib2.AuthorizedHttp(creds, http=httplib2.Http())
+        cache[id(creds)] = http
+    return http
+
+
 def execute(svc: GoogleService, request: Any, *, binary: bool = False) -> Any:
     """Run an API request with exponential backoff on 429/5xx and rate-limit 403s.
 
+    Status-less failures (connection reset, TLS fault, timeout) are retried
+    with the same backoff for GET requests only. A mutation whose response was
+    lost may still have landed on Google's side, so it fails fast and leaves
+    the re-check to the caller rather than risking a duplicated side effect.
+    Requests run on a per-thread transport (see ``_request_http``).
+
     ``binary=True`` returns the raw response (bytes from get_media/export_media);
     otherwise the JSON dict (or {} when the API returns no body). With 4 attempts
     the sleeps are 1s, 2s, 4s — worst case ~7s before the final raise.
     """
     base_delay = 1.0
+    http = _request_http(request)
     for attempt in range(4):
         try:
-            result = request.execute()
+            result = request.execute(http=http) if http is not None else request.execute()
             return result if binary else (result or {})
         except Exception as exc:  # googleapiclient.errors.HttpError and transport errors
             status = getattr(getattr(exc, 'resp', None), 'status', None)
             if status and (int(status) in _RETRY_STATUSES or _is_rate_limit_403(exc)) and attempt < 3:
                 _time.sleep(base_delay * (2**attempt))
                 continue
+            if status is None and attempt < 3 and getattr(request, 'method', None) == 'GET':
+                _time.sleep(base_delay * (2**attempt))
+                continue
             detail = getattr(exc, 'reason', None) or str(exc)
             if status and int(status) == 403:
                 raise ValueError(

nodes/test/tool_google_workspace/test_google_client.py

@@ -294,3 +294,89 @@ def read(self):
         creds.refresh(None)
     # the half-update is the bug: neither token nor expiry may have changed
     assert creds.token == 'stale'
+
+
+def _scripted_request(method, outcomes):
+    """Request double whose execute() pops scripted outcomes in order."""
+    calls = {'count': 0, 'http_seen': []}
+
+    def execute(http=None):
+        calls['http_seen'].append(http)
+        index = min(calls['count'], len(outcomes) - 1)
+        calls['count'] += 1
+        outcome = outcomes[index]
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    request = types.SimpleNamespace(method=method, http=None, execute=execute)
+    return request, calls
+
+
+def test_execute_retries_transport_faults_for_get(monkeypatch, service):
+    monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None)
+    fault = OSError('[SSL] record layer failure (_ssl.c:2580)')
+    request, calls = _scripted_request('GET', [fault, fault, {'ok': True}])
+
+    assert google_client.execute(service, request) == {'ok': True}
+    assert calls['count'] == 3
+
+
+def test_execute_gives_up_on_get_after_four_transport_attempts(monkeypatch, service):
+    monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None)
+    fault = OSError('connection reset by peer')
+    request, calls = _scripted_request('GET', [fault])
+
+    with pytest.raises(ValueError, match='request failed'):
+        google_client.execute(service, request)
+    assert calls['count'] == 4
+
+
+def test_execute_does_not_retry_transport_faults_for_mutations(monkeypatch, service):
+    monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None)
+    request, calls = _scripted_request('POST', [OSError('connection reset by peer')])
+
+    with pytest.raises(ValueError, match='request failed'):
+        google_client.execute(service, request)
+    assert calls['count'] == 1
+
+
+def test_execute_uses_a_distinct_transport_per_thread(monkeypatch, service):
+    import threading
+
+    class FakeAuthorizedHttp:
+        def __init__(self, credentials, http=None):
+            self.credentials = credentials
+            self.http = http
+
+    fake_google_auth = types.SimpleNamespace(AuthorizedHttp=FakeAuthorizedHttp)
+    fake_httplib2 = types.SimpleNamespace(Http=lambda: object())
+    monkeypatch.setitem(sys.modules, 'google_auth_httplib2', fake_google_auth)
+    monkeypatch.setitem(sys.modules, 'httplib2', fake_httplib2)
+    monkeypatch.setattr(google_client, '_thread_transport', threading.local())
+
+    credentials = object()
+    per_thread = {}
+
+    def run(name):
+        seen = []
+        for _ in range(2):
+            request = types.SimpleNamespace(
+                method='GET',
+                http=types.SimpleNamespace(credentials=credentials),
+                execute=lambda http=None: seen.append(http) or {},
+            )
+            google_client.execute(service, request)
+        per_thread[name] = seen
+
+    threads = [threading.Thread(target=run, args=(i,)) for i in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    first, second = per_thread[0], per_thread[1]
+    assert first[0] is first[1], 'a thread must reuse its own transport'
+    assert second[0] is second[1], 'a thread must reuse its own transport'
+    assert first[0] is not second[0], 'threads must not share a transport'
+    assert isinstance(first[0], FakeAuthorizedHttp)