refactor: tighten runtime lifecycle and remove maintenance debt

pyproject.toml

@@ -119,6 +119,11 @@ select = [
   "I",  # isort
   "B",  # flake8-bugbear
   "C4", # flake8-comprehensions
+  "DTZ005", # datetime.now() without a timezone
+  "G004", # logging statement uses f-string
+  "RUF006", # unowned asyncio tasks
+  "RUF012", # mutable class attributes without ClassVar
+  "RUF100", # unused noqa directives
   "UP", # pyupgrade
 ]
 isort = { combine-as-imports = true, known-first-party = ["agents"] }
@@ -127,7 +132,9 @@ isort = { combine-as-imports = true, known-first-party = ["agents"] }
 convention = "google"
 
 [tool.ruff.lint.per-file-ignores]
-"examples/**/*.py" = ["E501"]
+"examples/**/*.py" = ["DTZ005", "E501", "G004", "RUF006", "RUF012", "RUF100"]
+"examples/**/*.ipynb" = ["RUF100"]
+"tests/**/*.py" = ["RUF006", "RUF012", "RUF100"]
 
 [tool.mypy]
 strict = true

src/agents/extensions/experimental/codex/codex_tool.py

@@ -532,7 +532,7 @@ def _validate_default_run_context_thread_id_suffix(value: str) -> str:
 def _parse_tool_input(parameters_model: type[BaseModel], input_json: str) -> BaseModel:
     try:
         json_data = json.loads(input_json) if input_json else {}
-    except Exception as exc:  # noqa: BLE001
+    except Exception as exc:
         if _debug.DONT_LOG_TOOL_DATA:
             logger.debug("Invalid JSON input for codex tool")
         else:
