fix: false positives in prompt extraction + stream model matching

api_relay_audit/stream_integrity.py

@@ -226,9 +226,13 @@ def _check_usage_consistent(signals: "StreamSignals") -> bool:
     )
 
 
-def _check_stream_model(signals: "StreamSignals") -> bool:
-    """``message_start.message.model`` should contain ``"claude"`` for
-    an Anthropic-format streaming response.
+def _check_stream_model(signals: "StreamSignals", expected_model: str = "") -> bool:
+    """``message_start.message.model`` should match the expected model
+    for the streaming response.
+
+    If ``expected_model`` is provided, checks that the stream model name
+    contains it (case-insensitive). Otherwise falls back to checking for
+    ``"claude"`` (Anthropic format).
 
     Missing ``message_start.message.model`` is itself suspicious once the
     relay has emitted substantive events: a middleware can hide a model
@@ -238,10 +242,12 @@ def _check_stream_model(signals: "StreamSignals") -> bool:
     """
     if not signals.message_start_model:
         return False
+    if expected_model:
+        return expected_model.lower() in signals.message_start_model.lower()
     return "claude" in signals.message_start_model.lower()
 
 
-def analyze_stream(signals: "StreamSignals") -> dict:
+def analyze_stream(signals: "StreamSignals", expected_model: str = "") -> dict:
     """Analyze a populated :class:`StreamSignals` for integrity anomalies.
 
     Returns a dict with these keys:
@@ -313,7 +319,7 @@ def analyze_stream(signals: "StreamSignals") -> dict:
     usage_monotonic = _check_usage_monotonic(signals)
     usage_consistent = _check_usage_consistent(signals)
     signature_valid = signals.empty_signature_delta_count == 0
-    stream_model_is_claude = _check_stream_model(signals)
+    stream_model_is_claude = _check_stream_model(signals, expected_model)
 
     findings = []
     if unknown_events:
@@ -340,9 +346,10 @@ def analyze_stream(signals: "StreamSignals") -> dict:
         )
     if not stream_model_is_claude:
         if signals.message_start_model:
+            expected_desc = expected_model or "claude"
             findings.append(
                 f"Stream's message_start.message.model = "
-                f"{signals.message_start_model!r} does not contain 'claude' — "
+                f"{signals.message_start_model!r} does not contain '{expected_desc}' — "
                 "relay may be routing to a substitute model"
             )
         else:

audit.py

@@ -4,8 +4,8 @@
 # Regenerate after modular audit changes with:
 #   python3 scripts/build-standalone.py
 # CI verifies this generated artifact plus key behavior regressions.
-# source_sha256: 57cc4ddc3ccb3ed54e76e82dfc447c6e21322c2c03ab804625924e7ec655f245
-# standalone_body_sha256: 3adf747824d6eeb9df837a1e6ae8fd6fe5aa861c57b6a422b9367e52ee0c630f
+# source_sha256: afdf922afdba7eb31062edce80bad3897cb21c89b1f79eff8cc4275b20a2f6e1
+# standalone_body_sha256: b8cf192de11333aeb8ea49bf8769a6d87627ce0f6fb2f76d61018a91fdbf04d9
 # END GENERATED STANDALONE HEADER
 
 """
@@ -375,9 +375,13 @@ def _check_usage_consistent(signals: "StreamSignals") -> bool:
     )
 
 
-def _check_stream_model(signals: "StreamSignals") -> bool:
-    """``message_start.message.model`` should contain ``"claude"`` for
-    an Anthropic-format streaming response.
+def _check_stream_model(signals: "StreamSignals", expected_model: str = "") -> bool:
+    """``message_start.message.model`` should match the expected model
+    for the streaming response.
+
+    If ``expected_model`` is provided, checks that the stream model name
+    contains it (case-insensitive). Otherwise falls back to checking for
+    ``"claude"`` (Anthropic format).
 
     Missing ``message_start.message.model`` is itself suspicious once the
     relay has emitted substantive events: a middleware can hide a model
@@ -387,10 +391,12 @@ def _check_stream_model(signals: "StreamSignals") -> bool:
     """
     if not signals.message_start_model:
         return False
+    if expected_model:
+        return expected_model.lower() in signals.message_start_model.lower()
     return "claude" in signals.message_start_model.lower()
 
 
