fix(connectors): bound and renew ephemeral connector secrets by their TTL

src/xagent/web/api/websocket.py

@@ -3166,10 +3166,22 @@ def _finalize_resumed_task(
     result: Dict[str, Any],
     task_lease: TaskLease,
     prepared_outputs: _PreparedTaskFileOutputs,
+    turn_id: str | None = None,
 ) -> dict[str, Any]:
-    """Persist one fenced resumed result in a single worker transaction."""
+    """Persist one fenced resumed result in a single worker transaction.
+
+    ``turn_id``, when given, pops that turn's ephemeral connector secrets
+    (see ``task_orchestrator.finish_turn``'s matching handling) once this
+    exact call has committed a genuine COMPLETED/FAILED transition - not
+    when the lease release below is fenced out and the whole transaction
+    rolls back. A committed WAITING_FOR_USER/PAUSED transition instead
+    renews those secrets' TTL: it is the same turn resuming again later
+    under this same turn_id, now carrying a fresh interaction lifetime of
+    its own.
+    """
     from ..models.agent import Agent
     from ..services.chat_history_service import persist_assistant_message_no_commit
+    from ..services.task_orchestrator import TERMINAL_TASK_STATUSES
 
     finalized: dict[str, Any] = {
         "task_title": None,
@@ -3306,6 +3318,40 @@ def _finalize_resumed_task(
         finalized["lease_released"] = True
         finalized["final_status"] = final_task_status.value
         finalized["control_event_state"] = control_snapshot.as_dict()
+        if turn_id is not None:
+            if final_task_status in TERMINAL_TASK_STATUSES:
+                try:
+                    from ..services.connector_runtime import (
+                        pop_ephemeral_runtime_values,
+                    )
+
+                    pop_ephemeral_runtime_values(turn_id)
+                except Exception:
+                    logger.warning(
+                        "connector runtime cleanup failed for task %s turn %s",
+                        task_id,
+                        turn_id,
+                        exc_info=True,
+                    )
+            else:
+                # WAITING_FOR_USER/PAUSED: the same turn resuming again later
+                # under this same turn_id, with a fresh interaction lifetime -
+                # keep whatever ephemeral secrets it may still need from
+                # expiring on the original pause's clock (see
+                # connector_runtime.renew_ephemeral_runtime_values).
+                try:
+                    from ..services.connector_runtime import (
+                        renew_ephemeral_runtime_values,
+                    )
+
+                    renew_ephemeral_runtime_values(turn_id)
+                except Exception:
+                    logger.warning(
+                        "connector runtime secret renewal failed for task %s turn %s",
+                        task_id,
+                        turn_id,
+                        exc_info=True,
+                    )
         return finalized
     finally:
         try:
@@ -3322,11 +3368,14 @@ def _settle_resumed_task_lease(
     lease: TaskLease,
     *,
     error_message: str | None,
+    turn_id: str | None = None,
 ) -> bool:
     """Delegate resume cleanup to the shared run/runner-fenced lifecycle."""
     from ..services.task_orchestrator import settle_task_lease_isolated
 
-    return settle_task_lease_isolated(lease, error_message=error_message)
+    return settle_task_lease_isolated(
+        lease, error_message=error_message, turn_id=turn_id
+    )
 
 
 async def execute_resume_background(
@@ -3362,6 +3411,22 @@ async def execute_resume_background(
     if resume_owner_task is None:
         raise RuntimeError(f"Task {task_id} resume has no asyncio task")
 
+    # The turn id this task's ephemeral connector secrets (if any were ever
+    # stored) live under. A resume deliberately never rebinds this on the
+    # cached agent's tool_config (see WebToolConfig.invalidate_connector_
+    # runtime_cache's docstring), so it is still the original pausing
+    # turn's id here - exactly what a terminal settlement below must pop
+    # under to actually free those secrets.
+    connector_runtime_turn_id: str | None = None
+    _resume_tool_config = getattr(agent_service, "tool_config", None)
+    _turn_id_getter = getattr(
+        _resume_tool_config, "get_connector_runtime_turn_id", None
+    )
+    if callable(_turn_id_getter):
+        _candidate_turn_id = _turn_id_getter()
+        if isinstance(_candidate_turn_id, str):
+            connector_runtime_turn_id = _candidate_turn_id
+
     lease_stop_event = preacquired_heartbeat_stop
     lease_heartbeat_task = preacquired_heartbeat_task
     lease: TaskLease | None = preacquired_lease
@@ -3515,6 +3580,7 @@ async def mark_deferred_delivery_failed() -> bool:
                 lambda acquired: _settle_resumed_task_lease(
                     acquired,
                     error_message="resume cancelled during lease acquisition",
+                    turn_id=connector_runtime_turn_id,
                 ),
             )
             if prior_status_box:
@@ -3815,6 +3881,7 @@ async def mark_deferred_delivery_failed() -> bool:
                     result=result,
                     task_lease=lease,
                     prepared_outputs=outputs_for_finalizer,
+                    turn_id=connector_runtime_turn_id,
                 )
             )
         finally:
@@ -4188,11 +4255,20 @@ async def finalize_resume_resources() -> None:
                     and not lease_released
                     and not defer_db_cleanup_to_ttl_recovery
                 ):