@@ -933,7 +933,7 @@ def _store_thread_id_in_run_context(
 
     try:
         setattr(context, key, thread_id)
-    except Exception as exc:  # noqa: BLE001
+    except Exception as exc:
         raise UserError(
             f'Unable to store Codex thread_id in run context field "{key}". '
             "Use a mutable dict context or set a writable attribute."
@@ -965,7 +965,7 @@ def _set_pydantic_context_value(context: BaseModel, key: str, value: str) -> boo
     if key in model_fields:
         try:
             setattr(context, key, value)
-        except Exception:  # noqa: BLE001
+        except Exception:
             return False
         return True
 
@@ -974,7 +974,7 @@ def _set_pydantic_context_value(context: BaseModel, key: str, value: str) -> boo
         return True
     except ValueError:
         pass
-    except Exception:  # noqa: BLE001
+    except Exception:
         return False
 
     state = getattr(context, "__dict__", None)

src/agents/extensions/experimental/codex/thread.py

@@ -150,7 +150,7 @@ async def _run_streamed_internal(
                         ) from exc
                     try:
                         parsed = _parse_event(item)
-                    except Exception as exc:  # noqa: BLE001
+                    except Exception as exc:
                         raise RuntimeError(f"Failed to parse event: {item}") from exc
                     if isinstance(parsed, ThreadStartedEvent):
                         # Capture the thread id so callers can resume later.

src/agents/extensions/memory/advanced_sqlite_session.py

@@ -268,7 +268,7 @@ async def store_run_usage(self, result: RunResult) -> None:
                 # Only update turn-level usage - session usage is aggregated on demand
                 await self._update_turn_usage_internal(current_turn, result.context_wrapper.usage)
         except Exception as e:
-            self._logger.error(f"Failed to store usage for session {self.session_id}: {e}")
+            self._logger.error("Failed to store usage for session %s: %s", self.session_id, e)
 
     def _get_next_turn_number(self, branch_id: str) -> int:
         """Get the next turn number for a specific branch.
@@ -491,7 +491,7 @@ def _cleanup_orphaned_messages_sync(self, conn: sqlite3.Connection) -> int:
 
             deleted_count = cursor.rowcount
             if deleted_count:
-                self._logger.info(f"Cleaned up {deleted_count} orphaned messages")
+                self._logger.info("Cleaned up %s orphaned messages", deleted_count)
             return deleted_count
 
     def _classify_message_type(self, item: TResponseInputItem) -> str:
@@ -639,7 +639,11 @@ def _validate_turn():
         self._current_branch_id = branch_name
 
         self._logger.debug(
-            f"Created branch '{branch_name}' from turn {turn_number} ('{turn_content}') in '{old_branch}'"  # noqa: E501
+            "Created branch '%s' from turn %s ('%s') in '%s'",
+            branch_name,
+            turn_number,
+            turn_content,
+            old_branch,
         )
         return branch_name
 
@@ -697,7 +701,7 @@ def _validate_branch():
 
         old_branch = self._current_branch_id
         self._current_branch_id = branch_id
-        self._logger.info(f"Switched from branch '{old_branch}' to '{branch_id}'")
+        self._logger.info("Switched from branch '%s' to '%s'", old_branch, branch_id)
 
     async def delete_branch(self, branch_id: str, force: bool = False) -> None:
         """Delete a branch and all its associated data.
@@ -778,8 +782,11 @@ def _delete_sync():
         )
 
         self._logger.info(
-            f"Deleted branch '{branch_id}': {structure_deleted} message entries, "
-            f"{usage_deleted} usage entries, {orphaned_messages_deleted} orphaned messages"
+            "Deleted branch '%s': %s message entries, %s usage entries, %s orphaned messages",
+            branch_id,
+            structure_deleted,
+            usage_deleted,
+            orphaned_messages_deleted,
         )
 
     async def list_branches(self) -> list[dict[str, Any]]:
@@ -1305,7 +1312,7 @@ def _update_sync():
                     try:
                         input_details_json = json.dumps(usage_data.input_tokens_details.__dict__)
                     except (TypeError, ValueError) as e:
-                        self._logger.warning(f"Failed to serialize input tokens details: {e}")
+                        self._logger.warning("Failed to serialize input tokens details: %s", e)
                         input_details_json = None
 
                 if (
@@ -1315,7 +1322,7 @@ def _update_sync():
                     try:
                         output_details_json = json.dumps(usage_data.output_tokens_details.__dict__)
                     except (TypeError, ValueError) as e:
-                        self._logger.warning(f"Failed to serialize output tokens details: {e}")
+                        self._logger.warning("Failed to serialize output tokens details: %s", e)
                         output_details_json = None
 
                 with closing(conn.cursor()) as cursor:

src/agents/extensions/memory/mongodb_session.py

@@ -35,7 +35,7 @@
 import threading
 import weakref
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, ClassVar
 
 from ._optional_imports import raise_optional_dependency_error
 
@@ -97,8 +97,8 @@ class MongoDBSession(SessionABC):
     #   one across loops raises RuntimeError.  create_index is idempotent, so
     #   we only need the threading lock to guard the boolean done flag — no
     #   async coordination is required.
-    _init_state: dict[int, dict[tuple[str, str, str], bool]] = {}
-    _init_guard: threading.Lock = threading.Lock()
+    _init_state: ClassVar[dict[int, dict[tuple[str, str, str], bool]]] = {}
+    _init_guard: ClassVar[threading.Lock] = threading.Lock()
 
     session_settings: SessionSettings | None = None
 

src/agents/extensions/sandbox/_rclone.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from ...sandbox.entries.mounts.patterns import RcloneMountPattern
+from ...sandbox.errors import MountConfigError
+from ...sandbox.session.base_sandbox_session import BaseSandboxSession
+
+_APT = "DEBIAN_FRONTEND=noninteractive DEBCONF_NOWARNINGS=yes apt-get -o Dpkg::Use-Pty=0"
+_RCLONE_CHECK = "command -v rclone >/dev/null 2>&1 || test -x /usr/local/bin/rclone"
+_INSTALL_RCLONE_COMMANDS = (
+    f"{_APT} update -qq",
+    f"{_APT} install -y -qq curl unzip ca-certificates",
+    "curl -fsSL https://rclone.org/install.sh | bash",
+)
+
+
+async def ensure_rclone(session: BaseSandboxSession) -> None:
+    rclone = await session.exec("sh", "-lc", _RCLONE_CHECK, shell=False)
+    if rclone.ok():
+        return
+
+    apt = await session.exec("sh", "-lc", "command -v apt-get >/dev/null 2>&1", shell=False)
+    if not apt.ok():
+        raise MountConfigError(
+            message="rclone is not installed and apt-get is unavailable; preinstall rclone",
+            context={"package": "rclone"},
+        )
+
+    for command in _INSTALL_RCLONE_COMMANDS:
+        install = await session.exec(
+            "sh",
+            "-lc",
+            command,
+            shell=False,
+            timeout=300,
+            user="root",
+        )
+        if not install.ok():
+            raise MountConfigError(
+                message="failed to install rclone",
+                context={"package": "rclone", "exit_code": install.exit_code},
+            )
+
+    rclone = await session.exec("sh", "-lc", _RCLONE_CHECK, shell=False)
+    if not rclone.ok():
+        raise MountConfigError(
+            message="rclone was installed but is still not available on PATH",
+            context={"package": "rclone"},
+        )
+
+
+async def _default_user_ids(session: BaseSandboxSession) -> tuple[str, str] | None:
+    result = await session.exec("sh", "-lc", "id -u; id -g", shell=False, timeout=30)
+    if not result.ok():
+        return None
+
+    lines = result.stdout.decode("utf-8", errors="replace").splitlines()
+    if len(lines) < 2 or not lines[0].isdigit() or not lines[1].isdigit():
+        return None
+    return lines[0], lines[1]
+
+
+def _append_option(args: list[str], option: str, *values: str) -> None:
+    if option not in args:
+        args.extend([option, *values])
+
+
+async def rclone_pattern_for_session(
+    session: BaseSandboxSession,
+    pattern: RcloneMountPattern,
+) -> RcloneMountPattern:
+    if pattern.mode != "fuse":
+        return pattern
+
+    extra_args = list(pattern.extra_args)
+    _append_option(extra_args, "--allow-other")
+    user_ids = await _default_user_ids(session)
+    if user_ids is not None:
+        uid, gid = user_ids
+        _append_option(extra_args, "--uid", uid)
+        _append_option(extra_args, "--gid", gid)
+
+    return pattern.model_copy(update={"extra_args": extra_args})

src/agents/extensions/sandbox/blaxel/sandbox.py

@@ -44,6 +44,7 @@
 from ....sandbox.session.base_sandbox_session import BaseSandboxSession
 from ....sandbox.session.dependencies import Dependencies
 from ....sandbox.session.manager import Instrumentation
+from ....sandbox.session.pty_output import collect_pty_output
 from ....sandbox.session.pty_types import (
     PTY_PROCESSES_MAX,
     PTY_PROCESSES_WARNING,
@@ -52,7 +53,6 @@
     clamp_pty_yield_time_ms,
     process_id_to_prune_from_meta,
     resolve_pty_write_yield_time_ms,
-    truncate_text_by_tokens,
 )
 from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER, RuntimeHelperScript
 from ....sandbox.session.sandbox_client import BaseSandboxClient
@@ -959,34 +959,14 @@ async def _collect_pty_output(
         yield_time_ms: int,
         max_output_tokens: int | None,
     ) -> tuple[bytes, int | None]:
-        deadline = time.monotonic() + (yield_time_ms / 1000)
-        output = bytearray()
-
-        while True:
-            async with entry.output_lock:
-                while entry.output_chunks:
-                    output.extend(entry.output_chunks.popleft())
-
-            if time.monotonic() >= deadline:
-                break
-            if entry.done:
-                async with entry.output_lock:
-                    while entry.output_chunks:
-                        output.extend(entry.output_chunks.popleft())
-                break
-
-            remaining_s = deadline - time.monotonic()
-            if remaining_s <= 0:
-                break
-            try:
-                await asyncio.wait_for(entry.output_notify.wait(), timeout=remaining_s)
-            except asyncio.TimeoutError:
-                break
-            entry.output_notify.clear()
-
-        text = output.decode("utf-8", errors="replace")
-        truncated, original_token_count = truncate_text_by_tokens(text, max_output_tokens)
-        return truncated.encode("utf-8", errors="replace"), original_token_count
+        return await collect_pty_output(
+            output_chunks=entry.output_chunks,
+            output_lock=entry.output_lock,
+            output_notify=entry.output_notify,
+            is_done=lambda: entry.done,
+            yield_time_ms=yield_time_ms,
+            max_output_tokens=max_output_tokens,
+        )
 
     async def _finalize_pty_update(
         self,