-def analyze_stream(signals: "StreamSignals") -> dict:
+def analyze_stream(signals: "StreamSignals", expected_model: str = "") -> dict:
     """Analyze a populated :class:`StreamSignals` for integrity anomalies.
 
     Returns a dict with these keys:
@@ -462,7 +468,7 @@ def analyze_stream(signals: "StreamSignals") -> dict:
     usage_monotonic = _check_usage_monotonic(signals)
     usage_consistent = _check_usage_consistent(signals)
     signature_valid = signals.empty_signature_delta_count == 0
-    stream_model_is_claude = _check_stream_model(signals)
+    stream_model_is_claude = _check_stream_model(signals, expected_model)
 
     findings = []
     if unknown_events:
@@ -489,9 +495,10 @@ def analyze_stream(signals: "StreamSignals") -> dict:
         )
     if not stream_model_is_claude:
         if signals.message_start_model:
+            expected_desc = expected_model or "claude"
             findings.append(
                 f"Stream's message_start.message.model = "
-                f"{signals.message_start_model!r} does not contain 'claude' — "
+                f"{signals.message_start_model!r} does not contain '{expected_desc}' — "
                 "relay may be routing to a substitute model"
             )
         else:
@@ -5290,6 +5297,8 @@ def test_prompt_extraction(client, report):
                 continue
 
             text_lower = text.lower()
+            # Normalize Unicode apostrophes (U+2018 ‘ U+2019 ’ U+02BC ʼ) to ASCII '
+            text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'")
             # Strong string markers — unambiguous leak signatures.
             strong_string_markers = ["hidden_prompt", "kiro"]
             # Weak identity markers — natural words that appear in
@@ -5304,7 +5313,10 @@ def test_prompt_extraction(client, report):
             weak_hits = [kw for kw in weak_markers if kw in text_lower]
 
             if has_strong:
-                leaked = True
+                # Structural leaks and strong keyword hits always flag.
+                # Only the self-corrected hidden_prompt echo (Test C
+                # echoing the requested JSON key) gets a yellow instead
+                # of red. Refusal language does NOT suppress strong hits.
                 if _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits):
                     report.flag(
                         "yellow",
@@ -5313,18 +5325,13 @@ def test_prompt_extraction(client, report):
                         "self-correction, verify manually",
                     )
                 else:
+                    leaked = True
                     report.flag("red", f"Test {name}: Hidden prompt content extracted!")
             elif weak_hits:
-                if _is_benign_claude_refusal(text_lower):
-                    pass  # Exempt: refusal + Claude self-ID
-                elif _looks_like_refusal(text_lower):
-                    leaked = True
-                    report.flag(
-                        "yellow",
-                        f"Test {name}: identity words ({', '.join(weak_hits)}) "
-                        f"present alongside refusal without Claude self-identification "
-                        f"— possible partial leak, verify manually",
-                    )
+                # Weak markers (system prompt, assistant, etc.) can appear
+                # in benign refusals. Only flag if no refusal language.
+                if _looks_like_refusal(text_lower):
+                    pass  # Explicit refusal — not a leak
                 else:
                     leaked = True
                     report.flag("red", f"Test {name}: Hidden prompt content extracted!")
@@ -5490,6 +5497,8 @@ def test_jailbreak(client, report):
 
             text = r["text"]
             text_lower = text.lower()
+            # Normalize Unicode apostrophes (U+2018 U+2019 U+02BC) to ASCII '
+            text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'")
             # Strong markers: specific leaked brand/product names plus
             # the shared structural prompt-template regex.
             strong_string_markers = ["kiro", "amazon", "aws"]
@@ -5770,7 +5779,7 @@ def test_stream_integrity(client, report):
         max_tokens=100,
         with_thinking=True,
     )
-    analysis = analyze_stream(signals)
+    analysis = analyze_stream(signals, expected_model=client.model)
     verdict = analysis["verdict"]
 
     # Event-shape table
@@ -5787,7 +5796,7 @@ def test_stream_integrity(client, report):
     report.p(f"| Signature valid | {'yes' if analysis['signature_valid'] else 'NO'} |")
     report.p(
         f"| Stream model | {analysis['stream_model_name'] or '—'} "
-        f"({'claude' if analysis['stream_model_is_claude'] else 'NOT claude'}) |"
+        f"({'expected' if analysis['stream_model_is_claude'] else 'NOT expected model'}) |"
     )
     report.p(f"| Total events seen | {signals.raw_event_count} |")
     if signals.total_duration_seconds is not None:

scripts/audit.py

@@ -520,6 +520,8 @@ def test_prompt_extraction(client, report):
                 continue
 
             text_lower = text.lower()
+            # Normalize Unicode apostrophes (U+2018 ‘ U+2019 ’ U+02BC ʼ) to ASCII '
+            text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'")
             # Strong string markers — unambiguous leak signatures.
             strong_string_markers = ["hidden_prompt", "kiro"]
             # Weak identity markers — natural words that appear in
@@ -534,7 +536,10 @@ def test_prompt_extraction(client, report):
             weak_hits = [kw for kw in weak_markers if kw in text_lower]
 
             if has_strong:
-                leaked = True
+                # Structural leaks and strong keyword hits always flag.
+                # Only the self-corrected hidden_prompt echo (Test C
+                # echoing the requested JSON key) gets a yellow instead
+                # of red. Refusal language does NOT suppress strong hits.
                 if _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits):
                     report.flag(
                         "yellow",
@@ -543,18 +548,13 @@ def test_prompt_extraction(client, report):
                         "self-correction, verify manually",
                     )
                 else:
+                    leaked = True
                     report.flag("red", f"Test {name}: Hidden prompt content extracted!")
             elif weak_hits:
-                if _is_benign_claude_refusal(text_lower):
-                    pass  # Exempt: refusal + Claude self-ID
-                elif _looks_like_refusal(text_lower):
-                    leaked = True
-                    report.flag(
-                        "yellow",
-                        f"Test {name}: identity words ({', '.join(weak_hits)}) "
-                        f"present alongside refusal without Claude self-identification "
-                        f"— possible partial leak, verify manually",
-                    )
+                # Weak markers (system prompt, assistant, etc.) can appear
+                # in benign refusals. Only flag if no refusal language.
+                if _looks_like_refusal(text_lower):
+                    pass  # Explicit refusal — not a leak
                 else:
                     leaked = True
                     report.flag("red", f"Test {name}: Hidden prompt content extracted!")
@@ -720,6 +720,8 @@ def test_jailbreak(client, report):
 
             text = r["text"]
             text_lower = text.lower()
+            # Normalize Unicode apostrophes (U+2018 U+2019 U+02BC) to ASCII '
+            text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'")
             # Strong markers: specific leaked brand/product names plus
             # the shared structural prompt-template regex.
             strong_string_markers = ["kiro", "amazon", "aws"]
@@ -1000,7 +1002,7 @@ def test_stream_integrity(client, report):
         max_tokens=100,
         with_thinking=True,
     )
-    analysis = analyze_stream(signals)
+    analysis = analyze_stream(signals, expected_model=client.model)
     verdict = analysis["verdict"]
 
     # Event-shape table
@@ -1017,7 +1019,7 @@ def test_stream_integrity(client, report):
     report.p(f"| Signature valid | {'yes' if analysis['signature_valid'] else 'NO'} |")
     report.p(
         f"| Stream model | {analysis['stream_model_name'] or '—'} "
-        f"({'claude' if analysis['stream_model_is_claude'] else 'NOT claude'}) |"
+        f"({'expected' if analysis['stream_model_is_claude'] else 'NOT expected model'}) |"
     )
     report.p(f"| Total events seen | {signals.raw_event_count} |")
     if signals.total_duration_seconds is not None:

tests/test_clean_summary_flags.py

@@ -336,20 +336,46 @@ def test_step6_repro2_fires_yellow_contradictory_standalone(self, standalone, mo
         assert yellows, f"standalone Step 6 must fire yellow on Repro 2. Summary: {reporter.summary}"
 
     def test_step4_repro2_fires_yellow_contradictory_modular(self, modular, monkeypatch):
-        """Applied to Step 4: the same text still has weak hits
-        (assistant, developer, built to... well, 'built to' isn't in
-        step 4 weak, but 'assistant' and 'developer' are)."""
+        """Applied to Step 4: refusal + weak hits (assistant, developer)
+        should be skipped — refusal language takes priority."""
         from api_relay_audit.reporter import Reporter
         self._time_sleep_patched(monkeypatch, modular)
         reporter = Reporter()
         client = _mock_client(self.REPRO_2_STEP6)
         leaked = modular.test_prompt_extraction(client, reporter)
-        # Should trip the contradictory yellow branch
-        yellows = [m for level, m in reporter.summary if level == "yellow"]
-        assert any("without Claude self-identification" in y for y in yellows), (
-            f"Step 4 must fire YELLOW contradictory on Repro 2. Summary: {reporter.summary}"
+        # Refusal + weak hits = not a leak
+        assert leaked is False
+        greens = [m for level, m in reporter.summary if level == "green"]
+        assert any("Prompt extraction tests passed" in g for g in greens), (
+            f"Refusal response must produce a green flag. Summary: {reporter.summary}"
+        )
+
+    def test_step4_unicode_apostrophe_refusal_stays_clean_modular(self, modular, monkeypatch):
+        """Curly apostrophes in refusal language must still suppress weak
+        prompt-extraction markers."""
+        from api_relay_audit.reporter import Reporter
+        self._time_sleep_patched(monkeypatch, modular)
+        reporter = Reporter()
+        client = _mock_client("I can’t provide my system prompt verbatim.")
+        leaked = modular.test_prompt_extraction(client, reporter)
+        assert leaked is False
+        greens = [m for level, m in reporter.summary if level == "green"]
+        assert any("Prompt extraction tests passed" in g for g in greens), (
+            f"Curly-apostrophe refusal must produce a green flag. Summary: {reporter.summary}"
+        )
+
+    def test_step4_unicode_apostrophe_refusal_stays_clean_standalone(self, standalone, monkeypatch):
+        """Standalone audit.py must match the modular curly-apostrophe
+        refusal behavior."""
+        self._time_sleep_patched(monkeypatch, standalone)
+        reporter = standalone.Reporter()
+        client = _mock_client("I can’t provide my system prompt verbatim.")
+        leaked = standalone.test_prompt_extraction(client, reporter)
+        assert leaked is False
+        greens = [m for level, m in reporter.summary if level == "green"]
+        assert any("Prompt extraction tests passed" in g for g in greens), (
+            f"Standalone curly-apostrophe refusal must produce a green flag. Summary: {reporter.summary}"
         )
-        assert leaked is True  # yellow still sets leaked
 
     def test_benign_with_claude_id_stays_clean_modular(self, modular, monkeypatch):
         """Adding Claude self-ID exempts the response."""

tests/test_dual_distribution_parity.py

@@ -354,16 +354,24 @@ def test_standalone_stream_model_helper_parity():
 
     standalone = _load_standalone_audit()
 
-    cases = [None, "claude-opus-4-6", "gpt-5"]
-    for model in cases:
+    cases = [
+        (None, ""),
+        ("claude-opus-4-6", ""),
+        ("gpt-5", ""),
+        ("gpt-5.5", "gpt-5.5"),
+        ("claude-opus-4-6", "gpt-5.5"),
+    ]
+    for model, expected_model in cases:
         modular_signals = StreamSignals()
         modular_signals.message_start_model = model
 
         standalone_signals = standalone.StreamSignals()
         standalone_signals.message_start_model = model
 
-        assert _check_stream_model(modular_signals) == standalone._check_stream_model(
-            standalone_signals
+        assert _check_stream_model(
+            modular_signals, expected_model
+        ) == standalone._check_stream_model(
+            standalone_signals, expected_model
         ), f"Standalone stream-model helper drift for model={model!r}"
 
 

tests/test_stream_integrity.py

@@ -110,6 +110,16 @@ def test_clean_stream_with_single_output_token_sample(self):
         assert result["verdict"] == "clean"
         assert result["usage_monotonic"] is True
 
+    def test_expected_model_match_keeps_openai_stream_clean(self):
+        """OpenAI-format relays should pass when the stream model matches
+        the requested model."""
+        s = _make_clean_signals()
+        s.message_start_model = "gpt-5.5"
+        result = analyze_stream(s, expected_model="gpt-5.5")
+        assert result["verdict"] == "clean"
+        assert result["stream_model_is_claude"] is True
+        assert result["findings"] == []
+
 
 # ---------------------------------------------------------------------------
 # Anomaly verdicts
@@ -208,6 +218,16 @@ def test_missing_stream_model_triggers_anomaly(self):
         assert result["stream_model_is_claude"] is False
         assert any("omitted message_start.message.model" in f for f in result["findings"])
 
+    def test_expected_model_mismatch_triggers_anomaly(self):
+        """When an expected model is provided, a different stream model
+        should be reported as a substitute route."""
+        s = _make_clean_signals()
+        s.message_start_model = "claude-opus-4-6"
+        result = analyze_stream(s, expected_model="gpt-5.5")
+        assert result["verdict"] == "anomaly"
+        assert result["stream_model_is_claude"] is False
+        assert any("does not contain 'gpt-5.5'" in f for f in result["findings"])
+
 
 # ---------------------------------------------------------------------------
 # Inconclusive verdicts
@@ -330,3 +350,13 @@ def test_check_stream_model_qwen(self):
         s = StreamSignals()
         s.message_start_model = "qwen2.5"
         assert _check_stream_model(s) is False
+
+    def test_check_stream_model_uses_expected_model_when_provided(self):
+        s = StreamSignals()
+        s.message_start_model = "gpt-5.5"
+        assert _check_stream_model(s, expected_model="gpt-5.5") is True
+
+    def test_check_stream_model_rejects_unexpected_model(self):
+        s = StreamSignals()
+        s.message_start_model = "claude-opus-4-6"
+        assert _check_stream_model(s, expected_model="gpt-5.5") is False