+                    # When this IS deferred, _finalize_resumed_task's
+                    # turn_id-scoped pop never runs for this turn - and
+                    # cannot safely run here either, for the same reason as
+                    # task_orchestrator.py's matching branch: this coroutine
+                    # does not know whether the task will land on a terminal
+                    # status or resume again under this same turn_id.
+                    # connector_runtime.py's _EPHEMERAL_RUNTIME_TTL_SECONDS
+                    # bounds that leak instead.
                     try:
                         settled = await run_db_io_cancellation_safe(
                             lambda: _settle_resumed_task_lease(
                                 lease,
                                 error_message=settlement_error,
+                                turn_id=connector_runtime_turn_id,
                             )
                         )
                         if settled:

src/xagent/web/services/connector_runtime.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+import time
 from collections.abc import Callable, Collection
 from dataclasses import dataclass
 from threading import RLock
@@ -45,6 +46,7 @@
 from ..models.custom_api import CustomApi, UserCustomApi
 from ..models.mcp import MCPServer, UserMCPServer
 from ..models.task import Task, TaskConnectorRuntimeContext
+from .task_interaction_service import _MAX_INTERACTION_TTL_SECONDS
 
 logger = logging.getLogger(__name__)
 
@@ -112,9 +114,57 @@ class _ConnectorRuntimeResolverRegistration:
 # through the resolver hook or a deployment-owned distributed secret store.
 _EPHEMERAL_RUNTIME_VALUES: dict[str, dict[str, Any]] = {}
 _EPHEMERAL_RUNTIME_MANIFESTS: dict[str, dict[str, dict[str, set[str]]]] = {}
+_EPHEMERAL_RUNTIME_STORED_AT: dict[str, float] = {}
 _EPHEMERAL_RUNTIME_VALUES_LOCK = RLock()
 _RUNTIME_RESOLVER_REGISTRATION: _ConnectorRuntimeResolverRegistration | None = None
 
