faultgate-zero 0.2.0: check plugins and formats.load

.github/workflows/ci.yml

@@ -20,7 +20,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install test dependencies
-        run: python -m pip install -e ".[dev,compat]"
+        run: python -m pip install -e ".[dev,compat,formats]"
       - name: Run tests
         env:
           LITELLM_LOCAL_MODEL_COST_MAP: "True"

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 0.2.0 (unreleased)
+
+Adds features only; everything in 0.1.0 works as before.
+
+- Check plugins: the `faultgate_zero.checks` entry point group. A plugin object has `name`, `default`
+  and `run(trace, options) -> Verdict`; default plugins join `DEFAULT_CHECKS`. A name clash with a
+  built-in check is a usage error at startup; a plugin that raises gives a verdict that did not fire
+  (`check crashed: <ExceptionType>`), a `FaultgateZeroWarning` and a summary line.
+- `faultgate_zero.formats.load(path, fmt="auto")`: OTLP through faultgate-zero's own loader, and
+  OpenInference, Langfuse, LangSmith and Braintrust through faultgate's loaders, behind the new
+  `faultgate-zero[formats]` extra. `formats.to_zero()` converts a faultgate trace.
+
 ## 0.1.0
 
 First usable version.

README.md

@@ -8,7 +8,7 @@ asserts miss: loops on empty results, repeated identical calls, tool errors the
 provider errors it hid behind a fallback message, answers cut at the token limit, and tool calls that
 break your policy (refund limits, identity checks, required steps, denied paths).
 
-> **Status:** 0.1.0 alpha. Trace-based checks for existing pytest tests; no dependency on the
+> **Status:** 0.2.0 alpha. Trace-based checks for existing pytest tests; no dependency on the
 > `faultgate` package. See the [known limitations](docs/known-limitations.md).
 
 ## 60-second quickstart
@@ -203,6 +203,31 @@ Paths on the command line are relative to where pytest runs; in `pyproject.toml`
 With pytest-xdist (`-n`), workers send their results to the controller and one merged report is
 written.
 
+## Check plugins
+
+Other packages can add checks. A plugin registers an object in the `faultgate_zero.checks` entry point
+group:
+
+```toml
+[project.entry-points."faultgate_zero.checks"]
+contract = "fault_contract.zero_check:CHECK"
+```
+
+The object has three members: `name` (str), `default` (bool: whether it joins the default checks) and
+`run(trace, options) -> Verdict`. Plugins are discovered once, when faultgate-zero loads; installing one
+is enough. A reason made of several parts joined by `; ` prints one part per line, as policy
+reasons do. A plugin check works like a built-in one: `--fgz-checks`, `@pytest.mark.faultgate(skip=[...])`,
+the modes and the reports. A plugin named like a built-in check, or two plugins with one name, stop
+the session with a usage error. A plugin that raises gives a verdict that did not fire (`check crashed:
+<ExceptionType>`), a `FaultgateZeroWarning` and a "check crashed" line in the summary: your test is
+never failed by a bug in a plugin.
+
+## Reading other trace formats
+
+`faultgate_zero.formats.load(path, fmt="auto")` returns the traces in an export file. OTLP/JSON is read
+by faultgate-zero itself. OpenInference, Langfuse, LangSmith and Braintrust exports are read by
+faultgate's loaders: `pip install 'faultgate-zero[formats]'`.
+
 ## Supported SDKs and frameworks
 
 Tested here, with offline fakes that run the SDKs' own request and response code:
