Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 38 additions & 17 deletions src/inspect_robots/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,19 @@
Entry-point groups:
``inspect_robots.tasks``, ``inspect_robots.policies``, ``inspect_robots.embodiments``,
``inspect_robots.scorers``, ``inspect_robots.sinks``.

Set ``INSPECT_ROBOTS_DISABLE_PLUGIN_AUTOLOAD`` to any non-empty value to skip
entry-point discovery: only in-tree builtins and components registered by hand
are then resolvable. This is a defense-in-depth and reproducibility switch for
locked-down eval environments, mirroring pytest's
``PYTEST_DISABLE_PLUGIN_AUTOLOAD``. It is not a security boundary: an installed
package can still run code when imported. The switch only stops this framework
from importing plugins on your behalf during discovery.
"""

from __future__ import annotations

import os
import warnings
from collections.abc import Callable
from importlib.metadata import entry_points
Expand All @@ -21,6 +30,10 @@
Kind = str # "task" | "policy" | "embodiment" | "scorer" | "sink"
KINDS: tuple[Kind, ...] = ("task", "policy", "embodiment", "scorer", "sink")

# Any non-empty value opts out of entry-point plugin autoloading (see module
# docstring). Read at discovery time, not import time, so it stays togglable.
DISABLE_AUTOLOAD_ENV = "INSPECT_ROBOTS_DISABLE_PLUGIN_AUTOLOAD"

_GROUPS: dict[Kind, str] = {
"task": "inspect_robots.tasks",
"policy": "inspect_robots.policies",
Expand Down Expand Up @@ -76,28 +89,36 @@ def sink(name: str | None = None) -> Callable[[F], F]:
return register("sink", name)


def _autoload_disabled() -> bool:
"""Whether entry-point plugin autoloading is turned off via the environment."""
return bool(os.environ.get(DISABLE_AUTOLOAD_ENV))


def _ensure_loaded() -> None:
global _loaded_builtins, _loaded_entrypoints
if not _loaded_builtins:
_loaded_builtins = True
import inspect_robots._builtins # noqa: F401 (registers builtin components)
if not _loaded_entrypoints:
_loaded_entrypoints = True
for kind, group in _GROUPS.items():
for ep in entry_points(group=group):
try:
factory = ep.load()
except Exception as exc:
# A broken plugin must not crash discovery, but it must not
# vanish silently either — that is undebuggable.
warnings.warn(
f"failed to load inspect_robots plugin {ep.name!r} from "
f"entry-point group {group!r}: {exc!r}",
RuntimeWarning,
stacklevel=2,
)
continue
_FACTORIES[kind].setdefault(ep.name, factory)
# The opt-out is deliberately not latched: it skips discovery without
# marking it done, so clearing the env var later still loads plugins.
if _loaded_entrypoints or _autoload_disabled():
return
_loaded_entrypoints = True
for kind, group in _GROUPS.items():
for ep in entry_points(group=group):
try:
factory = ep.load()
except Exception as exc:
# A broken plugin must not crash discovery, but it must not
# vanish silently either — that is undebuggable.
warnings.warn(
f"failed to load inspect_robots plugin {ep.name!r} from "
f"entry-point group {group!r}: {exc!r}",
RuntimeWarning,
stacklevel=2,
)
continue
_FACTORIES[kind].setdefault(ep.name, factory)


def registered(kind: Kind) -> dict[str, Callable[..., Any]]:
Expand Down
53 changes: 53 additions & 0 deletions tests/test_registry_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,59 @@ def fake_entry_points(*, group: str) -> list[object]:
assert "plugin_policy" in registered("policy")


def test_autoload_opt_out_skips_entrypoints_but_keeps_builtins(
monkeypatch: pytest.MonkeyPatch,
) -> None:
probe = "optout-skip-probe-policy"

class _FakeEP:
name = probe

def load(self) -> object: # pragma: no cover - must never run when opted out
raise AssertionError("entry point loaded despite the autoload opt-out")

def fake_entry_points(*, group: str) -> list[object]:
return [_FakeEP()] if group == "inspect_robots.policies" else []

monkeypatch.setattr(reg, "entry_points", fake_entry_points)
monkeypatch.setattr(reg, "_loaded_entrypoints", False)
monkeypatch.setenv(reg.DISABLE_AUTOLOAD_ENV, "1")

try:
policies = registered("policy")
finally:
reg._FACTORIES["policy"].pop(probe, None) # keep the shared registry clean
assert probe not in policies # discovery skipped, load() never called
assert "scripted" in policies # in-tree builtins still resolve


def test_autoload_opt_out_is_not_latched(monkeypatch: pytest.MonkeyPatch) -> None:
probe = "optout-latch-probe-policy"

class _FakeEP:
name = probe

def load(self) -> object:
return ScriptedPolicy

def fake_entry_points(*, group: str) -> list[object]:
return [_FakeEP()] if group == "inspect_robots.policies" else []

monkeypatch.setattr(reg, "entry_points", fake_entry_points)
monkeypatch.setattr(reg, "_loaded_entrypoints", False)

try:
monkeypatch.setenv(reg.DISABLE_AUTOLOAD_ENV, "1")
assert probe not in registered("policy")

# Clearing the opt-out re-enables discovery in the same process: the
# skip must not have marked entry points as already loaded.
monkeypatch.delenv(reg.DISABLE_AUTOLOAD_ENV)
assert probe in registered("policy")
finally:
reg._FACTORIES["policy"].pop(probe, None) # keep the shared registry clean


def test_cli_list_runs(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["list", "policies"]) == 0
out = capsys.readouterr().out
Expand Down
Loading