+# Not a real interaction deadline - matches task_interaction_service.py's
+# _MAX_INTERACTION_TTL_SECONDS by reference (not by copying the value) so
+# this can never silently drift shorter than the longest pause production
+# already allows a native waiting-for-user interaction to stay open for.
+# Its only job is to turn what would otherwise be a permanent leak into a
+# bounded one: task_orchestrator.py's and websocket.py's deferred-to-TTL-
+# recovery branches (lease lost, DB pool exhaustion, unhealthy heartbeat at
+# shutdown) cannot safely pop a turn's secrets themselves, because at the
+# point they bail out they genuinely do not know whether the task will land
+# on a terminal status or resume again under the same turn_id - and
+# task_lease_recovery.py's batch sweep, which does later decide that,
+# has no way to look up which turn_id belongs to which recovered task_id.
+_EPHEMERAL_RUNTIME_TTL_SECONDS = _MAX_INTERACTION_TTL_SECONDS
+
+
+def _prune_expired_ephemeral_runtime_values_locked() -> None:
+    """Evict every stale entry. Caller must hold _EPHEMERAL_RUNTIME_VALUES_LOCK."""
+
+    now = time.monotonic()
+    stale_turn_ids = [
+        turn_id
+        for turn_id, stored_at in _EPHEMERAL_RUNTIME_STORED_AT.items()
+        if now - stored_at > _EPHEMERAL_RUNTIME_TTL_SECONDS
+    ]
+    for turn_id in stale_turn_ids:
+        _evict_if_expired_locked(turn_id)
+
+
+def _evict_if_expired_locked(turn_id: str) -> None:
+    """Evict one entry if it has aged past the TTL. Caller must hold the lock.
+
+    The opportunistic reaper in ``store_ephemeral_runtime_values`` only runs
+    when *some* turn writes - a quiet process (no new turns starting) never
+    calls it, so an entry whose owning turn stalled out would otherwise stay
+    readable forever past its advertised TTL. Every read/pop path below
+    checks age itself first, so "expired" is enforced as an observable
+    property of this store, not merely a future cleanup side effect.
+    """
+
+    stored_at = _EPHEMERAL_RUNTIME_STORED_AT.get(turn_id)
+    if stored_at is not None and time.monotonic() - stored_at > (
+        _EPHEMERAL_RUNTIME_TTL_SECONDS
+    ):
+        _EPHEMERAL_RUNTIME_VALUES.pop(turn_id, None)
+        _EPHEMERAL_RUNTIME_MANIFESTS.pop(turn_id, None)
+        _EPHEMERAL_RUNTIME_STORED_AT.pop(turn_id, None)
+
 
 def set_connector_runtime_resolver(
     resolver: ConnectorRuntimeResolver | None,
@@ -200,16 +250,42 @@ def store_ephemeral_runtime_values(
         for ref, sections in values_by_ref.items()
     }
     with _EPHEMERAL_RUNTIME_VALUES_LOCK:
+        # Opportunistic reaper: there is no dedicated background sweep for
+        # this process-local store, so each new store is what eventually
+        # reclaims an entry a settlement path could never safely pop (see
+        # _EPHEMERAL_RUNTIME_TTL_SECONDS above).
+        _prune_expired_ephemeral_runtime_values_locked()
         _EPHEMERAL_RUNTIME_VALUES[turn_id] = encoded
         _EPHEMERAL_RUNTIME_MANIFESTS[turn_id] = manifest
+        _EPHEMERAL_RUNTIME_STORED_AT[turn_id] = time.monotonic()
 
 
 def pop_ephemeral_runtime_values(turn_id: str) -> dict[str, Any] | None:
     with _EPHEMERAL_RUNTIME_VALUES_LOCK:
+        _evict_if_expired_locked(turn_id)
         _EPHEMERAL_RUNTIME_MANIFESTS.pop(turn_id, None)
+        _EPHEMERAL_RUNTIME_STORED_AT.pop(turn_id, None)
         return _EPHEMERAL_RUNTIME_VALUES.pop(turn_id, None)
 
 
+def renew_ephemeral_runtime_values(turn_id: str) -> None:
+    """Refresh a still-live turn's stored-at timestamp, extending its TTL.
+
+    Called when a turn is confirmed to be re-pausing under the same
+    ``turn_id`` (see ``task_orchestrator.finish_turn`` and
+    ``websocket._finalize_resumed_task``'s matching non-terminal branches):
+    a fresh pause carries its own new interaction lifetime, so the secrets
+    it may still need must not expire on the *original* pause's clock. A
+    no-op when the turn has nothing stored (never used ephemeral secrets, or
+    already reaped) - there is nothing to keep alive either way.
+    """
+
+    with _EPHEMERAL_RUNTIME_VALUES_LOCK:
+        _evict_if_expired_locked(turn_id)
+        if turn_id in _EPHEMERAL_RUNTIME_VALUES:
+            _EPHEMERAL_RUNTIME_STORED_AT[turn_id] = time.monotonic()
+
+
 def drop_ephemeral_runtime_values_for_testing(turn_id: str) -> None:
     """Simulate losing the secret values while keeping safe provenance."""
 
@@ -219,6 +295,7 @@ def drop_ephemeral_runtime_values_for_testing(turn_id: str) -> None:
 
 def get_ephemeral_runtime_values(turn_id: str) -> dict[str, Any] | None:
     with _EPHEMERAL_RUNTIME_VALUES_LOCK:
+        _evict_if_expired_locked(turn_id)
         values = _EPHEMERAL_RUNTIME_VALUES.get(turn_id)
         return dict(values) if isinstance(values, dict) else None
 
@@ -227,6 +304,7 @@ def get_ephemeral_runtime_manifest(
     turn_id: str,
 ) -> dict[str, dict[str, set[str]]] | None:
     with _EPHEMERAL_RUNTIME_VALUES_LOCK:
+        _evict_if_expired_locked(turn_id)
         manifest = _EPHEMERAL_RUNTIME_MANIFESTS.get(turn_id)
         if not isinstance(manifest, dict):
             return None