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..dc7a93d1 100644 --- a/src/strands_evals/chaos/__init__.py +++ b/src/strands_evals/chaos/__init__.py @@ -4,13 +4,20 @@ under tool failures and response corruption scenarios. """ -from .case import ChaosCase +from .case import ChaosCase, ChaosEffectsConfig from .effects import ( ChaosEffect, + Confabulation, CorruptValues, + EmptyResponse, ExecutionError, + FullRefusal, + MalformedJson, + ModelEffect, + ModelEffectUnion, NetworkError, RemoveFields, + SuccessFraming, Timeout, ToolEffect, ToolEffectUnion, @@ -23,19 +30,27 @@ __all__ = [ # Core classes "ChaosCase", + "ChaosEffectsConfig", "ChaosExperiment", "ChaosPlugin", # Effect hierarchy "ChaosEffect", "ToolEffect", "ToolEffectUnion", - # Pre-hook effects (tool call failures) + "ModelEffect", + "ModelEffectUnion", + # 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..d3c70058 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 -from pydantic import Field, model_validator -from typing_extensions import Generic +from pydantic import ConfigDict, Field, model_validator +from typing_extensions import Generic, TypedDict from ..case import Case from ..types.evaluation import InputT, OutputT -from .effects import ToolEffectUnion +from .effects import ModelEffectUnion, ToolEffectUnion + + +class ChaosEffectsConfig(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: ChaosEffectsConfig = Field( + default_factory=ChaosEffectsConfig, + 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, ChaosEffectsConfig], 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, ChaosEffectsConfig]] = [] 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..7c8e1675 100644 --- a/src/strands_evals/chaos/effects.py +++ b/src/strands_evals/chaos/effects.py @@ -13,6 +13,7 @@ import math import random +import re from abc import abstractmethod from typing import Annotated, Any, ClassVar, Literal, Union @@ -43,11 +44,6 @@ class ToolEffect(ChaosEffect): """ -# --------------------------------------------------------------------------- -# Pre-hook effects: cancel the tool call before execution -# --------------------------------------------------------------------------- - - class Timeout(ToolEffect): """Simulates a tool call timeout. @@ -152,11 +148,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 +299,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 +316,181 @@ 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. + + On final text responses this truncates JSON-like content. For structured output the + plugin instead injects a single parse failure per agent invocation at the tool + boundary, so the agent must recover; the corrected attempt is left untouched and a + typed caller still receives validated structured 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.""" + stripped = text.strip() + if stripped.startswith("{") or stripped.startswith("["): + return stripped[: len(stripped) // 2] + return text + + @staticmethod + def _malform_blocks(blocks: list) -> list: + """Corrupt JSON-like text in each content block.""" + result = [] + for block in blocks: + if 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 apply(self, content: Any = None) -> str: + """Return the cancel text that skips the real model call. + + A single space rather than an empty string: the SDK only honours a truthy + `event.cancel`, so `""` would let the real call proceed. + """ + return " " + + +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 apply(self, content: Any = None) -> str: + """Return a random refusal template for use as `event.cancel`.""" + return random.choice(self._REFUSAL_TEMPLATES) + + +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..19e450ae 100644 --- a/src/strands_evals/chaos/plugin.py +++ b/src/strands_evals/chaos/plugin.py @@ -1,65 +1,106 @@ """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.), and + injects one structured-output parse failure per invocation for MalformedJson +- AfterToolCallEvent: corrupts tool responses for post-hook effects (TruncateFields, etc.) +- BeforeModelCallEvent: cancels model call for pre-hook effects (FullRefusal, EmptyResponse) +- MessageAddedEvent: corrupts final text model output for post-hook effects + (Confabulation, MalformedJson, SuccessFraming) + +MalformedJson does not corrupt structured-output payloads in the message history. +Instead it injects a single structured-output parse failure per agent invocation via +BeforeToolCallEvent.cancel_tool, which tests whether the agent recovers; the SDK's +corrected attempt passes through unchanged, so a typed caller still receives validated +structured output. after_model_invocation never touches messages carrying toolUse blocks. """ import json import logging -from strands.hooks import AfterToolCallEvent, BeforeToolCallEvent +from strands.hooks import ( + AfterInvocationEvent, + AfterToolCallEvent, + BeforeModelCallEvent, + BeforeToolCallEvent, + MessageAddedEvent, +) from strands.plugins import Plugin, hook from ._context import _current_chaos_case -from .effects import ChaosEffect, TruncateFields +from .case import ChaosCase +from .effects import ( + ChaosEffect, + MalformedJson, + ModelEffectUnion, + SuccessFraming, + TruncateFields, +) logger = logging.getLogger(__name__) +_CHAOS_STATE_KEY = "strands_evals.chaos" +_MALFORMED_OUTPUT_APPLIED = "malformed_structured_output_applied" + 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: + + Tool chaos: + - BeforeToolCallEvent: cancels tool calls for pre-hook effects + - AfterToolCallEvent: corrupts tool responses for post-hook effects - 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.) + 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. + Post effects apply to final text responses only and are applied sequentially, with + SuccessFraming always LAST (composable post-step). Messages carrying toolUse blocks + are never corrupted; MalformedJson instead injects one structured-output parse + failure per invocation at the tool boundary. 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__() - @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: + if chaos_case is None: + return + + if self._inject_structured_output_failure(event, chaos_case): + return + + if not chaos_case.tool_effects: return tool_name = event.tool_use.get("name", "") @@ -74,12 +115,38 @@ def before_tool_call(self, event: BeforeToolCallEvent) -> None: logger.info("effect=<%s>, tool=<%s> | injected chaos pre-hook", type(effect).__name__, tool_name) return + def _inject_structured_output_failure(self, event: BeforeToolCallEvent, chaos_case: ChaosCase) -> bool: + """Fail the first structured-output attempt so the agent must recover. + + Cancelling the tool produces an error toolResult, which drives the SDK's + structured-output correction loop. The invocation_state marker makes this fire + exactly once per agent invocation, so the corrected attempt passes through and + the caller still receives validated structured output. + """ + effect = next((e for e in chaos_case.model_effects if isinstance(e, MalformedJson)), None) + if effect is None: + return False + if event.selected_tool is None: + return False + if event.selected_tool.tool_type != "structured_output": + return False + + state = event.invocation_state.setdefault(_CHAOS_STATE_KEY, {}) + if state.get(_MALFORMED_OUTPUT_APPLIED): + return False + state[_MALFORMED_OUTPUT_APPLIED] = True + + event.cancel_tool = ( + "Structured output was malformed and could not be parsed. Please produce a corrected response." + ) + logger.info("effect=<%s> | injected structured output parse failure", type(effect).__name__) + return True + @hook # type: ignore[call-overload] 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 +169,108 @@ 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.""" + @hook # type: ignore[call-overload] + def after_invocation(self, event: AfterInvocationEvent) -> None: + """Clear per-invocation chaos state. + + The SDK uses the caller-supplied invocation_state by reference and does not strip + plugin keys when the invocation ends, so a reused dict would suppress injection on + every subsequent invocation. + """ + chaos_state = event.invocation_state.get(_CHAOS_STATE_KEY) + if not isinstance(chaos_state, dict): + return + chaos_state.pop(_MALFORMED_OUTPUT_APPLIED, None) + if not chaos_state: + event.invocation_state.pop(_CHAOS_STATE_KEY, None) + + @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.apply() + logger.info("effect=<%s> | injected model pre-hook cancel", type(effect).__name__) + + @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() + if not effects: + return + content = self._classify_model_output(event) + if content is None: + return + event.message["content"] = self._apply_to_model_blocks(effects, content) + logger.info( + "effects=<%s> | applied model output chaos", + ", ".join(type(e).__name__ for e in effects), + ) + + def _select_pre_model_effect(self) -> ModelEffectUnion | 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 effect + return None + + def _get_post_model_effects(self) -> list[ModelEffectUnion]: + """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) -> list | None: + """Return the final-text content eligible for corruption, or None to leave the message alone. + + Any message carrying a toolUse block is left alone: MessageAddedEvent fires before + dispatch, and structured-output failures are injected at the tool boundary instead. + """ + message = event.message + if message.get("role") != "assistant": + return None + content = message.get("content") + if content is None: + return None + if isinstance(content, list) and any(isinstance(b, dict) and "toolUse" in b for b in content): + return None + return content + + def _apply_to_model_blocks(self, post_effects: list, content: list) -> 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: + corrupted = effect.apply(corrupted) + for effect in framing: + corrupted = effect.apply(corrupted) + return corrupted + + 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..8b85d244 100644 --- a/tests/strands_evals/chaos/test_case.py +++ b/tests/strands_evals/chaos/test_case.py @@ -1,10 +1,18 @@ """Unit tests for ChaosCase.""" import pytest +from pydantic import ValidationError as PydanticValidationError from strands_evals import Case from strands_evals.chaos import ChaosCase -from strands_evals.chaos.effects import CorruptValues, Timeout, TruncateFields +from strands_evals.chaos.effects import ( + CorruptValues, + EmptyResponse, + FullRefusal, + MalformedJson, + Timeout, + TruncateFields, +) class TestChaosCase: @@ -53,8 +61,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 ChaosEffectsConfig schema.""" + with pytest.raises(ValueError, match="extra_forbidden"): ChaosCase( name="bad_category", input="hello", @@ -116,6 +124,153 @@ def test_model_dump_roundtrip(self): assert restored.tool_effects["tool_b"][0].max_length == 5 +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 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 TestChaosCaseExpand: """Tests for the ChaosCase.expand() class method.""" diff --git a/tests/strands_evals/chaos/test_effects.py b/tests/strands_evals/chaos/test_effects.py index fdfd5890..89d41fdb 100644 --- a/tests/strands_evals/chaos/test_effects.py +++ b/tests/strands_evals/chaos/test_effects.py @@ -4,7 +4,9 @@ from strands_evals.chaos.effects import ( CorruptValues, + EmptyResponse, ExecutionError, + FullRefusal, NetworkError, RemoveFields, Timeout, @@ -225,3 +227,22 @@ def test_corrupted_value_differs_from_original(self): def test_effect_type(self): effect = CorruptValues() assert effect.effect_type == "corrupt_values" + + +class TestPreEffectApply: + """Pre-hook model effects return cancel text from apply().""" + + def test_empty_response_apply_returns_single_space(self): + """EmptyResponse.apply() returns a single space so event.cancel is truthy.""" + result = EmptyResponse().apply() + assert result == " " + + def test_full_refusal_apply_returns_template(self): + """FullRefusal.apply() returns one of the refusal templates.""" + result = FullRefusal().apply() + assert result in FullRefusal._REFUSAL_TEMPLATES + + def test_pre_effect_apply_ignores_content(self): + """The content argument exists for the base signature and does not affect the result.""" + assert EmptyResponse().apply("ignored") == " " + assert FullRefusal().apply([{"text": "ignored"}]) in FullRefusal._REFUSAL_TEMPLATES diff --git a/tests/strands_evals/chaos/test_plugin.py b/tests/strands_evals/chaos/test_plugin.py index 6ff99e4a..868d8863 100644 --- a/tests/strands_evals/chaos/test_plugin.py +++ b/tests/strands_evals/chaos/test_plugin.py @@ -1,17 +1,29 @@ -"""Unit tests for ChaosPlugin.""" +"""Unit tests for ChaosPlugin (tool hooks and model hooks).""" +import copy import json +import logging from unittest.mock import MagicMock import pytest +from pydantic import BaseModel +from strands import Agent +from strands.hooks import BeforeModelCallEvent +from strands.models.model import Model from strands_evals.chaos import ChaosCase, ChaosPlugin from strands_evals.chaos._context import _current_chaos_case from strands_evals.chaos.effects import ( + Confabulation, + EmptyResponse, + FullRefusal, + MalformedJson, NetworkError, + SuccessFraming, Timeout, TruncateFields, ) +from strands_evals.chaos.plugin import _CHAOS_STATE_KEY, _MALFORMED_OUTPUT_APPLIED @pytest.fixture @@ -41,6 +53,133 @@ def after_event(): return event +@pytest.fixture +def activate_case(): + """Factory that activates a ChaosCase for the duration of the test. + + Accepts either a full ChaosCase or a list of model effects, which is wrapped as + effects={"model_effects": {"*": effects}}. The ContextVar is restored to its + pre-test state on teardown. + """ + tokens = [] + + def _activate(effects_or_case: list | ChaosCase) -> ChaosCase: + if isinstance(effects_or_case, ChaosCase): + case = effects_or_case + else: + case = ChaosCase( + name="test_case", + input="test input", + effects={"model_effects": {"*": effects_or_case}}, + ) + token = _current_chaos_case.set(case) + if not tokens: + tokens.append(token) + return case + + yield _activate + if tokens: + _current_chaos_case.reset(tokens[0]) + + +@pytest.fixture +def message_added_event(): + """Factory for a mock MessageAddedEvent.""" + + def _make(message: dict, dynamic_tools: dict | None = None) -> MagicMock: + """Build the event. + + 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 + + return _make + + +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"}]}}], + } + + +class _StructuredOutput(BaseModel): + """Structured output model for the agent-loop regression.""" + + answer: str + + +class _ScriptedModel(Model): + """Emits one valid structured-output toolUse per call and counts calls. + + The chaos plugin supplies the failure, so the model itself never needs to + produce an invalid payload. + """ + + def __init__(self, tool_name: str) -> None: + self.tool_name = tool_name + self.call_count = 0 + + def get_config(self): + return {} + + def update_config(self, **model_config): + pass + + async def structured_output(self, output_model, prompt, system_prompt=None, **kwargs): + yield {"output": output_model(answer="direct")} + + async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs): + self.call_count += 1 + yield {"messageStart": {"role": "assistant"}} + yield { + "contentBlockStart": {"start": {"toolUse": {"name": self.tool_name, "toolUseId": f"tu_{self.call_count}"}}} + } + yield {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"answer": "ok"}'}}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "tool_use"}} + yield { + "metadata": { + "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}, + "metrics": {"latencyMs": 1}, + } + } + + class TestChaosPluginBeforeToolCall: """Tests for the before_tool_call hook.""" @@ -200,3 +339,408 @@ def test_plain_text_truncation(self, chaos_plugin): assert event.result["content"][0]["text"] == "This" finally: _current_chaos_case.reset(token) + + +class TestEmptyResponsePreHook: + """EmptyResponse is a pre-hook effect — cancels model call with single space.""" + + def test_empty_response_cancels_with_single_space(self, chaos_plugin, activate_case): + """before_model_invocation sets event.cancel to ' ' (single space).""" + activate_case([EmptyResponse()]) + event = BeforeModelCallEvent(agent=MagicMock()) + + chaos_plugin.before_model_invocation(event) + + assert event.cancel == " " + + def test_empty_response_model_not_called(self, chaos_plugin, activate_case, message_added_event): + """When EmptyResponse fires as pre-hook, post-hook does not apply effects.""" + activate_case([EmptyResponse()]) + + # Pre-hook fires + pre_event = BeforeModelCallEvent(agent=MagicMock()) + chaos_plugin.before_model_invocation(pre_event) + assert pre_event.cancel == " " + + # SDK builds cancel message, MessageAddedEvent fires + cancel_message = {"role": "assistant", "content": [{"text": " "}]} + post_event = message_added_event(cancel_message) + chaos_plugin.after_model_invocation(post_event) + + # Content should be unchanged (pre effects skip post processing) + assert cancel_message["content"] == [{"text": " "}] + + +class TestFullRefusalPreHook: + """FullRefusal is a pre-hook effect — cancels model call with refusal text.""" + + def test_full_refusal_cancels_model_call(self, chaos_plugin, activate_case): + """before_model_invocation sets event.cancel to a refusal template.""" + activate_case([FullRefusal()]) + event = BeforeModelCallEvent(agent=MagicMock()) + + chaos_plugin.before_model_invocation(event) + + assert event.cancel in FullRefusal._REFUSAL_TEMPLATES + + def test_full_refusal_produces_single_turn(self, chaos_plugin, activate_case, message_added_event): + """FullRefusal cancels model call, SDK builds cancel message, run ends.""" + activate_case([FullRefusal()]) + + # Step 1: before_model_invocation fires + pre_event = BeforeModelCallEvent(agent=MagicMock()) + chaos_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 = message_added_event(cancel_message) + chaos_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 + + +class TestStructuredOutputFailureInjection: + """MalformedJson injects one structured-output parse failure per invocation.""" + + _EXPECTED_MESSAGE = "Structured output was malformed and could not be parsed. Please produce a corrected response." + + def _tool_event(self, tool_type, invocation_state=None): + event = MagicMock() + event.tool_use = {"name": "MyModel"} + event.selected_tool = MagicMock(tool_type=tool_type) + event.invocation_state = {} if invocation_state is None else invocation_state + event.cancel_tool = False + return event + + def test_structured_output_attempt_is_failed(self, chaos_plugin, activate_case, caplog): + """The first structured-output attempt is cancelled with the parse-failure message.""" + activate_case([MalformedJson()]) + event = self._tool_event("structured_output") + + with caplog.at_level(logging.INFO): + chaos_plugin.before_tool_call(event) + + assert event.cancel_tool == self._EXPECTED_MESSAGE + assert "injected structured output parse failure" in caplog.text + + def test_injection_is_one_shot_per_invocation(self, chaos_plugin, activate_case): + """A second attempt in the same invocation passes through so the agent can recover.""" + activate_case([MalformedJson()]) + invocation_state = {} + + first = self._tool_event("structured_output", invocation_state) + chaos_plugin.before_tool_call(first) + assert first.cancel_tool == self._EXPECTED_MESSAGE + + second = self._tool_event("structured_output", invocation_state) + chaos_plugin.before_tool_call(second) + assert second.cancel_tool is False + + def test_ordinary_tool_is_unaffected(self, chaos_plugin, activate_case): + """A non-structured-output tool is not cancelled by MalformedJson.""" + activate_case([MalformedJson()]) + event = self._tool_event("function") + + chaos_plugin.before_tool_call(event) + + assert event.cancel_tool is False + assert _CHAOS_STATE_KEY not in event.invocation_state + + def test_no_malformed_json_configured_writes_no_state(self, chaos_plugin, activate_case): + """Without MalformedJson the structured-output tool runs and no marker is written.""" + activate_case([Confabulation()]) + event = self._tool_event("structured_output") + + chaos_plugin.before_tool_call(event) + + assert event.cancel_tool is False + assert _CHAOS_STATE_KEY not in event.invocation_state + + +class TestToolUseMessagesNeverCorrupted: + """after_model_invocation leaves every message carrying a toolUse block untouched.""" + + def test_structured_output_tooluse_untouched(self, chaos_plugin, activate_case, message_added_event, caplog): + """A structured-output toolUse message is not corrupted and nothing is logged.""" + activate_case([MalformedJson(), SuccessFraming()]) + message = { + "role": "assistant", + "content": [ + {"toolUse": {"toolUseId": "so_1", "name": "MyModel", "input": {"field1": "value1"}}}, + ], + } + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + with caplog.at_level(logging.INFO): + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + assert "applied model output chaos" not in caplog.text + + def test_ordinary_tooluse_untouched(self, chaos_plugin, activate_case, message_added_event): + """An ordinary mid-turn toolUse message is not corrupted.""" + activate_case([MalformedJson()]) + message = _tooluse_assistant_message() + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def test_text_only_response_still_corrupted(self, chaos_plugin, activate_case, message_added_event): + """Final text responses remain corruptible.""" + activate_case([MalformedJson()]) + message = _final_assistant_message('{"key": "value", "nested": {"a": 1}}') + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"][0]["text"] != '{"key": "value", "nested": {"a": 1}}' + + +class TestInvocationStateCleanup: + """The per-invocation chaos marker is cleared at the invocation boundary.""" + + def test_marker_and_empty_parent_are_removed(self, chaos_plugin): + """A consumed marker is popped along with the now-empty chaos namespace.""" + invocation_state = {_CHAOS_STATE_KEY: {_MALFORMED_OUTPUT_APPLIED: True}} + event = MagicMock() + event.invocation_state = invocation_state + + chaos_plugin.after_invocation(event) + + assert _CHAOS_STATE_KEY not in invocation_state + + def test_parent_retained_when_other_chaos_state_remains(self, chaos_plugin): + """Unrelated chaos state keeps the namespace alive.""" + invocation_state = {_CHAOS_STATE_KEY: {_MALFORMED_OUTPUT_APPLIED: True, "other": 1}} + event = MagicMock() + event.invocation_state = invocation_state + + chaos_plugin.after_invocation(event) + + assert invocation_state[_CHAOS_STATE_KEY] == {"other": 1} + + def test_absent_or_non_dict_state_is_ignored(self, chaos_plugin): + """Missing or malformed chaos state is left untouched without raising.""" + empty = {} + event = MagicMock() + event.invocation_state = empty + chaos_plugin.after_invocation(event) + assert empty == {} + + non_dict = {_CHAOS_STATE_KEY: "unexpected"} + event = MagicMock() + event.invocation_state = non_dict + chaos_plugin.after_invocation(event) + assert non_dict == {_CHAOS_STATE_KEY: "unexpected"} + + +class TestStructuredOutputAgentLoop: + """Regression through the real SDK loop: injection fires once per invocation.""" + + async def test_reused_invocation_state_still_injects(self, chaos_plugin): + """Two invocations sharing one state dict each get exactly one injected failure.""" + # The async test body runs in its own context copy, so the case is set and reset + # inline rather than via the activate_case fixture (whose teardown runs outside it). + case = ChaosCase( + name="test_case", + input="test input", + effects={"model_effects": {"*": [MalformedJson()]}}, + ) + token = _current_chaos_case.set(case) + try: + model = _ScriptedModel(tool_name=_StructuredOutput.__name__) + agent = Agent(model=model, plugins=[chaos_plugin], callback_handler=None) + shared_state = {} + + first = await agent.invoke_async( + "first", invocation_state=shared_state, structured_output_model=_StructuredOutput + ) + calls_after_first = model.call_count + + second = await agent.invoke_async( + "second", invocation_state=shared_state, structured_output_model=_StructuredOutput + ) + finally: + _current_chaos_case.reset(token) + + # one failed attempt plus the corrected retry, per invocation + assert calls_after_first == 2 + assert model.call_count == 4 + # the caller still receives validated structured output both times + assert first.structured_output == _StructuredOutput(answer="ok") + assert second.structured_output == _StructuredOutput(answer="ok") + # the boundary cleanup ran, so the marker cannot leak into a later invocation + assert _CHAOS_STATE_KEY not in shared_state + + +class TestPostEffectsOnText: + """Post effects (Confabulation, MalformedJson-on-text, SuccessFraming) work.""" + + def test_confabulation_injects_template(self, chaos_plugin, activate_case, message_added_event): + """Confabulation injects fabricated citations into text content.""" + activate_case([Confabulation()]) + original_text = "The weather is sunny. It is warm outside. Birds are singing." + message = _final_assistant_message(original_text) + event = message_added_event(message) + + chaos_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, chaos_plugin, activate_case, message_added_event): + """MalformedJson truncates JSON-like text content.""" + activate_case([MalformedJson()]) + message = _final_assistant_message('{"key": "value", "nested": {"a": 1}}') + event = message_added_event(message) + + chaos_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, chaos_plugin, activate_case, message_added_event): + """SuccessFraming prepends a confident prefix to text content.""" + activate_case([SuccessFraming()]) + message = _final_assistant_message("Here is the result.") + event = message_added_event(message) + + chaos_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, chaos_plugin, activate_case, message_added_event): + """Confabulation + SuccessFraming compose: citation injected, then prefix prepended.""" + activate_case([Confabulation(), SuccessFraming()]) + original_text = "The weather is sunny. It is warm outside. Birds are singing." + message = _final_assistant_message(original_text) + event = message_added_event(message) + + chaos_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 + + +class TestMixedPrePostCase: + """Mixed pre+post effects: pre wins, post does NOT double-corrupt.""" + + def test_full_refusal_plus_malformed_json(self, chaos_plugin, activate_case, message_added_event): + """FullRefusal (pre) + MalformedJson (post): pre cancels, post skipped.""" + activate_case([FullRefusal(), MalformedJson()]) + + pre_event = BeforeModelCallEvent(agent=MagicMock()) + chaos_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 = message_added_event(cancel_message) + chaos_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, chaos_plugin, activate_case, message_added_event): + """EmptyResponse (pre) + SuccessFraming (post): pre cancels, post skipped.""" + activate_case([EmptyResponse(), SuccessFraming()]) + + pre_event = BeforeModelCallEvent(agent=MagicMock()) + chaos_plugin.before_model_invocation(pre_event) + assert pre_event.cancel == " " + + cancel_message = {"role": "assistant", "content": [{"text": " "}]} + post_event = message_added_event(cancel_message) + chaos_plugin.after_model_invocation(post_event) + + # SuccessFraming (post) should NOT have been applied + assert cancel_message["content"] == [{"text": " "}] + + +class TestGuardRoleFiltering: + """User and tool result messages are NOT corrupted.""" + + def test_user_message_not_corrupted(self, chaos_plugin, activate_case, message_added_event): + activate_case([Confabulation()]) + message = _user_message() + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + + def test_tool_result_message_not_corrupted(self, chaos_plugin, activate_case, message_added_event): + activate_case([Confabulation()]) + message = _tool_result_message() + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + + +class TestPassthrough: + """No corruption when no model_effects is set.""" + + def test_no_config_passes_through(self, chaos_plugin, message_added_event): + token = _current_chaos_case.set(None) + try: + message = _final_assistant_message("Hello world") + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + finally: + _current_chaos_case.reset(token) + + def test_empty_effects_passes_through(self, chaos_plugin, activate_case, message_added_event): + """ChaosCase with empty effects dict does not corrupt.""" + activate_case(ChaosCase(name="baseline", input="test", effects={})) + message = _final_assistant_message("Hello world") + original_content = copy.deepcopy(message["content"]) + event = message_added_event(message) + + chaos_plugin.after_model_invocation(event) + + assert message["content"] == original_content + + +class TestSelectPreModelEffect: + """_select_pre_model_effect returns the single pre effect regardless of position.""" + + def test_no_effects_returns_none(self, chaos_plugin, activate_case): + activate_case([]) + assert chaos_plugin._select_pre_model_effect() is None + + def test_post_only_returns_none(self, chaos_plugin, activate_case): + activate_case([MalformedJson(), SuccessFraming()]) + assert chaos_plugin._select_pre_model_effect() is None + + def test_single_pre_returned(self, chaos_plugin, activate_case): + pre = FullRefusal() + activate_case([pre]) + assert chaos_plugin._select_pre_model_effect() is pre + + def test_pre_after_posts_returned(self, chaos_plugin, activate_case): + pre = EmptyResponse() + activate_case([MalformedJson(), Confabulation(), pre]) + assert chaos_plugin._select_pre_model_effect() is pre