feat: add model-output corruption effects

AGENTS.md

@@ -96,7 +96,7 @@ strands-evals/
 │   │   │                                     # ValidationError / TruncateFields /
 │   │   │                                     # RemoveFields / CorruptValues
 │   │   ├── experiment.py                     # ChaosExperiment (sets active case via ContextVar)
-│   │   ├── plugin.py                         # ChaosPlugin (BeforeToolCallEvent / AfterToolCallEvent)
+│   │   ├── plugin.py                         # ChaosPlugin (tool + model hooks via ContextVar)
 │   │   └── _context.py                       # ContextVar holding the active ChaosCase
 │   │
 │   ├── experimental/                         # Stable public API, evolving surface

pyproject.toml

@@ -16,7 +16,7 @@ authors = [
 dependencies = [
     "pydantic>=2.4.0,<3.0.0",
     "rich>=14.0.0,<15.0.0",
-    "strands-agents>=1.42.0",
+    "strands-agents>=1.45.0",
     "strands-agents-tools>=0.1.0,<1.0.0",
     "typing-extensions>=4.13.2,<5.0.0",
     "opentelemetry-api>=1.20.0",

src/strands_evals/chaos/__init__.py

@@ -7,10 +7,17 @@
 from .case import ChaosCase
 from .effects import (
     ChaosEffect,
+    Confabulation,
     CorruptValues,
+    EmptyResponse,
     ExecutionError,
+    FullRefusal,
+    MalformedJson,
+    ModelEffect,
+    ModelEffectUnion,
     NetworkError,
     RemoveFields,
+    SuccessFraming,
     Timeout,
     ToolEffect,
     ToolEffectUnion,
@@ -38,4 +45,11 @@
     "TruncateFields",
     "RemoveFields",
     "CorruptValues",
+    "ModelEffect",
+    "ModelEffectUnion",
+    "MalformedJson",
+    "EmptyResponse",
+    "Confabulation",
+    "FullRefusal",
+    "SuccessFraming",
 ]

src/strands_evals/chaos/case.py

@@ -6,13 +6,17 @@
 """
 
 import uuid
+from typing import cast
 
 from pydantic import Field, model_validator
 from typing_extensions import Generic
 
 from ..case import Case
 from ..types.evaluation import InputT, OutputT
-from .effects import ToolEffectUnion
+from .effects import ModelEffect, ModelEffectUnion, ToolEffect, ToolEffectUnion
+
+# Type alias for the effects dict structure
+EffectsDict = dict[str, dict[str, list[ToolEffectUnion | ModelEffectUnion]]]
 
 
 class ChaosCase(Case, Generic[InputT, OutputT]):
@@ -26,16 +30,26 @@ class ChaosCase(Case, Generic[InputT, OutputT]):
     ChaosExperiment.
 
     Attributes:
-        effects: A dict keyed by effect category. Currently supports
-            ``"tool_effects"`` mapping tool_name -> list of effects.
+        effects: A dict keyed by effect category. Supports ``"tool_effects"``
+            mapping tool_name -> list of effects, and ``"model_effects"``
+            mapping model_name (or ``"*"`` wildcard) -> list of effects.
 
     Example::
 
         from strands_evals import Case
         from strands_evals.chaos import ChaosCase
-        from strands_evals.chaos.effects import Timeout, TruncateFields
+        from strands_evals.chaos.effects import FullRefusal, Timeout, TruncateFields
+
+        # Direct construction with model effects
+        chaos_case = ChaosCase(
+            name="refusal_test",
+            input="Tell me something",
+            effects={
+                "model_effects": {"*": [FullRefusal()]},
+            },
+        )
 
-        # Direct construction
+        # Direct construction with tool effects
         chaos_case = ChaosCase(
             name="search_timeout",
             input="Find flights to Tokyo",
@@ -55,35 +69,73 @@ class ChaosCase(Case, Generic[InputT, OutputT]):
         # Produces 6 ChaosCase objects: 2 cases × (2 effect maps + 1 baseline)
     """
 
-    effects: dict[str, dict[str, list[ToolEffectUnion]]] = Field(
+    effects: EffectsDict = Field(
         default_factory=dict,
-        description="Effect categories. Currently supports 'tool_effects' mapping "
-        "tool_name -> list of effects. Empty dict means baseline (no chaos).",
+        description="Effect categories. Supports 'tool_effects' mapping "
+        "tool_name -> list of effects, and 'model_effects' mapping "
+        "model_name (or '*' wildcard) -> list of effects. "
+        "Empty dict means baseline (no chaos).",
     )
 
     @model_validator(mode="after")
-    def _validate_tool_effects(self) -> "ChaosCase":
-        """Validate tool effects configuration."""
-        allowed_categories = {"tool_effects"}
+    def _validate_effects(self) -> "ChaosCase":
+        """Validate effects configuration structure."""
+        allowed_categories = {"tool_effects", "model_effects"}
         unknown = set(self.effects.keys()) - allowed_categories
         if unknown:
             raise ValueError(
                 f"Unknown effect categories: {sorted(unknown)}. Allowed categories: {sorted(allowed_categories)}."
             )
 
+        # Validate tool_effects: dict[str, list[ToolEffectUnion]]
         for tool_name, effects_list in self.tool_effects.items():
             if len(effects_list) > 1:
                 raise ValueError(
                     f"Tool '{tool_name}' has {len(effects_list)} effects — only 1 is allowed per "
                     f"ChaosCase. Use separate ChaosCase instances to test effects independently."
                 )
+            # Fix A: enforce effect-family membership
+            for effect in effects_list:
+                if not isinstance(effect, ToolEffect):
+                    raise ValueError(
+                        f"Effect {type(effect).__name__} in tool_effects['{tool_name}'] is not a {ToolEffect.__name__}"
+                    )
+
+        # Validate model_effects: dict[str, list[ModelEffectUnion]]
+        model_effects_map = self.effects.get("model_effects", {})
+        if model_effects_map:
+            if not isinstance(model_effects_map, dict):
+                raise ValueError("'model_effects' must be a dict keyed by model name (or '*' wildcard).")
+            # Fix B: reject non-"*" keys
+            for model_name in model_effects_map:
+                if model_name != "*":
+                    raise ValueError(
+                        f"model_effects key '{model_name}' is not supported; "
+                        f"model targeting not yet implemented. Use '*' for all models."
+                    )
+            for model_name, effects_list in model_effects_map.items():  # type: ignore[assignment]
+                if not isinstance(model_name, str):
+                    raise ValueError(f"model_effects keys must be strings, got {type(model_name).__name__}.")
+                if not isinstance(effects_list, list):
+                    raise ValueError(
+                        f"model_effects['{model_name}'] must be a list of model effects, "
+                        f"got {type(effects_list).__name__}."
+                    )
+                # Fix A: enforce effect-family membership
+                for effect in effects_list:
+                    if not isinstance(effect, ModelEffect):
+                        raise ValueError(
+                            f"Effect {type(effect).__name__} in model_effects['{model_name}'] "
+                            f"is not a {ModelEffect.__name__}"
+                        )
+
         return self
 
     @classmethod
     def expand(
         cls,
         cases: list[Case],
-        effect_maps: dict[str, dict[str, dict[str, list[ToolEffectUnion]]]],
+        effect_maps: dict[str, EffectsDict],
         include_no_effect_baseline: bool = False,
     ) -> list["ChaosCase"]:
         """Generate the Cartesian product of cases × named effect maps.
@@ -96,14 +148,17 @@ def expand(
             cases: Base test cases to expand.
             effect_maps: Named effect configurations. Keys are short human-readable
                 names (used in the composite case name); values are dicts keyed by
-                effect category (e.g. ``"tool_effects"``) mapping tool_name -> list
-                of effect instances.
+                effect category (e.g. ``"tool_effects"``, ``"model_effects"``)
+                mapping target -> list of effect instances.
                 Example::
 
                     {
                         "search_timeout": {
                             "tool_effects": {"search_tool": [Timeout()]}
                         },
+                        "refusal": {
+                            "model_effects": {"*": [FullRefusal()]}
+                        },
                     }
             include_no_effect_baseline: If True, includes a baseline (no chaos)
                 variant for each case. Defaults to False.
@@ -112,7 +167,7 @@ def expand(
             Flat list of ChaosCase objects with composite names like
             "flight_search|baseline" or "flight_search|search_timeout".
         """
-        all_entries: list[tuple[str, dict[str, dict[str, list[ToolEffectUnion]]]]] = []
+        all_entries: list[tuple[str, EffectsDict]] = []
 
         if include_no_effect_baseline:
             all_entries.append(("baseline", {}))
@@ -146,10 +201,23 @@ def expand(
     @property
     def tool_effects(self) -> dict[str, list[ToolEffectUnion]]:
         """Convenience accessor for effects['tool_effects']."""
-        return self.effects.get("tool_effects", {})
+        return cast(dict[str, list[ToolEffectUnion]], self.effects.get("tool_effects", {}))
+
+    @property
+    def model_effects(self) -> list[ModelEffectUnion]:
+        """Resolve model effects. '*' wildcard applies to all models."""
+        model_effects_map = self.effects.get("model_effects", {})
+        if not model_effects_map:
+            return []
+        # For now, resolve "*" (wildcard = applies to all models)
+        return cast(list[ModelEffectUnion], model_effects_map.get("*", []))
 
     def __repr__(self) -> str:
         effects_str = ", ".join(
             f"{target}: [{', '.join(type(e).__name__ for e in effs)}]" for target, effs in self.tool_effects.items()
         )
-        return f"ChaosCase(name='{self.name}', effects={{{effects_str}}})"
+        parts = [f"name='{self.name}'", f"effects={{{effects_str}}}"]
+        if self.model_effects:
+            model_str = ", ".join(type(e).__name__ for e in self.model_effects)
+            parts.append(f"model_effects=[{model_str}]")
+        return f"ChaosCase({', '.join(parts)})"