diff --git a/docs/SYSTEM_DESIGN.md b/docs/SYSTEM_DESIGN.md index 56ef57e..5154bea 100644 --- a/docs/SYSTEM_DESIGN.md +++ b/docs/SYSTEM_DESIGN.md @@ -305,7 +305,7 @@ Multiple operators on the same constraint are combined with logical AND: all pre ## 8. Minimization lifecycle -Goal: smallest message list that still yields the **same rule violation** under re-execution. +Goal: reduce the conversation while preserving **every target rule violation** under re-execution. ```mermaid flowchart TD @@ -314,14 +314,29 @@ flowchart TD RX -->|still violates| Keep[Accept subset] RX -->|not| Reject[Reject subset] Keep --> DD - DD --> Short{optional AI shorten?} - Short -->|yes| AI[Propose shorter text] - AI --> RX2[Re-exec gate] - RX2 -->|ok| Out[Minimized genome] + DD --> Short{intra_message enabled?} + Short -->|yes| Chars[Delete character partitions] + Chars --> RX2[Re-exec + PolicyEval] + RX2 -->|ok| Accept[Accept shorter text] RX2 -->|fail| KeepMsg[Keep previous text] + Accept --> Chars + KeepMsg --> Chars + Chars -->|partitions exhausted| Out[Minimized genome] Short -->|no| Out ``` +The Core Python API accepts `minimize_genome(..., intra_message=True)` to run +character-level ddmin on each retained message after turn-level shrinking. +The default is `False`; existing CLI and Hosted callers retain turn-only behavior. +Every accepted deletion resets and executes the target, then checks all requested +rule IDs through the same policy evaluator. Empty or whitespace-only proposals +are skipped. `reexec_count` includes both phases and `last_trace` describes the +accepted final conversation. Enabled runs record `ddmin_intra_message` in mutation +history and `intra_message=true` in genome metadata. + +This phase uses no LLM or tokenizer. It can require many more target executions, +especially when long text cannot be reduced, so callers must opt in deliberately. + **Invariant:** CLI and Hosted API must refuse regression save unless `still_reproduces=true` on the final genome. --- diff --git a/packages/mutiny_core/src/mutiny_core/minimize/__init__.py b/packages/mutiny_core/src/mutiny_core/minimize/__init__.py index 7adbc99..f14fc71 100644 --- a/packages/mutiny_core/src/mutiny_core/minimize/__init__.py +++ b/packages/mutiny_core/src/mutiny_core/minimize/__init__.py @@ -4,7 +4,7 @@ import copy import uuid -from typing import Sequence +from typing import Sequence, TypeVar from pydantic import BaseModel, Field @@ -15,6 +15,8 @@ from mutiny_core.policy.models import PolicySet from mutiny_core.trace.models import ExecutionTrace +_T = TypeVar("_T") + class MinimizeResult(BaseModel): """Outcome of minimizing a violating genome.""" @@ -40,11 +42,16 @@ def minimize_genome( target_rule_ids: Sequence[str], campaign_id: str | None = None, candidate_id: str | None = None, + intra_message: bool = False, ) -> MinimizeResult: """Delta-debug attacker messages; keep only subsets that preserve *same* rules. Reproduction = re-exec target → real trace → PolicyEvaluator → each ``target_rule_id`` still violated. No LLM involvement. + + ``intra_message=True`` additionally removes character partitions from each + retained message after turn-level ddmin. This opt-in phase can require many + more target executions; every accepted deletion uses the same rule gate. """ rules = list(target_rule_ids) or list(genome.target_rule_ids) original_turns = len(genome.messages) @@ -120,6 +127,38 @@ def _interesting(messages: list[AttackMessage]) -> tuple[bool, ExecutionTrace | break n = min(len(current), n * 2) + if intra_message: + for index in range(len(current)): + characters = list(current[index].content) + n = 2 + while len(characters) >= 2: + chunks = _partition(characters, n) + reduced = False + for i in range(len(chunks)): + complement = [ + c for j, chunk in enumerate(chunks) if j != i for c in chunk + ] + content = "".join(complement) + # Keep a real user turn rather than an empty/blank message. + if not content.strip(): + continue + probe_messages = [m.model_copy(deep=True) for m in current] + probe_messages[index] = AttackMessage(content=content) + ok, tr, viol = _interesting(probe_messages) + if ok: + current = probe_messages + characters = complement + last_trace = tr + last_violated = viol + n = max(n - 1, 2) + reduced = True + break + if reduced: + continue + if n >= len(characters): + break + n = min(len(characters), n * 2) + minimized = genome.model_copy(deep=True) minimized.id = str(uuid.uuid4()) minimized.parent_id = genome.id @@ -132,6 +171,9 @@ def _interesting(messages: list[AttackMessage]) -> tuple[bool, ExecutionTrace | "original_turns": original_turns, "minimized_turns": len(current), } + if intra_message: + minimized.mutations.append("ddmin_intra_message") + minimized.metadata["intra_message"] = True return MinimizeResult( original_genome=genome, @@ -148,12 +190,12 @@ def _interesting(messages: list[AttackMessage]) -> tuple[bool, ExecutionTrace | ) -def _partition(items: list[AttackMessage], n: int) -> list[list[AttackMessage]]: +def _partition(items: list[_T], n: int) -> list[list[_T]]: if n <= 1 or len(items) == 0: return [list(items)] n = min(n, len(items)) size, rem = divmod(len(items), n) - parts: list[list[AttackMessage]] = [] + parts: list[list[_T]] = [] idx = 0 for i in range(n): take = size + (1 if i < rem else 0) diff --git a/tests/unit/test_minimize.py b/tests/unit/test_minimize.py index 2ae08fa..cfc288a 100644 --- a/tests/unit/test_minimize.py +++ b/tests/unit/test_minimize.py @@ -4,6 +4,8 @@ from pathlib import Path +import pytest + from mutiny_core.adapter import TargetAdapter from mutiny_core.genome import AttackGenome, AttackMessage from mutiny_core.minimize import MinimizeResult, minimize_genome @@ -128,3 +130,144 @@ def context(self, session_id: str | None = None) -> dict: assert result.still_reproduces assert result.reexec_count == adapter.resets assert result.reexec_count >= 2 # at least original verify + one reduction attempt + + +class _TriggerAdapter(TargetAdapter): + """Mock tools fire only when their literal trigger is executed by a turn.""" + + def __init__(self): + self.probes: list[list[str]] = [] + + def reset(self, session_id: str) -> None: + self.probes.append([]) + + def step(self, session_id: str, user_message: str) -> AdapterTurnResult: + self.probes[-1].append(user_message) + return AdapterTurnResult(tool_calls=[ + ToolCall(id=f"{session_id}-{name}", name=name) + for name, trigger in (("alpha", "TRIGGER_A"), ("beta", "TRIGGER_B")) + if trigger in user_message + ]) + + def context(self, session_id: str | None = None) -> dict: + return {} + + +def _trigger_policy() -> PolicySet: + return PolicySet.model_validate({ + "version": "1", "target": "offline", + "rules": [ + {"id": f"no_{tool}", "tool": tool, "kind": "deny_tool", "description": "Mock restriction"} + for tool in ("alpha", "beta") + ], + }) + + +def _verbose_genome(*contents: str) -> AttackGenome: + return AttackGenome( + id="verbose", mutations=["seed"], target_rule_ids=["no_alpha"], + metadata={"source": {"label": "original"}}, + messages=[AttackMessage(content=text) for text in contents], + ) + + +@pytest.mark.parametrize("flag", [None, False]) +def test_intra_message_default_off_keeps_single_turn_content(flag): + genome = _verbose_genome("Hello there. Please run TRIGGER_A. Thank you for your help.") + adapter = _TriggerAdapter() + options = {} if flag is None else {"intra_message": flag} + result = minimize_genome( + genome, adapter=adapter, policy_set=_trigger_policy(), target_rule_ids=["no_alpha"], + **options, + ) + assert result.still_reproduces + assert result.minimized_genome.messages == genome.messages + assert result.reexec_count == 1 + assert result.minimized_genome.mutations == ["seed", "ddmin"] + assert "intra_message" not in result.minimized_genome.metadata + + +def test_intra_message_shrinks_single_verbose_turn_and_saves_replayable_evidence(): + from mutiny_core.regress import replay_regression, save_regression + + genome = _verbose_genome("My case reference is irrelevant. Please run TRIGGER_A. Thanks again.") + original = genome.model_dump() + adapter = _TriggerAdapter() + policy = _trigger_policy() + result = minimize_genome( + genome, adapter=adapter, policy_set=policy, target_rule_ids=["no_alpha"], + intra_message=True, campaign_id="campaign", candidate_id="candidate", + ) + assert result.still_reproduces + assert result.original_turn_count == result.minimized_turn_count == 1 + assert result.minimized_genome.messages[0].content == "TRIGGER_A" + assert len(result.minimized_genome.messages[0].content) < len(genome.messages[0].content) + assert result.violated_rule_ids == ["no_alpha"] + assert result.reexec_count == len(adapter.probes) > 1 + assert result.last_trace.turns[0].user_message == "TRIGGER_A" + assert result.minimized_genome.parent_id == genome.id + assert result.minimized_genome.mutations == ["seed", "ddmin", "ddmin_intra_message"] + assert result.minimized_genome.metadata["intra_message"] is True + assert result.campaign_id == "campaign" + assert result.candidate_id == "candidate" + assert genome.model_dump() == original + artifact = save_regression(result, name="small", target="offline", policy_set=policy) + assert artifact.conversation == ["TRIGGER_A"] + assert replay_regression(artifact, adapter=_TriggerAdapter(), policy_set=policy).status == "FAIL" + + +@pytest.mark.parametrize("targets,expected", [ + (["no_alpha", "no_beta"], ["TRIGGER_A", "TRIGGER_B"]), + (["no_beta"], ["TRIGGER_B"]), +]) +def test_intra_message_keeps_every_target_rule_instead_of_any_violation(targets, expected): + adapter = _TriggerAdapter() + result = minimize_genome( + _verbose_genome("Introduction. TRIGGER_A and TRIGGER_B. Closing remarks."), + adapter=adapter, policy_set=_trigger_policy(), target_rule_ids=targets, + intra_message=True, + ) + assert result.still_reproduces + content = result.minimized_genome.messages[0].content + for trigger in expected: + assert trigger in content + assert set(targets).issubset(result.violated_rule_ids) + assert result.reexec_count == len(adapter.probes) + assert result.last_trace.turns[0].user_message == content + + +def test_intra_message_runs_after_turn_shrink_and_reduces_each_retained_turn(): + result = minimize_genome( + _verbose_genome("Benign setup.", "First request: TRIGGER_A. Thanks.", "Next: TRIGGER_B. Goodbye."), + adapter=_TriggerAdapter(), policy_set=_trigger_policy(), + target_rule_ids=["no_alpha", "no_beta"], intra_message=True, + ) + assert result.still_reproduces + assert result.original_turn_count == 3 + assert result.minimized_turn_count == 2 + assert [m.content for m in result.minimized_genome.messages] == ["TRIGGER_A", "TRIGGER_B"] + assert [t.user_message for t in result.last_trace.turns] == ["TRIGGER_A", "TRIGGER_B"] + + +@pytest.mark.parametrize("contents", [(), ("Nothing happens here.",)]) +def test_intra_message_does_not_shrink_non_reproducing_input(contents): + genome = _verbose_genome(*contents) + adapter = _TriggerAdapter() + result = minimize_genome( + genome, adapter=adapter, policy_set=_trigger_policy(), + target_rule_ids=["no_alpha"], intra_message=True, + ) + assert not result.still_reproduces + assert result.minimized_genome == genome + assert result.reexec_count == len(adapter.probes) == bool(contents) + + +def test_intra_message_retains_already_minimal_content(): + adapter = _TriggerAdapter() + result = minimize_genome( + _verbose_genome("TRIGGER_A"), adapter=adapter, policy_set=_trigger_policy(), + target_rule_ids=["no_alpha"], intra_message=True, + ) + assert result.still_reproduces + assert result.minimized_genome.messages[0].content == "TRIGGER_A" + assert all(text.strip() for probe in adapter.probes for text in probe)