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

[yiboyasss]

Summary

Part of splitting #1967 into smaller, independently-reviewable pieces (secret lifecycle → identity contract → eligibility gate → frontend answer contract → runtime enable). This is the first slice: the process-local per-turn ephemeral connector secret store's TTL.

• _EPHEMERAL_RUNTIME_STORED_AT + TTL (referencing _MAX_INTERACTION_TTL_SECONDS) with an opportunistic reaper, plus a terminal-status-gated atomic pop from finish_turn / settle_task_lease_isolated / _finalize_resumed_task — a turn's secrets are freed the moment its outcome is decided as COMPLETED/FAILED, not on a later separate read.
• Expiry is now enforced on every read/pop, not only via the next unrelated store's opportunistic prune — a quiet process (no turn ever starts a new store after this one goes stale) previously kept serving an entry past its advertised TTL.
• A turn that pauses again under the same turn_id (WAITING_FOR_USER/PAUSED) renews the TTL instead of leaving the original store's timestamp in place, so a later unrelated store's prune can't reap an active turn's still-needed secrets.

No behavior change for a turn that reaches a terminal outcome normally; this only bounds/protects the deferred-to-TTL-recovery paths (lease lost, DB pool exhaustion, unhealthy heartbeat at shutdown) that can never safely pop or renew themselves.

Test plan

• New unit tests for expiry-on-read, renewal extending the TTL, renewal not resurrecting an already-expired entry, renewal no-op for an unknown turn (tests/web/services/test_connector_runtime_ephemeral.py)
• Integration tests through finish_turn and execute_resume_background/_finalize_resumed_task confirming a WAITING_FOR_USER outcome renews rather than pops
• Every new test cp-backup verified (fails without the fix)
• Full tests/web + tests/core/tools/adapters/vibe suite green; only pre-existing environment failures (missing libcairo, docker-dependent sandbox tests) and 2 order-dependent flakes untouched by this diff
• ruff / mypy / isort clean

[gemini-code-assist]

Code Review

This pull request introduces a TTL-based expiration and renewal mechanism for ephemeral per-turn connector secrets to prevent memory leaks in deferred settlement paths. It updates connector_runtime.py to track storage timestamps, enforce TTL on reads/writes, and allow TTL renewal for paused turns. Additionally, it integrates turn_id propagation across task lease settlement and finalization paths in websocket.py and task_orchestrator.py. A review comment correctly identifies a critical issue in websocket.py where an undefined task_id variable is referenced in exception handlers, which would raise a NameError and mask the original exception.

[gemini-code-assist]

The variable task_id is not defined in the scope of _finalize_resumed_task. This function receives task_lease: TaskLease as a parameter, but does not define task_id locally. Referencing task_id in the exception handlers will raise a NameError, which would mask the original exception and make debugging difficult. Please use task_lease.task_id instead.

        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_lease.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_lease.task_id,
                        turn_id,
                        exc_info=True,
                    )