diff --git a/AGENTS.md b/AGENTS.md index 78880852..e8e1bb26 100644 --- a/AGENTS.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 80265ff0..5c3490b0 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/strands_evals/chaos/__init__.py b/src/strands_evals/chaos/__init__.py index 8670012d..cfd78b4d 100644 --- a/src/strands_evals/chaos/__init__.py +++ b/src/strands_evals/chaos/__init__.py @@ -4,16 +4,18 @@ under tool failures and response corruption scenarios. """ -from .case import ChaosCase +from .case import ChaosCase, ChaosEffects from .effects import ( - ChaosEffect, + Confabulation, CorruptValues, + EmptyResponse, ExecutionError, + FullRefusal, + MalformedJson, NetworkError, RemoveFields, + SuccessFraming, Timeout, - ToolEffect, - ToolEffectUnion, TruncateFields, ValidationError, ) @@ -23,19 +25,21 @@ __all__ = [ # Core classes "ChaosCase", + "ChaosEffects", "ChaosExperiment", "ChaosPlugin", - # Effect hierarchy - "ChaosEffect", - "ToolEffect", - "ToolEffectUnion", - # Pre-hook effects (tool call failures) + # Tool effects "Timeout", "NetworkError", "ExecutionError", "ValidationError", - # Post-hook effects (response corruption) "TruncateFields", "RemoveFields", "CorruptValues", + # Model effects + "MalformedJson", + "EmptyResponse", + "Confabulation", + "FullRefusal", + "SuccessFraming", ] diff --git a/src/strands_evals/chaos/case.py b/src/strands_evals/chaos/case.py index 29d6de9d..82a0f785 100644 --- a/src/strands_evals/chaos/case.py +++ b/src/strands_evals/chaos/case.py @@ -6,13 +6,23 @@ """ import uuid +from typing import Literal, TypedDict -from pydantic import Field, model_validator +from pydantic import ConfigDict, 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 ModelEffectUnion, ToolEffectUnion + + +class ChaosEffects(TypedDict, total=False): + """Typed schema for chaos effects configuration.""" + + __pydantic_config__ = ConfigDict(extra="forbid") # type: ignore[misc] + + tool_effects: dict[str, list[ToolEffectUnion]] + model_effects: dict[Literal["*"], list[ModelEffectUnion]] class ChaosCase(Case, Generic[InputT, OutputT]): @@ -26,16 +36,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 ``"*"`` 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 + # Direct construction with model effects + chaos_case = ChaosCase( + name="refusal_test", + input="Tell me something", + effects={ + "model_effects": {"*": [FullRefusal()]}, + }, + ) + + # Direct construction with tool effects chaos_case = ChaosCase( name="search_timeout", input="Find flights to Tokyo", @@ -55,35 +75,46 @@ 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( - default_factory=dict, - description="Effect categories. Currently supports 'tool_effects' mapping " - "tool_name -> list of effects. Empty dict means baseline (no chaos).", + effects: ChaosEffects = Field( + default_factory=ChaosEffects, + description="Effect categories. Supports 'tool_effects' mapping " + "tool_name -> list of effects, and 'model_effects' mapping " + "'*' 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"} - unknown = set(self.effects.keys()) - allowed_categories - if unknown: - raise ValueError( - f"Unknown effect categories: {sorted(unknown)}. Allowed categories: {sorted(allowed_categories)}." - ) + def _validate_effects(self) -> "ChaosCase": + """Validate behavioral constraints the type system cannot express.""" + self._validate_tool_effects() + self._validate_pre_model_effects() + return self + def _validate_tool_effects(self) -> None: + """At most one effect per tool.""" 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." ) - return self + + def _validate_pre_model_effects(self) -> None: + """At most one pre-hook model effect: pre effects cancel the model call, so only one can win.""" + pre_effects = [e for e in self.model_effects if e.hook == "pre"] + if len(pre_effects) > 1: + names = ", ".join(type(e).__name__ for e in pre_effects) + raise ValueError( + f"model_effects has {len(pre_effects)} pre-hook effects ({names}) — only 1 is allowed per " + f"ChaosCase. Pre-hook effects cancel the model call, so only one can take effect. " + f"Use separate ChaosCase instances to test them independently." + ) @classmethod def expand( cls, cases: list[Case], - effect_maps: dict[str, dict[str, dict[str, list[ToolEffectUnion]]]], + effect_maps: dict[str, ChaosEffects], include_no_effect_baseline: bool = False, ) -> list["ChaosCase"]: """Generate the Cartesian product of cases × named effect maps. @@ -96,14 +127,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 +146,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, ChaosEffects]] = [] if include_no_effect_baseline: all_entries.append(("baseline", {})) @@ -148,8 +182,20 @@ def tool_effects(self) -> dict[str, list[ToolEffectUnion]]: """Convenience accessor for effects['tool_effects'].""" return 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 [] + return 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)})" diff --git a/src/strands_evals/chaos/effects.py b/src/strands_evals/chaos/effects.py index 7a22dd1b..c3cb547d 100644 --- a/src/strands_evals/chaos/effects.py +++ b/src/strands_evals/chaos/effects.py @@ -11,8 +11,10 @@ discriminated-union serialization, ensuring full round-trip fidelity. """ +import json import math import random +import re from abc import abstractmethod from typing import Annotated, Any, ClassVar, Literal, Union @@ -43,11 +45,6 @@ class ToolEffect(ChaosEffect): """ -# --------------------------------------------------------------------------- -# Pre-hook effects: cancel the tool call before execution -# --------------------------------------------------------------------------- - - class Timeout(ToolEffect): """Simulates a tool call timeout. @@ -152,11 +149,6 @@ def apply(self, context: Any = None) -> str: return self.error_message -# --------------------------------------------------------------------------- -# Post-hook effects: corrupt the tool response after execution -# --------------------------------------------------------------------------- - - class TruncateFields(ToolEffect): """Truncates string values in the tool response. @@ -308,10 +300,6 @@ def apply(self, response: Any = None) -> Any: return result -# --------------------------------------------------------------------------- -# Discriminated union type for Pydantic serialization -# --------------------------------------------------------------------------- - ToolEffectUnion = Annotated[ Union[ Annotated[Timeout, Tag("timeout")], @@ -329,3 +317,202 @@ def apply(self, response: Any = None) -> Any: Used in ChaosCase.effects to ensure full round-trip serialization fidelity with Pydantic's model_dump() / model_validate(). """ + + +class ModelEffect(ChaosEffect): + """Effect that operates on model output content. + + Intermediate class parallel to ToolEffect. Enables type-based dispatch + so the plugin can distinguish model-output effects from tool-level effects. + """ + + hook: ClassVar[Literal["pre", "post"]] = "post" + + +class MalformedJson(ModelEffect): + """Corrupts JSON structures in model output.""" + + hook: ClassVar[Literal["pre", "post"]] = "post" + effect_type: Literal["malformed_json"] = "malformed_json" + + def apply(self, content: Any = None) -> Any: + if content is None: + raise ValueError("MalformedJson.apply() requires content") + if isinstance(content, str): + return self.malform_text(content) + elif isinstance(content, list): + return self._malform_blocks(content) + raise ValueError(f"MalformedJson.apply() received unsupported type {type(content).__name__}") + + @staticmethod + def malform_text(text: str) -> str: + """Corrupt JSON-like text — the ONE place text malformation lives.""" + stripped = text.strip() + if stripped.startswith("{") or stripped.startswith("["): + return stripped[: len(stripped) // 2] + return text + + @staticmethod + def malform_tool_use_block(block: dict) -> dict: + """Corrupt a single toolUse block's input JSON — the ONE place this logic lives.""" + block = dict(block) + tool_use = dict(block["toolUse"]) + raw = json.dumps(tool_use.get("input", {})) + tool_use["input"] = raw[:-1] if raw.endswith("}") else raw + "{{{" + block["toolUse"] = tool_use + return block + + @staticmethod + def _malform_blocks(blocks: list) -> list: + """Apply malformation to all blocks — delegates toolUse corruption to malform_tool_use_block.""" + result = [] + for block in blocks: + if isinstance(block, dict) and "toolUse" in block: + block = MalformedJson.malform_tool_use_block(block) + elif isinstance(block, dict) and "text" in block and isinstance(block["text"], str): + block = dict(block) + block["text"] = MalformedJson.malform_text(block["text"]) + result.append(block) + return result + + +class EmptyResponse(ModelEffect): + """Returns empty content.""" + + hook: ClassVar[Literal["pre", "post"]] = "pre" + effect_type: Literal["empty_response"] = "empty_response" + + def cancel_message(self) -> str: + """Pre-cancel with single space (truthy) skips real model call.""" + return " " + + def apply(self, content: Any = None) -> Any: + if content is None: + raise ValueError("EmptyResponse.apply() requires content") + if isinstance(content, str): + return "" + elif isinstance(content, list): + return [] + raise ValueError(f"EmptyResponse.apply() received unsupported type {type(content).__name__}") + + +class Confabulation(ModelEffect): + """Injects fabricated citations into model output text.""" + + hook: ClassVar[Literal["pre", "post"]] = "post" + effect_type: Literal["confabulation"] = "confabulation" + + _CONFABULATION_TEMPLATES: ClassVar[list[str]] = [ + "According to a 2019 study published in Nature, ", + "Research conducted by MIT in 2021 demonstrated that ", + "A comprehensive report by the World Health Organization found that ", + "Based on data from the U.S. Census Bureau (2020), ", + "As documented in the IEEE Transactions on Software Engineering, ", + "A peer-reviewed analysis by Stanford researchers confirmed that ", + "The European Commission's 2022 technical report states that ", + "Historical records from the Smithsonian Institution indicate that ", + ] + + def apply(self, content: Any = None) -> Any: + if content is None: + raise ValueError("Confabulation.apply() requires content") + if isinstance(content, str): + return self._confabulate(content) + elif isinstance(content, list): + result = [] + for block in content: + block = dict(block) + if "text" in block and isinstance(block["text"], str): + block["text"] = self._confabulate(block["text"]) + result.append(block) + return result + raise ValueError(f"Confabulation.apply() received unsupported type {type(content).__name__}") + + def _confabulate(self, text: str) -> str: + if not text: + return text + template = random.choice(self._CONFABULATION_TEMPLATES) + sentences = re.split(r"(?<=[.!?])\s+", text) + if len(sentences) <= 1: + return template + text + insert_pos = random.randint(1, len(sentences) - 1) + sentences.insert(insert_pos, template) + return " ".join(sentences) + + +class FullRefusal(ModelEffect): + """Replaces model output with a refusal message.""" + + hook: ClassVar[Literal["pre", "post"]] = "pre" + effect_type: Literal["full_refusal"] = "full_refusal" + + _REFUSAL_TEMPLATES: ClassVar[list[str]] = [ + "I'm sorry, but I can't help with that request.", + "I'm not able to help with this particular request. Could you try rephrasing?", + "I can't assist with that. Please try a different approach.", + "I'm unable to process this request at this time.", + "This is beyond what I'm able to do. Please consider an alternative approach.", + "I can't assist with the request as described. Could you provide more context?", + ] + + def cancel_message(self) -> str: + """Return a random refusal template string for use with event.cancel.""" + return random.choice(self._REFUSAL_TEMPLATES) + + def apply(self, content: Any = None) -> Any: + if content is None: + raise ValueError("FullRefusal.apply() requires content") + template = random.choice(self._REFUSAL_TEMPLATES) + if isinstance(content, str): + return template + elif isinstance(content, list): + return [{"text": template}] + raise ValueError(f"FullRefusal.apply() received unsupported type {type(content).__name__}") + + +class SuccessFraming(ModelEffect): + """Prepends a confident success prefix to content. + + This is composable — applied AFTER another effect to disguise corruption. + """ + + hook: ClassVar[Literal["pre", "post"]] = "post" + effect_type: Literal["success_framing"] = "success_framing" + + _SUCCESS_PREFIXES: ClassVar[list[str]] = [ + "Successfully completed the requested operation.", + "Done! Here are the results you asked for.", + "Great news — everything worked as expected.", + "Operation finished successfully. Here's what I found:", + "All done! The task has been completed without issues.", + "I've successfully processed your request. Here's the output:", + "Task completed. Below are the verified results:", + ] + + def apply(self, content: Any = None) -> Any: + if content is None: + raise ValueError("SuccessFraming.apply() requires content") + prefix = random.choice(self._SUCCESS_PREFIXES) + if isinstance(content, str): + return prefix + " " + content + elif isinstance(content, list): + # Prepend into first text block if one exists + for block in content: + if isinstance(block, dict) and "text" in block and isinstance(block["text"], str): + block["text"] = prefix + " " + block["text"] + return content + # No text block — prepend a new one + return [{"text": prefix}] + content + raise ValueError(f"SuccessFraming.apply() received unsupported type {type(content).__name__}") + + +ModelEffectUnion = Annotated[ + Union[ + Annotated[MalformedJson, Tag("malformed_json")], + Annotated[EmptyResponse, Tag("empty_response")], + Annotated[Confabulation, Tag("confabulation")], + Annotated[FullRefusal, Tag("full_refusal")], + Annotated[SuccessFraming, Tag("success_framing")], + ], + Discriminator("effect_type"), +] diff --git a/src/strands_evals/chaos/plugin.py b/src/strands_evals/chaos/plugin.py index aa4b326f..6aab1cb9 100644 --- a/src/strands_evals/chaos/plugin.py +++ b/src/strands_evals/chaos/plugin.py @@ -1,62 +1,107 @@ """Chaos Plugin for Strands Agents. Implements chaos injection as a standard Strands Plugin using the SDK's -native hook system (BeforeToolCallEvent / AfterToolCallEvent). +native hook system. Handles BOTH tool-level and model-output chaos: -The plugin reads the active ChaosCase from a module-level ContextVar at hook -time. The ChaosExperiment manages the ContextVar lifecycle. +- BeforeToolCallEvent: cancels tool calls for pre-hook effects (Timeout, etc.) +- AfterToolCallEvent: corrupts tool responses for post-hook effects (TruncateFields, etc.) +- BeforeModelCallEvent: cancels model call for pre-hook effects (FullRefusal, EmptyResponse) +- MessageAddedEvent: corrupts model output for post-hook effects (MalformedJson, Confabulation, etc.) """ import json import logging +from enum import Enum, auto +from typing import NamedTuple, Protocol, cast -from strands.hooks import AfterToolCallEvent, BeforeToolCallEvent +from strands.hooks import ( + AfterToolCallEvent, + BeforeModelCallEvent, + BeforeToolCallEvent, + MessageAddedEvent, +) from strands.plugins import Plugin, hook from ._context import _current_chaos_case -from .effects import ChaosEffect, TruncateFields +from .effects import ( + ChaosEffect, + MalformedJson, + SuccessFraming, + TruncateFields, +) logger = logging.getLogger(__name__) +class MessageKind(Enum): + """Kind of corruptible model output.""" + + STRUCTURED_OUTPUT = auto() + FINAL_TEXT = auto() + + +class ModelOutputTarget(NamedTuple): + """A model output eligible for corruption.""" + + kind: MessageKind + content: list + structured_output_tool_names: set[str] + + +class PreModelEffect(Protocol): + """A model effect that cancels the model call with a message.""" + + def cancel_message(self) -> str: ... + + class ChaosPlugin(Plugin): - """Strands Plugin that injects deterministic chaos based on the active ChaosCase. + """Strands Plugin that injects deterministic chaos based on configuration. + + Handles both tool-level chaos and model-output chaos: - The plugin intercepts tool calls via Strands' native hook system: - - BeforeToolCallEvent: cancels tool calls for pre-hook effects (Timeout, NetworkError, etc.) - - AfterToolCallEvent: corrupts tool responses for post-hook effects (TruncateFields, etc.) + Tool chaos: + - BeforeToolCallEvent: cancels tool calls for pre-hook effects + - AfterToolCallEvent: corrupts tool responses for post-hook effects + + Model output chaos: + - BeforeModelCallEvent: cancels model call for pre-hook effects (FullRefusal, EmptyResponse) + - MessageAddedEvent: corrupts the final assistant response content (post effects) The active ChaosCase is managed via a ContextVar (set by ChaosExperiment). - When no ChaosCase is active or the case has no effects, all tools behave normally. + When no ChaosCase is active or the case has no model_effects, all hooks + pass through without modification. + + Model output effects are configured via `model_effects` on the ChaosCase. + Effects are applied sequentially. SuccessFraming is always applied LAST + (composable post-step). MalformedJson can reach structured-output toolUse + blocks; other post effects skip toolUse messages. Example:: from strands import Agent - from strands_evals.chaos import ChaosPlugin + from strands_evals.chaos import ChaosCase, ChaosPlugin + from strands_evals.chaos.effects import FullRefusal, EmptyResponse - chaos = ChaosPlugin() - agent = Agent( - model=my_model, - tools=[search_tool, database_tool], - plugins=[chaos], + chaos_case = ChaosCase( + name="refusal_test", + input="Tell me about quantum physics", + effects={ + "model_effects": {"*": [FullRefusal()]}, + }, ) - - # The ChaosExperiment handles ChaosCase activation via ContextVar. - # The user's task body contains zero chaos concepts. + chaos = ChaosPlugin() + agent = Agent(model=my_model, tools=[...], plugins=[chaos]) """ name = "chaos-testing" - def __init__(self) -> None: - super().__init__() + # Tool chaos hooks @hook # type: ignore[call-overload] def before_tool_call(self, event: BeforeToolCallEvent) -> None: """Intercept tool calls to inject pre-hook (error) effects. - For pre-hook effects (Timeout, NetworkError, ExecutionError, - ValidationError), cancels the tool call with the effect's error_message - before the tool executes. + Cancels the tool call with the effect's error_message before execution. """ chaos_case = _current_chaos_case.get() if chaos_case is None or not chaos_case.tool_effects: @@ -78,8 +123,7 @@ def before_tool_call(self, event: BeforeToolCallEvent) -> None: def after_tool_call(self, event: AfterToolCallEvent) -> None: """Intercept tool results to inject post-hook (corruption) effects. - For corruption effects (TruncateFields, RemoveFields, CorruptValues), - applies effect.apply() to JSON content blocks in the tool response. + Applies corruption effects to JSON content blocks in the tool response. """ chaos_case = _current_chaos_case.get() if chaos_case is None or not chaos_case.tool_effects: @@ -102,12 +146,145 @@ def after_tool_call(self, event: AfterToolCallEvent) -> None: content = result.get("content") if isinstance(content, list): - result["content"] = self._apply_to_blocks(effect, content) # type: ignore[assignment] + result["content"] = self._apply_to_tool_blocks(effect, content) # type: ignore[assignment] logger.info("effect=<%s>, tool=<%s> | applied chaos post-hook", type(effect).__name__, tool_name) - def _apply_to_blocks(self, effect: ChaosEffect, blocks: list) -> list: - """Apply effect to text blocks in a content list.""" + # Model output chaos hooks + + @hook # type: ignore[call-overload] + def before_model_invocation(self, event: BeforeModelCallEvent) -> None: + """Cancel the model call when a pre-hook model effect is configured.""" + effect = self._select_pre_model_effect() + if effect is None: + return + event.cancel = effect.cancel_message() + + @hook # type: ignore[call-overload] + def after_model_invocation(self, event: MessageAddedEvent) -> None: + """Corrupt eligible model output with the configured post-hook model effects.""" + effects = self._get_post_model_effects() + target = self._classify_model_output(event) + if target is None or not effects: + return + event.message["content"] = self._apply_model_effects(effects, target) + + def _select_pre_model_effect(self) -> PreModelEffect | None: + """Return the single configured pre-hook model effect, or None. + + ChaosCase validation guarantees at most one pre effect, so no ordering policy is needed. + """ + chaos_case = _current_chaos_case.get() + if chaos_case is None: + return None + for effect in chaos_case.model_effects: + if effect.hook == "pre": + return cast(PreModelEffect, effect) + return None + + def _get_post_model_effects(self) -> list: + """Return the configured post-hook model effects. + + Empty when a pre effect is configured: the pre effect already produced the turn, + so applying post effects would corrupt it twice. + """ + chaos_case = _current_chaos_case.get() + if chaos_case is None: + return [] + if any(e.hook == "pre" for e in chaos_case.model_effects): + return [] + return [e for e in chaos_case.model_effects if e.hook == "post"] + + def _classify_model_output(self, event: MessageAddedEvent) -> ModelOutputTarget | None: + """Return the corruptible target for this message, or None if it must be left alone. + + Ordinary tool dispatch is excluded: MessageAddedEvent fires before dispatch, so + corrupting those blocks breaks the agent loop. + """ + message = event.message + if message.get("role") != "assistant": + return None + content = message.get("content") + if content is None: + return None + if not isinstance(content, list): + return ModelOutputTarget(MessageKind.FINAL_TEXT, content, set()) + + if not any(isinstance(block, dict) and "toolUse" in block for block in content): + return ModelOutputTarget(MessageKind.FINAL_TEXT, content, set()) + + structured_output_tool_names = self._get_structured_output_tool_names(event.agent) + targets_structured_output = any( + isinstance(block, dict) + and "toolUse" in block + and block["toolUse"].get("name", "") in structured_output_tool_names + for block in content + ) + if targets_structured_output: + return ModelOutputTarget(MessageKind.STRUCTURED_OUTPUT, content, structured_output_tool_names) + return None + + def _get_structured_output_tool_names(self, agent) -> set[str]: # type: ignore[type-arg] + """Identify structured-output tools via isinstance(tool, StructuredOutputTool).""" + from strands.tools.structured_output.structured_output_tool import StructuredOutputTool + + return { + name for name, tool in agent.tool_registry.dynamic_tools.items() if isinstance(tool, StructuredOutputTool) + } + + def _apply_model_effects(self, effects: list, target: ModelOutputTarget) -> list: + """Corrupt the target content with the given post-hook model effects. + + Structured-output toolUse is reachable only by MalformedJson; any other effect + would break the structured-output contract, so it is skipped for that target. + """ + if target.kind is MessageKind.STRUCTURED_OUTPUT: + effects = [e for e in effects if isinstance(e, MalformedJson)] + if not effects: + return target.content + return self._apply_to_model_blocks(effects, target.content, target.structured_output_tool_names) + + def _apply_to_model_blocks( + self, post_effects: list, content: list, structured_output_tool_names: set[str] | None = None + ) -> list: + """Apply model post effects to content blocks sequentially. + + SuccessFraming runs last so it frames whatever the other effects produced. + """ + primary = [e for e in post_effects if not isinstance(e, SuccessFraming)] + framing = [e for e in post_effects if isinstance(e, SuccessFraming)] + + corrupted = content + for effect in primary: + if isinstance(effect, MalformedJson) and structured_output_tool_names: + corrupted = self._apply_malformed_json_selective(effect, corrupted, structured_output_tool_names) + else: + corrupted = effect.apply(corrupted) + for effect in framing: + corrupted = effect.apply(corrupted) + return corrupted + + def _apply_malformed_json_selective( + self, effect: MalformedJson, blocks: list, structured_output_tool_names: set[str] + ) -> list: + """Apply MalformedJson: text blocks get malformed; only SO toolUse blocks get corrupted.""" + result = [] + for block in blocks: + if isinstance(block, dict) and "toolUse" in block: + tool_name = block["toolUse"].get("name", "") + if tool_name in structured_output_tool_names: + block = MalformedJson.malform_tool_use_block(block) + # else: ordinary toolUse — leave untouched + elif isinstance(block, dict) and "text" in block and isinstance(block["text"], str): + block = dict(block) + block["text"] = MalformedJson.malform_text(block["text"]) + result.append(block) + return result + + # Tool corruption helpers + + def _apply_to_tool_blocks(self, effect: ChaosEffect, blocks: list) -> list: + """Apply effect to text blocks in a tool content list.""" corrupted_blocks = [] for block in blocks: if isinstance(block, dict) and "text" in block: diff --git a/tests/strands_evals/chaos/test_case.py b/tests/strands_evals/chaos/test_case.py index 644112c8..e90d494f 100644 --- a/tests/strands_evals/chaos/test_case.py +++ b/tests/strands_evals/chaos/test_case.py @@ -53,8 +53,8 @@ def test_case_with_multiple_effects_per_tool(self): ) def test_unknown_effect_category_raises(self): - """Unknown effect category keys should be rejected.""" - with pytest.raises(ValueError, match="Unknown effect categories"): + """Unknown effect category keys should be rejected by the ChaosEffects schema.""" + with pytest.raises(ValueError, match="extra_forbidden"): ChaosCase( name="bad_category", input="hello", diff --git a/tests/strands_evals/chaos/test_model_chaos.py b/tests/strands_evals/chaos/test_model_chaos.py new file mode 100644 index 00000000..a7137c3c --- /dev/null +++ b/tests/strands_evals/chaos/test_model_chaos.py @@ -0,0 +1,617 @@ +"""Unit tests for model output chaos via ChaosPlugin two-hook architecture. + +Tests cover: +- Effects constructed via keyed dict {"model_effects": {"*": [...]}} +- EmptyResponse as pre-hook: model not called, turn is single space +- FullRefusal as pre-hook: model not called, turn is refusal text +- MalformedJson on structured-output toolUse: toolUse input corrupted +- Post effects (Confabulation, MalformedJson-on-text, SuccessFraming) work +- Mixed pre+post still produces one turn (pre wins) +- MalformedJson DOES reach/corrupt structured-output toolUse +- Effect family validation: wrong-category effects rejected +- Wildcard rejection: non-'*' model_effects keys rejected +- Ordinary dynamic tool not corrupted: isinstance-based detection +""" + +import copy +from unittest.mock import MagicMock + +import pytest +from pydantic import ValidationError as PydanticValidationError +from strands.hooks import BeforeModelCallEvent +from strands.tools.structured_output.structured_output_tool import StructuredOutputTool + +from strands_evals.chaos._context import _current_chaos_case +from strands_evals.chaos.case import ChaosCase +from strands_evals.chaos.effects import ( + Confabulation, + EmptyResponse, + FullRefusal, + MalformedJson, + SuccessFraming, + Timeout, +) +from strands_evals.chaos.plugin import ChaosPlugin + + +def _make_event(message: dict, dynamic_tools: dict | None = None) -> MagicMock: + """Create a mock MessageAddedEvent with the given message. + + Args: + message: The message dict. + dynamic_tools: Optional dict of dynamic tool names -> tools (structured-output tools). + If None, defaults to empty dict (no structured-output tools registered). + """ + event = MagicMock() + event.message = message + event.agent.tool_registry.dynamic_tools = dynamic_tools or {} + return event + + +def _final_assistant_message(text: str = "The answer is 42.") -> dict: + """An end_turn assistant message with text content only (no toolUse).""" + return { + "role": "assistant", + "content": [{"text": text}], + } + + +def _tooluse_assistant_message() -> dict: + """A tool_use assistant message containing a toolUse block.""" + return { + "role": "assistant", + "content": [ + {"text": "Let me search for that."}, + {"toolUse": {"toolUseId": "tu_1", "name": "search", "input": {"query": "test"}}}, + ], + } + + +def _user_message() -> dict: + """A user message.""" + return { + "role": "user", + "content": [{"text": "Hello, what is 2+2?"}], + } + + +def _tool_result_message() -> dict: + """A tool result message.""" + return { + "role": "user", + "content": [{"toolResult": {"toolUseId": "tu_1", "status": "success", "content": [{"text": "4"}]}}], + } + + +def _set_chaos_case(model_effects): + """Helper to set the _current_chaos_case ContextVar with given model_effects. + + Uses keyed dict form: effects={"model_effects": {"*": model_effects}} + """ + case = ChaosCase( + name="test_case", + input="test input", + effects={"model_effects": {"*": model_effects}}, + ) + _current_chaos_case.set(case) + return case + + +class TestKeyedDictConstruction: + """Effects are constructed via keyed dict form.""" + + def test_keyed_dict_form_is_valid(self): + """ChaosCase accepts effects={"model_effects": {"*": [...]}}.""" + case = ChaosCase( + name="keyed", + input="test", + effects={"model_effects": {"*": [MalformedJson()]}}, + ) + assert case.model_effects == [MalformedJson()] + + def test_wildcard_resolver(self): + """model_effects property resolves '*' wildcard to flat list.""" + case = ChaosCase( + name="wildcard", + input="test", + effects={"model_effects": {"*": [FullRefusal(), MalformedJson()]}}, + ) + assert len(case.model_effects) == 2 + assert isinstance(case.model_effects[0], FullRefusal) + assert isinstance(case.model_effects[1], MalformedJson) + + def test_empty_effects_baseline(self): + """Empty effects dict produces no model_effects.""" + case = ChaosCase(name="baseline", input="test", effects={}) + assert case.model_effects == [] + + +class TestEmptyResponsePreHook: + """EmptyResponse is a pre-hook effect — cancels model call with single space.""" + + def test_empty_response_cancels_with_single_space(self): + """before_model_invocation sets event.cancel to ' ' (single space).""" + _set_chaos_case([EmptyResponse()]) + plugin = ChaosPlugin() + event = BeforeModelCallEvent(agent=MagicMock()) + + plugin.before_model_invocation(event) + + assert event.cancel == " " + + def test_empty_response_model_not_called(self): + """When EmptyResponse fires as pre-hook, post-hook does not apply effects.""" + _set_chaos_case([EmptyResponse()]) + plugin = ChaosPlugin() + + # Pre-hook fires + pre_event = BeforeModelCallEvent(agent=MagicMock()) + plugin.before_model_invocation(pre_event) + assert pre_event.cancel == " " + + # SDK builds cancel message, MessageAddedEvent fires + cancel_message = {"role": "assistant", "content": [{"text": " "}]} + post_event = _make_event(cancel_message) + plugin.after_model_invocation(post_event) + + # Content should be unchanged (pre effects skip post processing) + assert cancel_message["content"] == [{"text": " "}] + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestFullRefusalPreHook: + """FullRefusal is a pre-hook effect — cancels model call with refusal text.""" + + def test_full_refusal_cancels_model_call(self): + """before_model_invocation sets event.cancel to a refusal template.""" + _set_chaos_case([FullRefusal()]) + plugin = ChaosPlugin() + event = BeforeModelCallEvent(agent=MagicMock()) + + plugin.before_model_invocation(event) + + assert event.cancel in FullRefusal._REFUSAL_TEMPLATES + + def test_full_refusal_produces_single_turn(self): + """FullRefusal cancels model call, SDK builds cancel message, run ends.""" + _set_chaos_case([FullRefusal()]) + plugin = ChaosPlugin() + + # Step 1: before_model_invocation fires + pre_event = BeforeModelCallEvent(agent=MagicMock()) + plugin.before_model_invocation(pre_event) + cancel_text = pre_event.cancel + assert cancel_text in FullRefusal._REFUSAL_TEMPLATES + + # Step 2: SDK builds the cancel message and fires MessageAddedEvent + cancel_message = {"role": "assistant", "content": [{"text": cancel_text}]} + post_event = _make_event(cancel_message) + plugin.after_model_invocation(post_event) + + # Step 3: verify the cancel message is unchanged (not double-corrupted) + assert cancel_message["content"] == [{"text": cancel_text}] + assert len(cancel_message["content"]) == 1 + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestMalformedJsonStructuredOutput: + """MalformedJson DOES reach and corrupt structured-output toolUse blocks only.""" + + def test_malformed_json_corrupts_structured_output_tooluse(self): + """MalformedJson corrupts toolUse input when tool is a StructuredOutputTool.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = { + "role": "assistant", + "content": [ + {"toolUse": {"toolUseId": "so_1", "name": "MyModel", "input": {"field1": "value1"}}}, + ], + } + mock_so_tool = MagicMock(spec=StructuredOutputTool) + event = _make_event(message, dynamic_tools={"MyModel": mock_so_tool}) + + plugin.after_model_invocation(event) + + tool_use_block = message["content"][0]["toolUse"] + corrupted_input = tool_use_block["input"] + assert isinstance(corrupted_input, str) + assert not corrupted_input.endswith("}") + + def test_plain_tooluse_not_corrupted_even_with_malformed_json(self): + """A plain mid-turn toolUse (not in dynamic_tools) is NOT corrupted by MalformedJson.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = { + "role": "assistant", + "content": [ + {"toolUse": {"toolUseId": "tu_1", "name": "search", "input": {"query": "test"}}}, + ], + } + original_content = copy.deepcopy(message["content"]) + event = _make_event(message, dynamic_tools={}) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def test_mixed_tooluse_only_structured_output_corrupted(self): + """In a message with both regular and structured-output toolUse, only SO is corrupted.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = { + "role": "assistant", + "content": [ + {"toolUse": {"toolUseId": "tu_1", "name": "search", "input": {"query": "test"}}}, + {"toolUse": {"toolUseId": "so_1", "name": "MyModel", "input": {"field1": "value1"}}}, + ], + } + mock_so_tool = MagicMock(spec=StructuredOutputTool) + event = _make_event(message, dynamic_tools={"MyModel": mock_so_tool}) + + plugin.after_model_invocation(event) + + # "search" toolUse should be UNCHANGED + search_block = message["content"][0]["toolUse"] + assert search_block["input"] == {"query": "test"} + # "MyModel" toolUse should be CORRUPTED + so_block = message["content"][1]["toolUse"] + assert isinstance(so_block["input"], str) + assert not so_block["input"].endswith("}") + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestPostEffectsOnText: + """Post effects (Confabulation, MalformedJson-on-text, SuccessFraming) work.""" + + def test_confabulation_injects_template(self): + """Confabulation injects fabricated citations into text content.""" + _set_chaos_case([Confabulation()]) + plugin = ChaosPlugin() + original_text = "The weather is sunny. It is warm outside. Birds are singing." + message = _final_assistant_message(original_text) + event = _make_event(message) + + plugin.after_model_invocation(event) + + result_text = message["content"][0]["text"] + assert result_text != original_text + assert "sunny" in result_text or "warm" in result_text + + def test_malformed_json_on_text(self): + """MalformedJson truncates JSON-like text content.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = _final_assistant_message('{"key": "value", "nested": {"a": 1}}') + event = _make_event(message) + + plugin.after_model_invocation(event) + + result_text = message["content"][0]["text"] + assert result_text != '{"key": "value", "nested": {"a": 1}}' + assert len(result_text) < len('{"key": "value", "nested": {"a": 1}}') + + def test_success_framing_prepends_prefix(self): + """SuccessFraming prepends a confident prefix to text content.""" + _set_chaos_case([SuccessFraming()]) + plugin = ChaosPlugin() + message = _final_assistant_message("Here is the result.") + event = _make_event(message) + + plugin.after_model_invocation(event) + + result_text = message["content"][0]["text"] + has_prefix = any(result_text.startswith(p) for p in SuccessFraming._SUCCESS_PREFIXES) + assert has_prefix + assert "Here is the result." in result_text + + def test_confabulation_plus_success_framing(self): + """Confabulation + SuccessFraming compose: citation injected, then prefix prepended.""" + _set_chaos_case([Confabulation(), SuccessFraming()]) + plugin = ChaosPlugin() + original_text = "The weather is sunny. It is warm outside. Birds are singing." + message = _final_assistant_message(original_text) + event = _make_event(message) + + plugin.after_model_invocation(event) + + result_text = message["content"][0]["text"] + has_prefix = any(result_text.startswith(p) for p in SuccessFraming._SUCCESS_PREFIXES) + assert has_prefix + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestMixedPrePostCase: + """Mixed pre+post effects: pre wins, post does NOT double-corrupt.""" + + def test_full_refusal_plus_malformed_json(self): + """FullRefusal (pre) + MalformedJson (post): pre cancels, post skipped.""" + _set_chaos_case([FullRefusal(), MalformedJson()]) + plugin = ChaosPlugin() + + pre_event = BeforeModelCallEvent(agent=MagicMock()) + plugin.before_model_invocation(pre_event) + cancel_text = pre_event.cancel + assert cancel_text in FullRefusal._REFUSAL_TEMPLATES + + cancel_message = {"role": "assistant", "content": [{"text": cancel_text}]} + post_event = _make_event(cancel_message) + plugin.after_model_invocation(post_event) + + # Post effect (MalformedJson) should NOT have corrupted the content + assert cancel_message["content"] == [{"text": cancel_text}] + assert len(cancel_message["content"]) == 1 + + def test_empty_response_plus_success_framing(self): + """EmptyResponse (pre) + SuccessFraming (post): pre cancels, post skipped.""" + _set_chaos_case([EmptyResponse(), SuccessFraming()]) + plugin = ChaosPlugin() + + pre_event = BeforeModelCallEvent(agent=MagicMock()) + plugin.before_model_invocation(pre_event) + assert pre_event.cancel == " " + + cancel_message = {"role": "assistant", "content": [{"text": " "}]} + post_event = _make_event(cancel_message) + plugin.after_model_invocation(post_event) + + # SuccessFraming (post) should NOT have been applied + assert cancel_message["content"] == [{"text": " "}] + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestMalformedJsonReachesStructuredOutput: + """MalformedJson reaches structured-output toolUse (relaxed for it).""" + + def test_malformed_json_corrupts_structured_output_tooluse(self): + """MalformedJson DOES corrupt a structured-output toolUse block.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = { + "role": "assistant", + "content": [ + { + "toolUse": { + "toolUseId": "so_1", + "name": "MyModel", + "input": {"field1": "value1"}, + } + }, + ], + } + mock_so_tool = MagicMock(spec=StructuredOutputTool) + event = _make_event(message, dynamic_tools={"MyModel": mock_so_tool}) + + plugin.after_model_invocation(event) + + tool_use_block = message["content"][0]["toolUse"] + assert isinstance(tool_use_block["input"], str) + assert not tool_use_block["input"].endswith("}") + + def test_other_post_effects_still_skip_tooluse(self): + """Confabulation on a toolUse message is skipped (only relaxed for MalformedJson).""" + _set_chaos_case([Confabulation()]) + plugin = ChaosPlugin() + message = _tooluse_assistant_message() + original_content = copy.deepcopy(message["content"]) + event = _make_event(message) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestEffectFamilyValidation: + """Effects placed in the wrong category are rejected structurally by Pydantic.""" + + def test_tool_effect_in_model_effects_rejected(self): + """A ToolEffect under model_effects is rejected by discriminated union.""" + with pytest.raises(PydanticValidationError, match="union_tag_invalid"): + ChaosCase( + name="bad", + input="test", + effects={"model_effects": {"*": [Timeout()]}}, + ) + + def test_model_effect_in_tool_effects_rejected(self): + """A ModelEffect under tool_effects is rejected by discriminated union.""" + with pytest.raises(PydanticValidationError, match="union_tag_invalid"): + ChaosCase( + name="bad", + input="test", + effects={"tool_effects": {"search": [FullRefusal()]}}, + ) + + def test_model_effect_in_tool_effects_rejected_via_model_validate(self): + """A ModelEffect under tool_effects is rejected on the model_validate (dict) path.""" + with pytest.raises(PydanticValidationError, match="union_tag_invalid"): + ChaosCase.model_validate( + { + "name": "bad_tool", + "input": "test", + "effects": {"tool_effects": {"search": [{"effect_type": "full_refusal"}]}}, + } + ) + + def test_tool_effect_in_model_effects_rejected_via_model_validate(self): + """A ToolEffect under model_effects is rejected on the model_validate (dict) path.""" + with pytest.raises(PydanticValidationError, match="union_tag_invalid"): + ChaosCase.model_validate( + { + "name": "bad_model", + "input": "test", + "effects": {"model_effects": {"*": [{"effect_type": "timeout"}]}}, + } + ) + + def test_named_model_key_rejected(self): + """A non-'*' key in model_effects is rejected by Literal constraint.""" + with pytest.raises(PydanticValidationError, match="literal_error"): + ChaosCase( + name="bad", + input="test", + effects={"model_effects": {"claude-sonnet": [MalformedJson()]}}, + ) + + def test_bogus_category_rejected(self): + """An unknown effects category is rejected by extra='forbid'.""" + with pytest.raises(PydanticValidationError, match="extra_forbidden"): + ChaosCase( + name="bad", + input="test", + effects={"bogus": {"x": []}}, + ) + + +class TestSinglePreModelEffect: + """At most one pre-hook model effect per case — pre effects cancel the model call.""" + + def test_two_pre_effects_rejected(self): + """FullRefusal + EmptyResponse (both pre) is rejected, naming both effects.""" + with pytest.raises(PydanticValidationError, match="only 1 is allowed"): + ChaosCase( + name="two_pre", + input="test", + effects={"model_effects": {"*": [FullRefusal(), EmptyResponse()]}}, + ) + + def test_two_pre_effects_rejected_via_model_validate(self): + """Two pre effects are rejected on the model_validate (dict) path.""" + with pytest.raises(PydanticValidationError, match="only 1 is allowed"): + ChaosCase.model_validate( + { + "name": "two_pre", + "input": "test", + "effects": { + "model_effects": {"*": [{"effect_type": "full_refusal"}, {"effect_type": "empty_response"}]} + }, + } + ) + + def test_rejection_names_both_effects(self): + """The error message identifies both offending pre effects.""" + with pytest.raises(PydanticValidationError) as exc_info: + ChaosCase( + name="two_pre", + input="test", + effects={"model_effects": {"*": [FullRefusal(), EmptyResponse()]}}, + ) + message = str(exc_info.value) + assert "FullRefusal" in message + assert "EmptyResponse" in message + + def test_single_pre_effect_accepted(self): + """One pre effect alone is valid.""" + case = ChaosCase( + name="one_pre", + input="test", + effects={"model_effects": {"*": [FullRefusal()]}}, + ) + assert len(case.model_effects) == 1 + + def test_pre_plus_post_mix_accepted(self): + """A pre + post mix is valid — only multiple pre effects are rejected.""" + case = ChaosCase( + name="mixed", + input="test", + effects={"model_effects": {"*": [FullRefusal(), MalformedJson()]}}, + ) + assert len(case.model_effects) == 2 + + +class TestOrdinaryDynamicToolNotCorrupted: + """An ordinary dynamic tool (not StructuredOutputTool) is NOT corrupted.""" + + def test_ordinary_dynamic_tool_unchanged(self): + """MalformedJson does NOT corrupt a regular dynamic tool's toolUse.""" + _set_chaos_case([MalformedJson()]) + plugin = ChaosPlugin() + message = { + "role": "assistant", + "content": [ + {"toolUse": {"toolUseId": "dt_1", "name": "my_dynamic_tool", "input": {"key": "val"}}}, + ], + } + original_content = copy.deepcopy(message["content"]) + # Register as a plain MagicMock (NOT spec'd to StructuredOutputTool) + mock_tool = MagicMock() + event = _make_event(message, dynamic_tools={"my_dynamic_tool": mock_tool}) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestGuardRoleFiltering: + """User and tool result messages are NOT corrupted.""" + + def test_user_message_not_corrupted(self): + _set_chaos_case([Confabulation()]) + plugin = ChaosPlugin() + message = _user_message() + original_content = copy.deepcopy(message["content"]) + event = _make_event(message) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def test_tool_result_message_not_corrupted(self): + _set_chaos_case([Confabulation()]) + plugin = ChaosPlugin() + message = _tool_result_message() + original_content = copy.deepcopy(message["content"]) + event = _make_event(message) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def teardown_method(self): + _current_chaos_case.set(None) + + +class TestPassthrough: + """No corruption when no model_effects is set.""" + + def test_no_config_passes_through(self): + _current_chaos_case.set(None) + plugin = ChaosPlugin() + message = _final_assistant_message("Hello world") + original_content = copy.deepcopy(message["content"]) + event = _make_event(message) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def test_empty_effects_passes_through(self): + """ChaosCase with empty effects dict does not corrupt.""" + case = ChaosCase(name="baseline", input="test", effects={}) + _current_chaos_case.set(case) + plugin = ChaosPlugin() + message = _final_assistant_message("Hello world") + original_content = copy.deepcopy(message["content"]) + event = _make_event(message) + + plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def teardown_method(self): + _current_chaos_case.set(None)