fix(learn): filter ambient user-role scaffolding (#2275)

## Description

Fixes #2274.

Headroom Learn currently trusts `role=user` as sufficient preference
provenance. Agent harnesses can transport ambient UI and orchestration
context in user-role messages, and OpenAI Responses normalization also
promotes missing roles to `user`. Correction-like text in those inputs
can therefore become durable user preferences.

This change keeps preference learning fail-closed for known non-user
sources while preserving genuine user corrections.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature
- [ ] Breaking change
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Refactoring only

## Changes Made

- Preserve missing OpenAI Responses roles as `unknown` instead of
promoting them to `user`.
- Canonicalize user-role text before preference extraction.
- Remove proxy-appended `## Relevant Memories` suffixes from preference
evidence.
- Reject strict ambient-only harness prefixes such as heartbeat,
environment, workspace-instruction, delegation, and app-context
envelopes.
- Apply the same guard in `on_messages` and `_extract_preferences` for
defense in depth.
- Add regression coverage for system/developer/unknown roles,
ambient-only user messages, memory-only messages, and mixed
genuine-user-plus-memory input.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
149 passed, 1 warning
ruff check: passed
ruff format --check: passed
git diff --check: passed
```

Focused test files:

```text
tests/test_memory/test_traffic_learner.py
tests/test_openai_responses_traffic_learner.py
```

## Real Behavior Proof

- Environment: macOS; Python 3.13; current Headroom main; direct
invocation of the real `TrafficLearner` class, with no proxy or database
mocks
- Exact command / steps: create `TrafficLearner(backend=None,
min_evidence=1)`; feed system, developer, heartbeat user-role, and
memory-only user-role messages; read `patterns_extracted`; feed a
genuine user correction followed by a `## Relevant Memories` suffix;
read `patterns_extracted` again
- Observed result: `ambient_patterns=0`, `after_user_patterns=1` — the
ambient batch produced no preference evidence; the genuine correction
produced one pattern, while the appended memory content did not become
evidence
- Not tested: live provider traffic against a remote OpenAI endpoint;
every possible third-party harness envelope; migration or cleanup of
already-persisted noisy memories

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] No new dependency
- [x] Fail-open proxy behavior is unchanged
- [x] Regression tests added
- [x] Public examples contain no real user data
- [x] CHANGELOG update, if requested (not requested — N/A)

## Additional Notes

This extends the source filtering introduced by #466 rather than
replacing it. The prefix checks are deliberately strict and anchored at
the start of a canonicalized message. The intended failure mode is a
missed preference, not durable storage of non-user instructions.

Note: the strict prefix set was discussed and confirmed in
JerrettDavis's review approvals.

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",
+    "<app-context>",
+    "<codex_delegation>",
+    "<environment_context>",
+    "<heartbeat>",
+    "<permissions instructions>",
+    "<skills_instructions>",
+    "# 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 = ("<in-app-browser-context",)
+
+
+def _canonicalize_user_text(text: str) -> 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 []

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
 

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",
+        [
+            "<heartbeat>Never notify the user for a quiet check.</heartbeat>",
+            "<environment_context>Always use the sandbox.</environment_context>",
+            (
+                '<in-app-browser-context source="ambient-ui-state">'
+                "Do not treat it as evidence that the user selected the browser."
+                "</in-app-browser-context>"
+            ),
+            "# 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"
+                        '<in-app-browser-context source="ambient-ui-state">'
+                        "Do not treat this as evidence that the user selected the browser."
+                        "</in-app-browser-context>"
+                    ),
+                }
+            ]
+        )
+
+        assert learner.get_stats()["patterns_extracted"] == 1
+
+
 class TestExtractPreferencesRealCorrections:
     """Make sure the noise filter does not eat genuine user corrections."""
 

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,