feat(minimize): add optional intra-message ddmin after turn shrinking

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.
 
 ---

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)

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)