diff --git a/effectful/handlers/llm/governance.py b/effectful/handlers/llm/governance.py new file mode 100644 index 000000000..0a64e0ac6 --- /dev/null +++ b/effectful/handlers/llm/governance.py @@ -0,0 +1,129 @@ +"""Static tool-graph governance for LLM :class:`~effectful.handlers.llm.template.Template` s. + +A :class:`~effectful.handlers.llm.template.Tool` **is** an +:class:`~effectful.ops.types.Operation` (``class Tool(Operation)``, +``class Template(Tool)``), so a template's tools *are* part of its effect row. These +compute the tool graph without ever calling the LLM: + +* :func:`toolsof` — the transitive tool graph reachable from a tool/template via ``.tools``. + Fully static: it reads lexically-captured ``.tools`` mappings, nothing is executed. +* :func:`reachable_tools` — the tools a zero-arg function can reach, *through* templates, + by reifying it to a :class:`~effectful.ops.types.Term` (never running the LLM or any + tool body) and walking it for the tools it mentions, then expanding via :func:`toolsof`. +* :func:`check_tools` — the leak check ``reachable_tools(fn) - allowed``. + +**Soundness precondition (important).** :func:`reachable_tools` obtains the term by +*running ``fn``'s own Python body* under a reifying interpretation. Operation calls become +term nodes, but **native Python control flow in ``fn`` is resolved at analysis time** — an +untaken ``if``/``for``/``try`` branch contributes no tools. So ``check_tools(fn) == set()`` +is a tool-safety guarantee **only for a straight-line ``fn``** (or one whose branching is +expressed with reifying conditional *operations*, which fold both arms). It is not a proof +over arbitrary Python control flow; for a branchy ``fn`` it is the set reached *on this +reification*, which may under-approximate. Keep governed entry points straight-line. +""" + +import collections.abc +from collections.abc import Callable +from typing import Any + +from effectful.handlers.llm.completions import _LexicalVariableTool +from effectful.handlers.llm.template import Tool +from effectful.internals.runtime import interpreter +from effectful.ops.semantics import apply +from effectful.ops.syntax import defdata +from effectful.ops.types import Term + +__all__ = ["toolsof", "reachable_tools", "check_tools"] + + +def _is_governed(tool: Any) -> bool: + """A ``Tool`` the governance graph should count. Excludes synthetic + :class:`~effectful.handlers.llm.completions._LexicalVariableTool` readers — they are + prompt-variable plumbing auto-wrapped from lexical values by ``LexicalReaders``, not + tools an agent "reaches", so treating them as reachable would flag plumbing as a leak. + """ + return isinstance(tool, Tool) and not isinstance(tool, _LexicalVariableTool) + + +def _tools_in(term: Any) -> frozenset[Tool]: + """Every governed :class:`Tool` appearing *anywhere* in a reified ``term`` — as an + operation or as an argument. ``Tool`` / ``Template`` subclass + :class:`~effectful.ops.types.Operation` and define their own ``__apply__``, so a called + tool sits in the ``args`` of an ``apply`` node rather than being the node's ``op``; a + structural walk catches both. Synthetic lexical readers are excluded (:func:`_is_governed`). + """ + found: set[Tool] = set() + + def walk(x: Any) -> None: + if _is_governed(x): + found.add(x) + if isinstance(x, Term): + walk(x.op) + for a in x.args: + walk(a) + for v in x.kwargs.values(): + walk(v) + elif isinstance(x, collections.abc.Mapping): + for k, v in x.items(): + walk(k) + walk(v) + elif isinstance(x, (list, tuple, set, frozenset)): + for e in x: + walk(e) + + walk(term) + return frozenset(found) + + +def toolsof(tool: Tool) -> frozenset[Tool]: + """The tools transitively reachable from ``tool`` through its ``.tools`` graph + (a template's tools are themselves tools, so this closes over sub-agents too). + + Fully static — it reads the lexically-captured ``.tools`` mapping, never calls the + LLM. ``tool`` itself is *not* included (it is the root, not something it reaches), and + synthetic lexical readers are excluded (:func:`_is_governed`). + """ + seen: set[Tool] = set() + stack: list[Tool] = [tool] + while stack: + cur = stack.pop() + for sub in getattr(cur, "tools", {}).values(): + if _is_governed(sub) and sub not in seen: + seen.add(sub) + stack.append(sub) + return frozenset(seen) + + +def reachable_tools(fn: Callable[[], Any]) -> frozenset[Tool]: + """Every tool a zero-arg ``fn`` can reach, *including through templates it calls*, + without ever running a tool body or the LLM. + + ``fn`` is reified to a :class:`~effectful.ops.types.Term` under ``defdata`` — so even + tools with real implementations become term nodes rather than executing — and the term + is walked for the :class:`Tool` s it mentions (:func:`_tools_in`). Those are the + *directly* reached tools; :func:`toolsof` expands each to the tools it in turn reaches. + A template's body is ``raise NotHandled`` so it performs no tools directly, but the + tools it captured lexically are recovered statically through :func:`toolsof`. + + The reifier uses ``interpreter`` (*replace*), never ``handler`` (*merge*): merging + would let an ambient ``Tool.__apply__`` handler win dispatch, so the tool would run + concretely instead of reifying and the walk would silently miss it — the static + guarantee must hold regardless of what is installed at the call site. + """ + with interpreter({apply: defdata}): + term = fn() # reify — no tool body runs, no LLM call, ambient handlers ignored + direct = _tools_in(term) + return direct.union(*(toolsof(t) for t in direct)) + + +def check_tools(fn: Callable[[], Any], *allowed: Tool) -> frozenset[Tool]: + """Tools ``fn`` can reach that are not in the ``allowed`` set — the static tool-safety + leak check, computed with **no LLM call**. Empty == ``fn`` reaches no tool outside + ``allowed`` *on this reification* (including through nested templates); this is a + guarantee only for a straight-line ``fn`` — see :func:`reachable_tools` for the + precondition (a branchy ``fn`` may under-approximate). + + The tool-graph analogue of :func:`~effectful.ops.effects.check_uses`: + ``reachable_tools(fn) - allowed``. + """ + return reachable_tools(fn) - frozenset(allowed) diff --git a/effectful/ops/effects.py b/effectful/ops/effects.py new file mode 100644 index 000000000..5f9921a37 --- /dev/null +++ b/effectful/ops/effects.py @@ -0,0 +1,352 @@ +"""Effect-row inference — the ``ε`` engine. + +Sibling of :func:`effectful.ops.semantics.typeof` / :func:`fvsof`: a fold over the +universal ``apply`` operation that computes which operations a term performs (its +**effect row**), plus the ``Uses`` / ``Computation`` / ``Requires`` annotations that +operations declare and the fold reads. + +The per-op rule and its annotation live on the core types, mirroring the ``τ`` / ``fvs`` +machinery exactly: + +====================== ===================================================== +symbol home +====================== ===================================================== +``Operation.__uses_rule__`` ``ops/types.py``, a ``@final`` method next to ``__type_rule__`` / ``__fvs_rule__`` +``Uses`` ``ops/syntax.py``, next to ``Scoped`` (read by ``__uses_rule__``) +``usesof`` / ``effectsof`` this module (fold, next to ``typeof`` / ``fvsof`` in spirit) +``Computation`` / ``Requires`` this module; argument annotations read by the fold +====================== ===================================================== + +Like ``Uses``, ``Computation`` and ``Requires`` are plain *read*-metadata, not +:class:`~effectful.ops.types.Annotation` signature-transforms: enforcement is at +``usesof``-time (``_fold_computation_args`` fails loudly on an unclassified callable), so +there is no build-time ``infer_annotations`` gate to wire. + +The static LLM tool-governance layer (``toolsof`` / ``reachable_tools`` / ``check_tools`` +— transitive tool-graph reachability with no LLM call) is built on top of this in +``handlers/llm/governance.py``. + +**Not yet implemented:** ``usagesof`` (usage multiset) and handler discharge; polymorphic +``Operation[[A], B]`` ``Uses`` members; and the *runtime* LLM tool-governance layer — tool +**restriction** as an off-by-default handler filtering the offered tool set (which must +leave synthetic ``LexicalReaders`` tools untouched — they are prompt-variable plumbing, +not effectful tools), a decode-time ``ε`` validator, and ``tool_choice`` forcing. This +module is the ``ε`` core (fold + argument annotations). +""" + +import typing +from dataclasses import dataclass +from typing import Annotated, Any + +from effectful.internals.runtime import interpreter +from effectful.ops.semantics import apply, evaluate, typeof +from effectful.ops.syntax import Uses +from effectful.ops.types import Expr, Operation + +__all__ = [ + "Computation", + "Requires", + "UndeclaredCallable", + "UnsoundCallbackFold", + "usesof", + "effectsof", + "effect_type", + "check_uses", + "requires_rule", + "check_requires", +] + + +# --------------------------------------------------------------------------- +# Annotations an op declares (read off ``Annotated[T, ...]``). +# ``Uses`` itself lives in ``ops/syntax.py`` (next to ``Scoped``) and is read by +# ``Operation.__uses_rule__``; ``Computation`` / ``Requires`` are argument annotations +# read by the fold below. +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class _Computation: + """``Annotated[Callable[[A], B], Computation]`` on an *argument*: a suspended + computation whose effect row joins the op's row when the op runs it. Higher-order + combinators (``map``/``filter``/…) mark their callback arg with this instead of the + checker hard-coding which ops are higher-order. + + Plain read-metadata (like :class:`~effectful.ops.syntax.Uses`), not an + :class:`~effectful.ops.types.Annotation` — it is *read* by the fold, not a signature + transform. An unclassified callable argument is caught loudly at fold time by + :func:`_fold_computation_args`, so no build-time gate is needed.""" + + +#: Singleton marker (data-less, like ``IsRecursive``) — use in ``Annotated[C, Computation]``. +Computation = _Computation() + + +@dataclass(frozen=True, init=False) +class Requires: + """``Annotated[T, Requires(op, ...)]`` on an *argument*: the value's provenance must + cover these ops — ``{op,...} ⊆ usesof(arg)``. The precondition dual of ``Uses`` + (#664). Plain read-metadata (like :class:`~effectful.ops.syntax.Uses`), read by + :func:`requires_rule`.""" + + ops: frozenset[Operation] + + def __init__(self, *ops: Operation) -> None: + object.__setattr__(self, "ops", frozenset(ops)) + + def missing(self, arg: Any) -> frozenset[Operation]: + """Required ops absent from the argument's provenance — the whole check is one + :func:`usesof`.""" + return self.ops - usesof(arg) + + +class UndeclaredCallable(Exception): + """Raised by the fold on a callable argument that is neither ``Computation`` nor + ``Uses[()]`` — the checker refuses to guess rather than silently under-approximate.""" + + +class UnsoundCallbackFold(Exception): + """Raised when folding a ``Computation`` callback that *inspects* its argument through + the operator/dunder protocol (branches on ``==``/``<``, does arithmetic, calls it, + accesses a missing attribute, iterates/indexes it). The fold runs the callback on an + opaque placeholder (:class:`_Opaque`) to collect its effects; such inspection would + path/structure-under-approximate, so the fold refuses loudly. This is a best-effort + tripwire — identity (``is``), ``type()``/``isinstance`` and existing-attribute access + bypass it (see :class:`_Opaque`); the sound contract is that callbacks stay + straight-line in their argument. A symbolic-execution provider would remove the + restriction entirely.""" + + +# --------------------------------------------------------------------------- +# The fold (upstream: ops/semantics.py, next to typeof/fvsof) +# --------------------------------------------------------------------------- +def usesof[S](term: Expr[S]) -> frozenset[Operation]: + """Return the effect row of a term: the set of operations it performs. The + effect-typing sibling of :func:`typeof` / :func:`fvsof`. + + Each applied op contributes :meth:`Operation.__uses_rule__` (default ``{self}``, + ``Uses[()]`` = pure). ``Computation``-marked callback args are *entered* so their + effects fold too; an undeclared callable arg raises :class:`UndeclaredCallable` + (never silent). Entering a callback runs it on an opaque placeholder, so it is sound + only for callbacks that stay straight-line in their argument (pass it to ops, don't + inspect it) — most inspection is caught loudly, with the caveats in :class:`_Opaque`. + """ + used: set[Operation] = set() + + def _update(op: Operation, *args: Any, **kwargs: Any) -> Any: + used.update(op.__uses_rule__()) + _fold_computation_args(op, args, kwargs) # enters callbacks; loud on undeclared + + with interpreter({apply: _update}): + evaluate(term) + return frozenset(used) + + +#: Reads better at effect-typing call sites; same function. +effectsof = usesof + + +def effect_type[S](term: Expr[S]) -> tuple[type[S], frozenset[Operation]]: + """The effect type ``(τ, ε)`` of a term: its result type and its effect row — + ``τ`` from :func:`typeof`, ``ε`` from :func:`usesof`. The two folds compose over the + same ``apply`` op.""" + return typeof(term), usesof(term) + + +def check_uses(op: Operation, body: Expr[Any]) -> frozenset[Operation]: + """Effects ``body`` performs that ``op``'s declared ``Uses[...]`` does not cover — + empty == the declaration is sound (and transitively closed, since ``usesof`` unions + the whole DAG). This is the checker for a composite op: ``usesof(body) ⊆ declared``. + An op with no ``Uses`` annotation declares nothing, so every effect is reported.""" + declared = Uses.declared(op.__signature__) + return usesof(body) - (declared if declared is not None else frozenset()) + + +# --------------------------------------------------------------------------- +# Requires verification (upstream: with usesof, in semantics.py) +# --------------------------------------------------------------------------- +def requires_rule( + op: Operation, *args: Any, **kwargs: Any +) -> dict[str, frozenset[Operation]]: + """Per-argument unmet provenance for ``op``: ``{arg_name: missing_ops}``. Empty == + every ``Requires`` on ``op`` is satisfied by the given args.""" + bound = op.__signature__.bind(*args, **kwargs) + bound.apply_defaults() + unmet: dict[str, frozenset[Operation]] = {} + for name, p in op.__signature__.parameters.items(): + for anno in _annotations(p.annotation): + if isinstance(anno, Requires) and ( + m := anno.missing(bound.arguments[name]) + ): + unmet[name] = m + return unmet + + +def check_requires(term: Expr[Any]) -> dict[Operation, dict[str, frozenset[Operation]]]: + """Provenance violations in ``term``: ``{op: {arg: missing_ops}}``. Empty == OK. + This is #664's "public hook to read a Term's effective row" — one fold over ``apply``.""" + violations: dict[Operation, dict[str, frozenset[Operation]]] = {} + + def _update(op: Operation, *args: Any, **kwargs: Any) -> Any: + if unmet := requires_rule(op, *args, **kwargs): + violations[op] = unmet + return op.__default_rule__(*args, **kwargs) + + with interpreter({apply: _update}): + evaluate(term) + return violations + + +# --------------------------------------------------------------------------- +# helpers (annotation reading) +# --------------------------------------------------------------------------- +def _annotations(annotation: Any) -> tuple[Any, ...]: + """All metadata of a (possibly *nested*) ``Annotated`` — ``defop`` wraps params in + ``Annotated[..., Scoped]`` so a manual annotation can end up one layer deep.""" + out: list[Any] = [] + while typing.get_origin(annotation) is Annotated: + args = typing.get_args(annotation) + annotation, meta = args[0], args[1:] + out.extend(meta) + return tuple(out) + + +def _has(annotation: Any, kinds: tuple[type, ...]) -> bool: + return any(isinstance(a, kinds) for a in _annotations(annotation)) + + +def _fold_computation_args(op: Operation, args: Any, kwargs: Any) -> None: + """Enter each ``Computation`` callback arg (its ops route back through the active + ``apply`` fold), and refuse any *undeclared* callable arg loudly.""" + try: + bound = op.__signature__.bind(*args, **kwargs) + except TypeError: + return + bound.apply_defaults() + for name, p in op.__signature__.parameters.items(): + val = bound.arguments.get(name) + if _has(p.annotation, (_Computation,)): + if callable(val): + val( + _Opaque() + ) # run under the active interpreter -> its ops fold; loud if it inspects its arg + elif ( + callable(val) + and type(val) is not _Opaque + and not _has(p.annotation, (Uses,)) + ): + raise UndeclaredCallable( + f"{op}: argument {name!r} is callable but not declared `Computation`/`Uses[()]`; " + "its effects can't be soundly folded — annotate it, or the check is unsound." + ) + + +def _refuse(*_a: Any, **_k: Any) -> Any: + raise UnsoundCallbackFold( + "usesof ran a Computation callback on an opaque placeholder to collect its " + "effects, but the callback *inspected* its argument (compared it, did arithmetic on " + "it, called it, took its length, accessed an attribute, iterated or indexed it, …). " + "Folding on a fake value would path/structure-under-approximate; refusing rather " + "than under-approximating. Write the callback to pass its argument straight to " + "operations without inspecting it." + ) + + +class _Opaque: + """Placeholder fed to a ``Computation`` callback so its op-calls fire. It is meant to be + used only as opaque *data* — passed straight through to operations, which never inspect + their argument *values* under the fold. So ``lambda x: op()`` and ``lambda x: op(x)`` + fold correctly. + + It is a **best-effort tripwire, not a soundness guarantee.** Inspection that goes + through the *type*-level special-method (dunder) protocol — operators (``x + 1``, + ``x == 0``, ``x < y``), ``len``, ``bool``, calling (``x()``), *missing*-attribute access + (``x.field``), iteration, indexing, formatting — is bound to :func:`_refuse` and raises + :class:`UnsoundCallbackFold` *loudly*. But several checks bypass this protocol and + **cannot** be intercepted (a blanket ``__getattribute__`` override would also break the + fold's own ``isinstance(arg, Operation)`` dispatch), so a callback branching on them + silently under-approximates: object **identity** (``x is None``, ``id(x)``), the **type** + builtins (``type(x)``, ``isinstance(x, …)``, and ``callable(x)`` — which reads the + ``__call__`` bound below and so is always ``True``), and access to an **existing** + attribute (``x.__class__``). These are the general precondition restated — the fold is a + path-insensitive over-approximation of the *reified* term, so a callback doing native + control flow on its argument violates the precondition regardless. The tripwire catches + the dunder-protocol cases; the rest are the caller's responsibility: keep callbacks + straight-line — pass the argument to operations, don't inspect it. See + :func:`_INSPECTION_DUNDERS` for the surface covered.""" + + __slots__ = () + + +# Bind the operator/protocol surface a callback could reach through *type*-level dunder +# lookup to a loud refusal. Enumerated broadly so a missed operator raises rather than +# silently returning a wrong answer (e.g. the identity ``__eq__`` would return ``False`` and +# drop a branch). This does not — cannot — cover the identity/type/existing-attribute checks +# noted in :class:`_Opaque`. +_INSPECTION_DUNDERS: tuple[str, ...] = ( + # truth / hashing / formatting / conversions + "__bool__", + "__hash__", + "__eq__", + "__ne__", + "__repr__", + "__str__", + "__format__", + "__bytes__", + "__int__", + "__float__", + "__complex__", + "__index__", + "__round__", + "__trunc__", + "__floor__", + "__ceil__", + # ordering + "__lt__", + "__le__", + "__gt__", + "__ge__", + # missing-attribute access (only fires on a miss) / call + "__getattr__", + "__call__", + # container protocol + "__len__", + "__length_hint__", + "__contains__", + "__getitem__", + "__setitem__", + "__delitem__", + "__iter__", + "__next__", + "__reversed__", + # context / async + "__enter__", + "__exit__", + "__await__", + "__aiter__", + "__anext__", + # unary numeric + "__neg__", + "__pos__", + "__abs__", + "__invert__", +) +# binary numeric, plus reflected (r) and in-place (i) forms +for _binop in ( + "add", + "sub", + "mul", + "matmul", + "truediv", + "floordiv", + "mod", + "divmod", + "pow", + "lshift", + "rshift", + "and", + "xor", + "or", +): + _INSPECTION_DUNDERS += (f"__{_binop}__", f"__r{_binop}__", f"__i{_binop}__") + +for _dunder in _INSPECTION_DUNDERS: + setattr(_Opaque, _dunder, _refuse) diff --git a/effectful/ops/syntax.py b/effectful/ops/syntax.py index 764016752..2c03931fe 100644 --- a/effectful/ops/syntax.py +++ b/effectful/ops/syntax.py @@ -382,6 +382,58 @@ def extract_operations(obj): return bound_vars +class Uses: + """Effect-row annotation metadata: ``Annotated[T, Uses[op1, op2, ...]]`` on a + *return* type, declaring the operations a composite operation performs. ``Uses[()]`` + is the empty row — explicitly pure. Read by :meth:`Operation.__uses_rule__` (the + ``ε`` sibling of ``__type_rule__``), the same way :class:`Scoped` on an argument is + read by :meth:`Operation.__fvs_rule__`. + + Members are bare :class:`Operation` s or ``Literal[op]``. (Not an :class:`Annotation` + subtype: it is plain metadata on the return type, not a signature transform.) + """ + + __slots__ = ("members",) + + def __class_getitem__(cls, items: Any) -> "Uses": + return cls(items if isinstance(items, tuple) else (items,)) + + def __init__(self, members: tuple[Any, ...]) -> None: + self.members = members + + def __repr__(self) -> str: + return f"Uses{list(self.members)!r}" + + @staticmethod + def _member_ops(m: Any) -> frozenset[Operation]: + if typing.get_origin(m) is typing.Literal: + return frozenset(a for a in typing.get_args(m) if isinstance(a, Operation)) + if isinstance(m, Operation): + return frozenset({m}) + raise NotImplementedError( # loud, not a silent frozenset() drop + f"Uses member {m!r} is not supported: use `Literal[op]` or a bare `Operation`. " + "Polymorphic `Operation[[A], B]` members are not yet supported." + ) + + @classmethod + def declared(cls, sig: inspect.Signature) -> frozenset[Operation] | None: + """The ``Uses[...]`` row off a signature's return annotation, or ``None`` if no + ``Uses`` metadata is present (distinct from ``Uses[()]`` = present-and-empty = + explicitly pure). ``defop`` may wrap the return in a nested ``Annotated``, so this + flattens all metadata layers.""" + anno = sig.return_annotation + found: frozenset[Operation] | None = None + while typing.get_origin(anno) is Annotated: + args = typing.get_args(anno) + for meta in args[1:]: + if isinstance(meta, cls): + found = found or frozenset() + for m in meta.members: + found |= cls._member_ops(m) + anno = args[0] + return found + + defop = Operation.define diff --git a/effectful/ops/types.py b/effectful/ops/types.py index 46419d7a8..10dd90686 100644 --- a/effectful/ops/types.py +++ b/effectful/ops/types.py @@ -442,6 +442,29 @@ def __fvs_rule__(self, *args: Q.args, **kwargs: Q.kwargs) -> inspect.BoundArgume return result_sig + @typing.final + def __uses_rule__(self) -> "frozenset[Operation]": + """Returns the effect row this operation contributes *itself*: its declared + ``Uses[...]`` if the return type is annotated, else ``{self}`` (the operation + performs itself). + + The ``ε`` sibling of :meth:`__type_rule__` — where ``__type_rule__`` reads the + result *type* off the return annotation, ``__uses_rule__`` reads the declared + effect *row* off the same annotation's :class:`~effectful.ops.syntax.Uses` + metadata. Used by :func:`effectful.ops.semantics.usesof` to accumulate a term's + row over :func:`evaluate`, exactly as ``__fvs_rule__`` is used by :func:`fvsof`. + + Like the other rule methods this is ``@final`` and annotation-driven: a subclass + such as ``Tool`` declares its row by annotating its return type ``Uses[...]``, not + by overriding this method. The row is arg-independent (a composite op declares the + ops it performs, not which it performs per call); the arg-dependent part — folding + the effects of ``Computation`` callback arguments — is handled by ``usesof``. + """ + from effectful.ops.syntax import Uses + + declared = Uses.declared(self.__signature__) + return declared if declared is not None else frozenset({self}) + def __repr__(self): return f"{self.__class__.__name__}({self.__name__}, {self.__signature__})" diff --git a/tests/test_effects.py b/tests/test_effects.py new file mode 100644 index 000000000..701837e6a --- /dev/null +++ b/tests/test_effects.py @@ -0,0 +1,197 @@ +"""The effect-row engine (``ε``): ``usesof`` / ``Uses`` / ``Computation`` / ``Requires``.""" + +from collections.abc import Callable +from typing import Annotated + +import pytest + +from effectful.ops.effects import ( + Computation, + Requires, + UndeclaredCallable, + UnsoundCallbackFold, + check_requires, + check_uses, + effect_type, + usesof, +) +from effectful.ops.syntax import Uses, defop +from effectful.ops.types import NotHandled + + +@defop +def read() -> int: + raise NotHandled + + +@defop +def write(v: int) -> None: + raise NotHandled + + +@defop +def pure_add(a: int, b: int) -> Annotated[int, Uses[()]]: # declared pure + raise NotHandled + + +def test_usesof_is_the_op_row(): + assert usesof(write(pure_add(read(), read()))) == frozenset({read, write}) + + +def test_uses_empty_is_pure(): + # a Uses[()] op contributes nothing itself + assert pure_add not in usesof(pure_add(read(), read())) + + +def test_computation_callback_is_entered(): + @defop + def apply_cb(fn: Annotated[Callable[[int], int], Computation]) -> int: + raise NotHandled + + assert usesof(apply_cb(lambda x: read())) == frozenset({apply_cb, read}) + + +def test_computation_callback_ignoring_or_passing_arg_folds(): + @defop + def cb_pass(fn: Annotated[Callable[[int], None], Computation]) -> int: + raise NotHandled + + # passes the arg straight to an op (no inspection) — folds soundly + assert usesof(cb_pass(lambda x: write(x))) == frozenset({cb_pass, write}) + + +def test_computation_callback_inspecting_arg_is_refused_not_silent(): + @defop + def cb(fn: Annotated[Callable[[int], int], Computation]) -> int: + raise NotHandled + + # Every way of *inspecting* the arg must refuse loudly — never fold on a fake value + # (which would silently drop a branch) and never leak a raw TypeError. The refusal is + # default-deny, so these cover the operator families, not a hand-picked few. + inspecting = [ + lambda x: write(x) if x else read(), # truthiness branch + lambda x: ( + write(x) if x == 0 else read() + ), # __eq__ branch (identity would say False) + lambda x: write(x) if x < 1 else read(), # ordering branch + lambda x: write(x + 1), # arithmetic + lambda x: write(len(x)), # __len__ + lambda x: write(str(x)), # formatting/conversion + lambda x: x(), # calls the arg + lambda x: x.field, # missing-attribute access + lambda x: x[0], # indexing + ] + for cb_fn in inspecting: + with pytest.raises(UnsoundCallbackFold): + usesof(cb(cb_fn)) + + +def test_computation_callback_identity_and_type_inspection_is_a_known_unsound_hole(): + # KNOWN LIMITATION (documented, not a bug silently ignored): object identity (`is`/`id`), + # the type builtins (`type(x)`, `isinstance`), and access to an *existing* attribute + # (`x.__class__`) bypass the type-level dunder protocol, so `_Opaque` cannot intercept + # them (a blanket `__getattribute__` override would break the fold's own + # `isinstance(arg, Operation)` dispatch). A callback branching on them silently + # under-approximates. This is the general precondition restated: callbacks must be + # straight-line in their argument. The tripwire catches the dunder-protocol cases (see + # the test above); these it cannot. + @defop + def cb(fn: Annotated[Callable[[int], int], Computation]) -> int: + raise NotHandled + + # `x is None` is False for the placeholder, so only the else-branch folds: read is + # dropped. We assert the (unsound) status quo so the limitation is visible and pinned — + # if a future sound implementation closes it, this test flips and must be updated. + assert usesof(cb(lambda x: read() if x is None else write(x))) == frozenset( + {cb, write} + ) + # `type(x)` / `isinstance` likewise are not intercepted. + assert usesof(cb(lambda x: read() if type(x) is str else write(x))) == frozenset( + {cb, write} + ) + assert usesof( + cb(lambda x: read() if isinstance(x, str) else write(x)) + ) == frozenset({cb, write}) + # `callable(x)` reads the tripwire's own __call__ binding (True), not the real arg, so it + # takes the *then* branch — the opposite under-approximation, but the same class of hole. + assert usesof(cb(lambda x: read() if callable(x) else write(x))) == frozenset( + {cb, read} + ) + + +def test_undeclared_callable_fails_loudly(): + @defop + def bad(fn: Callable[[int], int]) -> int: # callable arg, not Computation/Uses[()] + raise NotHandled + + with pytest.raises(UndeclaredCallable): + usesof(bad(lambda x: x)) + + +def test_effect_type_pairs_tau_and_epsilon(): + tau, eps = effect_type(pure_add(read(), read())) + assert tau is int and eps == frozenset({read}) + + +def test_check_uses_flags_undeclared_effects(): + @defop + def declared() -> Annotated[int, Uses[read]]: # declares read only + raise NotHandled + + assert check_uses(declared, read()) == frozenset() # body ⊆ declared + assert check_uses(declared, write(read())) == frozenset({write}) # write undeclared + + +def test_requires_provenance(): + @defop + def sink(x: Annotated[int, Requires(read)]) -> None: + raise NotHandled + + assert check_requires(sink(read())) == {} # x came from read + assert check_requires(sink(pure_add(1, 1))) == {sink: {"x": frozenset({read})}} + + +def test_requires_provenance_holds_transitively(): + # provenance is the arg's whole effect row, so a required op reached *through* other + # ops still satisfies Requires. + @defop + def sink(x: Annotated[int, Requires(read)]) -> None: + raise NotHandled + + # read is under a pure combinator but still in x's row -> satisfied + assert check_requires(sink(pure_add(read(), 1))) == {} + + +def test_requires_reports_only_the_missing_ops(): + # Requires(read, write) on an arg that provides only read -> report just write. + @defop + def sink(x: Annotated[int, Requires(read, write)]) -> None: + raise NotHandled + + assert check_requires(sink(read())) == {sink: {"x": frozenset({write})}} + assert check_requires(sink(write(read()))) == {} # both present -> satisfied + + +def test_requires_is_per_argument_and_by_keyword(): + # multiple Requires on different params, passed by keyword; each checked independently. + @defop + def move( + src: Annotated[int, Requires(read)], + dst: Annotated[int, Requires(write)], + ) -> None: + raise NotHandled + + assert check_requires(move(src=read(), dst=write(1))) == {} + # dst lacks write in its provenance -> only dst flagged + assert check_requires(move(src=read(), dst=read())) == { + move: {"dst": frozenset({write})} + } + + +def test_requires_absent_annotation_is_unconstrained(): + # an argument with no Requires imposes no provenance obligation. + @defop + def sink(x: int) -> None: + raise NotHandled + + assert check_requires(sink(pure_add(1, 1))) == {} diff --git a/tests/test_handlers_llm_governance.py b/tests/test_handlers_llm_governance.py new file mode 100644 index 000000000..76a7f8bf3 --- /dev/null +++ b/tests/test_handlers_llm_governance.py @@ -0,0 +1,131 @@ +"""Static tool governance: ``toolsof`` / ``reachable_tools`` / ``check_tools`` — no LLM call. + +These check the static tool graph: which tools a template, or a function that calls one, +can reach — computed by reading ``.tools`` and by reifying to a Term, never running the LLM. +""" + +from effectful.handlers.llm.completions import LexicalReaders +from effectful.handlers.llm.governance import check_tools, reachable_tools, toolsof +from effectful.handlers.llm.template import Template, Tool +from effectful.internals.runtime import interpreter +from effectful.ops.semantics import apply, handler + + +def _trip_planner(): + """Build a template with a captured tool graph and a caller of it. + + Returns ``(suggest_city, delete_everything, my_fn)`` where ``suggest_city`` is a + template that lexically captures ``cities``/``weather``/``delete_everything``, and + ``my_fn`` is a plain function that calls the template. + """ + + @Tool.define + def cities() -> list[str]: + """Return a list of cities.""" + return ["Chicago", "Barcelona"] + + @Tool.define + def weather(city: str) -> str: + """Return the weather in a city.""" + return "sunny" + + @Tool.define + def delete_everything() -> None: + """Dangerous: wipe all state.""" + raise RuntimeError("boom") + + @Template.define + def suggest_city() -> str: + """Use the `cities` and `weather` tools to suggest a city.""" + raise NotImplementedError + + def my_fn() -> str: + return suggest_city() + + return suggest_city, delete_everything, my_fn + + +def test_toolsof_is_the_static_tool_graph(): + suggest_city, delete_everything, _ = _trip_planner() + reached = toolsof(suggest_city) + # every lexically-captured tool is reachable, including the dangerous one + assert delete_everything in reached + # the root itself is not one of the tools it reaches + assert suggest_city not in reached + + +def test_reachable_tools_sees_through_a_template_without_calling_the_llm(): + suggest_city, delete_everything, my_fn = _trip_planner() + reached = reachable_tools(my_fn) + # the template it calls, and (transitively) that template's captured tools + assert suggest_city in reached + assert delete_everything in reached + assert toolsof(suggest_city) <= reached + + +def test_reachable_tools_is_the_leak_check(): + # `reachable_tools(fn) <= declared` is the static tool-safety guarantee. + suggest_city, delete_everything, my_fn = _trip_planner() + declared = {suggest_city} | (toolsof(suggest_city) - {delete_everything}) + leak = reachable_tools(my_fn) - declared + assert leak == frozenset({delete_everything}) # flagged, LLM never called + + +def test_governed_tool_graph_excludes_synthetic_lexical_readers(): + # Synthetic LexicalReaders tools are prompt-variable plumbing, not tools an agent + # reaches. Law: the governed tool graph is invariant to whether readers are exposed — + # so a plain lexical value never gets counted as a reachable tool (and thus never + # flagged as a leak). + @Tool.define + def real_tool() -> int: + """A real tool.""" + return 0 + + favorite_city = "Paris" # a plain lexical value -> becomes a synthetic reader + + @Template.define + def t() -> str: + """Use {favorite_city} with the real_tool.""" + raise NotImplementedError + + baseline = toolsof(t) # no readers exposed + with handler(LexicalReaders()): + # guard: a synthetic reader really was created, snapshotting the lexical value + assert t.tools["favorite_city"]() == favorite_city + with_readers = toolsof(t) + + assert real_tool in baseline + assert with_readers == baseline # readers do not enter the governed graph + + +def test_check_tools_flags_the_leak(): + # check_tools = reachable_tools - allowed; the L2 tool-safety check (no LLM). + suggest_city, delete_everything, my_fn = _trip_planner() + allowed = {suggest_city} | (toolsof(suggest_city) - {delete_everything}) + assert check_tools(my_fn, *allowed) == frozenset({delete_everything}) + # allowing everything reachable -> no leak + assert check_tools(my_fn, *reachable_tools(my_fn)) == frozenset() + + +def test_reachable_tools_ignores_ambient_apply_handler(): + # Soundness law: static reachability must not depend on what is installed at the call + # site. With `handler` (merge) an ambient apply interpretation would win dispatch, so a + # called tool runs concretely instead of reifying and is silently missed. The reifier + # uses `interpreter` (replace), so the row is identical either way and the reified + # tool ops never reach the ambient handler. + suggest_city, delete_everything, my_fn = _trip_planner() + baseline = reachable_tools(my_fn) + + ran = [] + + def ambient(op, *a, **k): # a valid ambient interpretation (concrete execution) + ran.append(op) + return op.__default_rule__(*a, **k) + + with interpreter({apply: ambient}): + under_ambient = reachable_tools(my_fn) + + assert under_ambient == baseline # row unaffected by the ambient handler + assert delete_everything in under_ambient + # reification isolated the tool calls — none of the trip tools executed concretely + assert not ({suggest_city, delete_everything} & set(ran))