@@ -222,7 +247,7 @@ are not claimed until a test covers them. See [docs/known-limitations.md](docs/k
 ## Development
 
 ```bash
-pip install -e ".[dev,compat]"
+pip install -e ".[dev,compat,formats]"
 pytest                                 # the repo's own suite; its pyproject turns the plugin off for itself
 ```
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "faultgate-zero"
-version = "0.1.0"
+version = "0.2.0"
 description = "Agent failure checks inside your existing pytest tests: loops, repeated calls, unrecovered errors, token cutoffs and policy breaks."
 readme = "README.md"
 license = "MIT"
@@ -15,6 +15,8 @@ classifiers = ["Development Status :: 3 - Alpha", "Framework :: Pytest", "Progra
 dependencies = ["pytest>=8"]
 
 [project.optional-dependencies]
+# every trace format through faultgate's loaders (formats.load); OTLP needs nothing extra
+formats = ["faultgate>=0.4,<0.5"]
 dev = ["openai>=2", "anthropic>=0.40", "pytest-xdist>=3"]
 # framework compatibility tests (tests/test_compat.py); litellm: CTO licence exception, test-only
 compat = ["litellm>=1.70", "langchain-openai>=0.3", "langchain-anthropic>=0.3",

src/faultgate_zero/__init__.py

@@ -1,5 +1,5 @@
 """faultgate-zero: agent failure checks that run inside your existing pytest tests."""
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 
 from .patch import install, uninstall  # noqa: E402
 from .policy import Policy, PolicyError  # noqa: E402

src/faultgate_zero/checks.py

@@ -6,9 +6,11 @@
 from __future__ import annotations
 
 import re
+import warnings
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import Callable
+from importlib import metadata
+from typing import Any, Callable, Iterable
 
 from .policy import Policy
 from .results import error_of, is_empty
@@ -145,13 +147,68 @@ def policy_check(t: Trace, policy: Policy) -> Verdict:
     "context_overflow": context_overflow,
     "abstention": abstention,
 }
-CHECK_NAMES = (*RULES, "policy")
-DEFAULT_CHECKS = ("search_loop", "repeated_call", "unrecovered_tool_error", "unrecovered_model_error",
-                  "context_overflow", "policy")
+BUILTIN_DEFAULTS = ("search_loop", "repeated_call", "unrecovered_tool_error", "unrecovered_model_error",
+                    "context_overflow", "policy")
+
+# Check plugins: other packages add checks through the ``faultgate_zero.checks`` entry point group.
+# Each entry point loads an object with ``name: str``, ``default: bool`` and
+# ``run(trace: Trace, options: Options) -> Verdict``. Discovery happens once, when this module loads.
+PLUGIN_GROUP = "faultgate_zero.checks"
+CRASHED = "check crashed: "
+
+
+class PluginError(ValueError):
+    """A check plugin that cannot be used: its name clashes with a built-in check, or it is malformed."""
+
+
+def discover(eps: Iterable[Any] | None = None) -> tuple[dict[str, Any], list[str]]:
+    """Load check plugins from entry points. Returns (plugins by name, errors). Never raises."""
+    if eps is None:
+        try:
+            eps = metadata.entry_points(group=PLUGIN_GROUP)
+        except Exception as exc:  # a broken install must not stop faultgate-zero from loading
+            return {}, [f"cannot read {PLUGIN_GROUP} entry points: {type(exc).__name__}"]
+    found: dict[str, Any] = {}
+    errors: list[str] = []
+    builtin = (*RULES, "policy")
+    for ep in eps:
+        where = f"check plugin {getattr(ep, 'name', '?')!r} ({getattr(ep, 'value', '?')})"
+        try:
+            obj = ep.load()
+        except Exception as exc:
+            errors.append(f"{where}: cannot load it: {type(exc).__name__}: {exc}")
+            continue
+        name = getattr(obj, "name", None)
+        if not isinstance(name, str) or not name or not callable(getattr(obj, "run", None))                 or not isinstance(getattr(obj, "default", None), bool):
+            errors.append(f"{where}: needs name (str), default (bool) and run(trace, options)")
+            continue
+        if name in builtin:
+            errors.append(f"{where}: name {name!r} clashes with a built-in check")
+            continue
+        if name in found:
+            errors.append(f"{where}: name {name!r} is used by two plugins")
+            continue
+        found[name] = obj
+    return found, errors
+
+
+PLUGINS, PLUGIN_ERRORS = discover()
+CHECK_NAMES = (*RULES, "policy", *PLUGINS)
+DEFAULT_CHECKS = (*BUILTIN_DEFAULTS, *(n for n, p in PLUGINS.items() if p.default))
+
+
+def plugin_errors() -> list[str]:
+    return list(PLUGIN_ERRORS)
+
+
+def _raise_plugin_errors() -> None:
+    if PLUGIN_ERRORS:
+        raise PluginError(PLUGIN_ERRORS[0])
 
 
 def resolve(names: str | list[str] | tuple[str, ...] | None) -> tuple[str, ...]:
     """Check names from a comma list or sequence; None means the default set. Unknown names are a usage error."""
+    _raise_plugin_errors()
     if names is None:
         return DEFAULT_CHECKS
     wanted = [n.strip() for n in (names.split(",") if isinstance(names, str) else names) if n.strip()]
@@ -161,8 +218,37 @@ def resolve(names: str | list[str] | tuple[str, ...] | None) -> tuple[str, ...]:
     return tuple(dict.fromkeys(wanted))
 
 
+def run_plugin(plugin: Any, t: Trace, options: Options) -> Verdict:
+    """Run one plugin check. A plugin that raises, or returns something else than a Verdict, gives a
+    verdict that did not fire and a warning: the user's test never fails because of a plugin bug."""
+    from .recorder import FaultgateZeroWarning
+
+    name = plugin.name
+    try:
+        v = plugin.run(t, options)
+        if not isinstance(v, Verdict):
+            raise TypeError(f"run() returned {type(v).__name__}, not Verdict")
+    except Exception as exc:
+        try:  # under -W error the warning raises; it must not fail the test either
+            warnings.warn(f"faultgate-zero: check {name!r} crashed ({type(exc).__name__}); it did not fire",
+                          FaultgateZeroWarning, stacklevel=2)
+        except Exception:
+            pass
+        return Verdict(name, False, f"{CRASHED}{type(exc).__name__}")
+    return v if v.check == name else Verdict(name, v.fired, v.reason, v.step)
+
+
+def crashed(v: Verdict) -> bool:
+    return not v.fired and v.reason.startswith(CRASHED)
+
+
 def run_checks(t: Trace, names: tuple[str, ...], policy: Policy, options: Options) -> list[Verdict]:
     out = []
     for n in names:
-        out.append(policy_check(t, policy) if n == "policy" else RULES[n](t, options))
+        if n == "policy":
+            out.append(policy_check(t, policy))
+        elif n in RULES:
+            out.append(RULES[n](t, options))
+        else:
+            out.append(run_plugin(PLUGINS[n], t, options))
     return out

src/faultgate_zero/formats.py

@@ -0,0 +1,139 @@
+"""Read any trace export as faultgate-zero traces: ``load(path, fmt="auto")``.
+
+OTLP/JSON goes through faultgate-zero's own loader. Every other format (OpenInference, Langfuse,
+LangSmith, Braintrust) is read by faultgate's loaders, imported only when needed (install
+``faultgate-zero[formats]``), then converted: both packages use the GenAI attribute names, so the
+conversion copies spans as they are.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from . import otlp
+from .trace import Span, Trace
+
+FORMATS = ("auto", "otlp", "openinference", "langfuse", "langsmith", "braintrust")
+
+
+class FormatError(ValueError):
+    """An export that cannot be read: unknown format, bad JSON, or a format that needs faultgate."""
+
+
+MESSAGE_ATTRS = ("gen_ai.input.messages", "gen_ai.output.messages")
+
+
+def _part_from_openai(p: Any) -> dict[str, Any] | None:
+    if isinstance(p, str):
+        return {"type": "text", "content": p}
+    if isinstance(p, dict) and p.get("type") in ("text", "input_text", "output_text"):
+        return {"type": "text", "content": str(p.get("text", p.get("content", "")))}
+    return None
+
+
+def normalize_messages(raw: Any) -> Any:
+    """Messages as GenAI ``parts``, which is what faultgate-zero reads. faultgate writes plain chat
+    messages (``{"role", "content"}``); their text becomes text parts. Tool calls are read the way
+    faultgate reads them, from parts and tool spans only, so a call is never counted twice.
+    Messages that already have parts are kept."""
+    if isinstance(raw, str):
+        try:
+            raw = json.loads(raw)
+        except ValueError:
+            return raw
+    if not isinstance(raw, list):
+        return raw
+    out = []
+    for m in raw:
+        if not isinstance(m, dict) or isinstance(m.get("parts"), list):
+            out.append(m)
+            continue
+        content = m.get("content")
+        parts: list[dict[str, Any]] = []
+        if isinstance(content, list):
+            parts = [x for x in map(_part_from_openai, content) if x]
+        elif content not in (None, ""):
+            parts = [{"type": "text", "content": str(content)}]
+        out.append({**{k: v for k, v in m.items() if k != "content"}, "parts": parts})
+    return out
+
+
+def to_zero(t: Any) -> Trace:
+    """Convert a faultgate Trace to a faultgate-zero Trace: spans as they are, messages as parts."""
+    spans = []
+    for s in t.spans:
+        attrs = dict(s.attrs)
+        for k in MESSAGE_ATTRS:
+            if k in attrs:
+                attrs[k] = normalize_messages(attrs[k])
+        spans.append(Span(span_id=s.span_id, parent_id=s.parent_id, name=s.name, start_ns=s.start_ns,
+                          end_ns=s.end_ns, attrs=attrs, error=s.status_code == 2, status_message=s.status_message))
+    root = t.root
+    attrs = root.attrs if root else {}
+    return Trace(trace_id=t.trace_id, scenario=str(attrs.get("faultgate.scenario") or t.key),
+                 agent=t.agent, user_id=attrs.get("enduser.id"), spans=spans)
+
+
+def _is_plain_otlp(doc: Any) -> bool:
+    if not (isinstance(doc, dict) and "resourceSpans" in doc):
+        return False
+    for rs in _dicts(doc.get("resourceSpans")):
+        for ss in _dicts(rs.get("scopeSpans")):
+            for s in _dicts(ss.get("spans")):
+                if any(a.get("key") == "openinference.span.kind" for a in _dicts(s.get("attributes"))):
+                    return False
+    return True
+
+
+def _dicts(v: Any) -> list[dict[str, Any]]:
+    return [x for x in v if isinstance(x, dict)] if isinstance(v, list) else []
+
+
+def detect(doc: Any) -> str:
+    if _is_plain_otlp(doc):
+        return "otlp"
+    return _faultgate().detect(doc)
+
+
+def _faultgate() -> Any:
+    try:
+        from faultgate import formats as fg
+    except ImportError:
+        raise FormatError("this trace format needs faultgate's loaders: pip install 'faultgate-zero[formats]'") from None
+    return fg
+
+
+def load(path: str | Path, fmt: str = "auto") -> list[Trace]:
+    """Every trace in one export file. Raises FormatError (a ValueError) on input it cannot read."""
+    if fmt not in FORMATS:
+        raise FormatError(f"unknown format {fmt!r}; available: {', '.join(FORMATS)}")
+    p = Path(path)
+    try:
+        doc = json.loads(p.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise FormatError(f"{p}: cannot read it ({type(exc).__name__})") from None
+    except ValueError:
+        raise FormatError(f"{p}: not valid JSON") from None
+    if fmt == "auto":
+        try:
+            fmt = detect(doc)
+        except FormatError:
+            raise
+        except ValueError as exc:
+            raise FormatError(f"{p}: {exc}") from None
+        except (KeyError, TypeError, AttributeError):
+            raise FormatError(f"{p}: unrecognised trace format") from None
+    if fmt == "otlp":
+        try:
+            return otlp.from_dict(doc)
+        except (KeyError, TypeError, AttributeError, ValueError) as exc:
+            raise FormatError(f"{p}: not a valid OTLP/JSON export ({type(exc).__name__})") from None
+    try:
+        return [to_zero(t) for t in _faultgate().load(p, fmt)]
+    except FormatError:
+        raise
+    except ValueError as exc:
+        raise FormatError(f"{p}: {exc}") from None
+    except (KeyError, TypeError, AttributeError):
+        raise FormatError(f"{p}: not a valid {fmt} export") from None

src/faultgate_zero/pytest_plugin.py

@@ -8,7 +8,7 @@
 
 import pytest
 
-from . import otlp, patch, recorder, runner
+from . import checks, otlp, patch, recorder, runner
 from . import report as rpt
 from .config import Config, ConfigError, for_marker, resolve_session
 from .policy import PolicyError
@@ -254,6 +254,11 @@ def pytest_terminal_summary(terminalreporter, exitstatus: int, config: pytest.Co
     tr.write_line(line)
     for name, n in s["fired"].items():
         tr.write_line(f"  {name}: {n}")
+    crashes = sum(1 for e in state.entries for v in e["verdicts"]
+                  if not v["fired"] and str(v["reason"]).startswith(checks.CRASHED))
+    if crashes:
+        tr.write_line(f"faultgate-zero: {crashes} check{'s' if crashes > 1 else ''} crashed and did not fire "
+                      "(a bug in a check plugin; see the warnings)")
     for label, path in (("JSON report", cfg.json), ("Markdown summary", cfg.markdown), ("OTLP export", cfg.export_otlp)):
         if path:
             tr.write_line(f"faultgate-zero: {label} written to {path}")