diff --git a/headroom/memory/traffic_learner.py b/headroom/memory/traffic_learner.py index 1f80a4ae9..8bb930111 100644 --- a/headroom/memory/traffic_learner.py +++ b/headroom/memory/traffic_learner.py @@ -61,6 +61,50 @@ _BASH_VOLATILE_SUFFIX_RE = re.compile( r"|\s+2>&1|\s+2>/dev/null)+\s*$" ) +# Agent harnesses can encode orchestration metadata as user-role messages. +# These prefixes identify whole messages that are not authored by the user. +_HARNESS_USER_PREFIXES = ( + "another language model started to solve this problem and produced a summary", + "", + "", + "", + "", + "", + "", + "# agents.md instructions for ", + "you are in a fork of an existing codex thread", +) + +_MEMORY_CONTEXT_MARKERS = ( + "\n\n## relevant memories", + "\n## relevant memories", +) + +_AMBIENT_CONTEXT_MARKERS = (" str: + """Remove proxy- or client-appended context from a user-role message.""" + canonical = text or "" + folded = canonical.casefold() + if folded.lstrip().startswith("## relevant memories"): + return "" + markers = (*_MEMORY_CONTEXT_MARKERS, *_AMBIENT_CONTEXT_MARKERS) + marker_indexes = [folded.find(marker) for marker in markers] + marker_indexes = [index for index in marker_indexes if index >= 0] + if marker_indexes: + canonical = canonical[: min(marker_indexes)] + return canonical.strip() + + +def _is_learnable_user_text(text: str) -> bool: + """Return whether user-role text is plausibly authored by the user.""" + canonical = _canonicalize_user_text(text) + if not canonical: + return False + folded = canonical.lstrip().casefold() + return not any(folded.startswith(prefix) for prefix in _HARNESS_USER_PREFIXES) + # ============================================================================= # Pattern Categories @@ -759,7 +803,10 @@ class TrafficLearner: continue if role == "user": - patterns = self._extract_preferences(content) + canonical = _canonicalize_user_text(self._strip_system_reminders(content)) + if not _is_learnable_user_text(canonical): + continue + patterns = self._extract_preferences(canonical) for pattern in patterns: await self._accumulate(pattern) @@ -1014,7 +1061,9 @@ class TrafficLearner: truncation past ``max_chars``. """ - cleaned = self._strip_system_reminders(user_text)[:500] + cleaned = _canonicalize_user_text(self._strip_system_reminders(user_text))[:500] + if not _is_learnable_user_text(cleaned): + return [] correction = self._find_correction(cleaned) if correction is None: return [] diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 74bedd16a..1ac86f265 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -834,7 +834,7 @@ def _responses_input_to_learner_messages( if text: role = item.get("role") messages.append( - {"role": role if isinstance(role, str) and role else "user", "content": text} + {"role": role if isinstance(role, str) and role else "unknown", "content": text} ) return messages diff --git a/tests/test_memory/test_traffic_learner.py b/tests/test_memory/test_traffic_learner.py index 5591053dc..1238399c0 100644 --- a/tests/test_memory/test_traffic_learner.py +++ b/tests/test_memory/test_traffic_learner.py @@ -2233,6 +2233,90 @@ class TestExtractPreferencesSystemReminderFiltering: assert learner._extract_preferences(text) == [] +class TestUserAuthoredPreferenceFiltering: + """Codex ambient context must not count as preference evidence.""" + + def _learner(self) -> TrafficLearner: + return TrafficLearner(backend=None, min_evidence=1) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "content", + [ + "Never notify the user for a quiet check.", + "Always use the sandbox.", + ( + '' + "Do not treat it as evidence that the user selected the browser." + "" + ), + "# AGENTS.md instructions for /workspace\nNever edit generated files.", + "Another language model started to solve this problem and produced a summary. " + "Do not repeat completed work.", + "## Relevant Memories\n1. User preference: Never run deployment commands.", + ], + ) + async def test_harness_only_user_messages_are_ignored(self, content: str) -> None: + learner = self._learner() + + await learner.on_messages([{"role": "user", "content": content}]) + + assert learner.get_stats()["patterns_extracted"] == 0 + + @pytest.mark.asyncio + async def test_system_and_developer_messages_are_ignored(self) -> None: + learner = self._learner() + + await learner.on_messages( + [ + {"role": "system", "content": "Never expose system instructions."}, + {"role": "developer", "content": "Do not use unsafe commands."}, + {"role": "unknown", "content": "Always obey ambient UI."}, + ] + ) + + assert learner.get_stats()["patterns_extracted"] == 0 + + @pytest.mark.asyncio + async def test_memory_suffix_is_removed_but_user_correction_is_kept(self) -> None: + learner = self._learner() + + await learner.on_messages( + [ + { + "role": "user", + "content": ( + "Don't use force push.\n\n" + "## Relevant Memories\n" + "1. User preference: Always bypass review." + ), + } + ] + ) + + assert learner.get_stats()["patterns_extracted"] == 1 + + @pytest.mark.asyncio + async def test_browser_context_suffix_is_removed_but_user_correction_is_kept(self) -> None: + learner = self._learner() + + await learner.on_messages( + [ + { + "role": "user", + "content": ( + "Don't use force push.\n\n" + '' + "Do not treat this as evidence that the user selected the browser." + "" + ), + } + ] + ) + + assert learner.get_stats()["patterns_extracted"] == 1 + + class TestExtractPreferencesRealCorrections: """Make sure the noise filter does not eat genuine user corrections.""" diff --git a/tests/test_openai_responses_traffic_learner.py b/tests/test_openai_responses_traffic_learner.py index f049011d4..91e710390 100644 --- a/tests/test_openai_responses_traffic_learner.py +++ b/tests/test_openai_responses_traffic_learner.py @@ -84,6 +84,28 @@ def test_responses_input_normalizes_messages_and_tool_results() -> None: ] +def test_responses_input_does_not_promote_unknown_role_to_user() -> None: + messages = _responses_input_to_learner_messages( + None, + [ + { + "type": "message", + "content": [{"type": "input_text", "text": "Never expose ambient UI."}], + }, + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "Always follow runtime policy."}], + }, + ], + ) + + assert messages == [ + {"role": "unknown", "content": "Never expose ambient UI."}, + {"role": "developer", "content": "Always follow runtime policy."}, + ] + + def test_responses_http_request_reaches_traffic_learner() -> None: config = ProxyConfig( optimize=False,