Merge d89568e524 into 27b4e2d147

headroom/learn/models.py

@@ -163,6 +163,18 @@ class Recommendation:
     # above one-off rules because their waste scales with repetition.
     is_loop_guardrail: bool = False
     loop_occurrences: int = 0  # Repetitions of the loop this rule guards against
+    # Preserve prior markdown-list items in the same section. This is opt-in
+    # because most analyzers treat a re-surfaced section as authoritative.
+    preserve_prior_items: bool = False
+    # Authoritative lifecycle signal for `preserve_prior_items`: the pattern
+    # ids the producing learner still considers active, including the ones
+    # left out of this batch by ranking or top-N capping. A prior item
+    # survives only while its id is in this set, which is what makes deletion
+    # possible — expired, tombstoned, or disproven items drop out of the set
+    # and are then removed from the file. `None` means the producer has no
+    # lifecycle signal at all; preservation then falls back to a plain union
+    # of new and prior items.
+    active_item_ids: frozenset[str] | None = None
 
 
 @dataclass

headroom/learn/writer.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import re
 from abc import ABC, abstractmethod
+from dataclasses import replace
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -109,6 +110,122 @@ def _build_section(recommendations: list[Recommendation]) -> str:
 
 # Matches the "*~N tokens/session saved*" annotation emitted by _build_section.
 _TOKENS_ANNOTATION_PATTERN = re.compile(r"\*~([\d,]+) tokens/session saved\*\n?")
+_PATTERN_ID_PATTERN = re.compile(r"<!--\s*headroom:pattern-id:([^\s>]+)\s*-->\s*$")
+
+
+def _merge_markdown_items(
+    new_content: str,
+    prior_content: str,
+    active_item_ids: frozenset[str] | None = None,
+) -> str | None:
+    """Merge simple markdown bullets, preferring new text for stable IDs.
+
+    ``active_item_ids`` is the producing learner's authoritative set of ids
+    that are still alive — it must include the items that this batch omitted
+    only because of ranking or top-N capping, otherwise still-active advice
+    is deleted. When it is supplied, a prior item is carried forward only
+    while its `headroom:pattern-id` is in the set, so an expired or
+    tombstoned item is genuinely removed instead of being pinned forever.
+    Prior items with no id predate id tagging: a still-active one is
+    re-emitted by the current run with an id and collapses into that line by
+    visible text, so dropping the untagged leftovers removes only items the
+    learner no longer considers active.
+
+    Passing ``None`` means the producer exposes no lifecycle signal, and the
+    merge degrades to a plain union of new and prior items.
+    """
+
+    def _items(content: str) -> list[tuple[str | None, str, str]] | None:
+        lines = [line.strip() for line in content.splitlines() if line.strip()]
+        if any(not line.startswith("- ") for line in lines):
+            return None
+        items: list[tuple[str | None, str, str]] = []
+        for line in lines:
+            id_match = _PATTERN_ID_PATTERN.search(line)
+            pattern_id = id_match.group(1) if id_match else None
+            visible = _PATTERN_ID_PATTERN.sub("", line).strip().casefold()
+            items.append((pattern_id, visible, line))
+        return items
+
+    new_items = _items(new_content)
+    prior_items = _items(prior_content)
+    if new_items is None or prior_items is None:
+        return None
+
+    if active_item_ids is not None:
+        prior_items = [
+            item for item in prior_items if item[0] is not None and item[0] in active_item_ids
+        ]
+
+    merged: list[str] = []
+    seen_ids: set[str] = set()
+    seen_content: set[str] = set()
+    for pattern_id, visible, line in (*new_items, *prior_items):
+        if (pattern_id is not None and pattern_id in seen_ids) or visible in seen_content:
+            continue
+        if pattern_id is not None:
+            seen_ids.add(pattern_id)
+        seen_content.add(visible)
+        merged.append(line)
+    return "\n".join(merged)
+
+
+def _authoritative_item_ids(
+    recommendations: list[Recommendation],
+) -> frozenset[str] | None:
+    """Union every lifecycle signal the current run carries, or None.
+
+    A section the new run did not re-emit cannot be judged by its own
+    recommendation — that recommendation is exactly what is missing. The
+    sets the run *does* carry stand in for it: a producer publishes one set
+    per run covering all of its live items, so an id absent from every set
+    in the run is one no producer still claims. Returns ``None`` when no
+    recommendation carries a signal, which keeps the historical
+    carry-everything behaviour for runs that cannot speak to lifecycle.
+    """
+    signals = [r.active_item_ids for r in recommendations if r.active_item_ids is not None]
+    if not signals:
+        return None
+    return frozenset().union(*signals)
+
+
+def _prune_carried_section(
+    content: str,
+    active_item_ids: frozenset[str],
+) -> str | None:
+    """Drop expired id-tagged bullets from a section the new run did not re-emit.
+
+    This is the deletion path for a heading whose last item expired: the
+    producer stops emitting the section entirely, so the same-section merge
+    never runs and the heading would otherwise be carried forward forever.
+
+    Only bullets carrying a `headroom:pattern-id` are removable — they come
+    from a lifecycle-tracked producer, so their absence from
+    ``active_item_ids`` means that producer dropped them. Untagged bullets
+    are kept: unlike the same-section merge, where a still-active legacy
+    item is re-emitted with an id by the same run and collapses by visible
+    text, nothing in this path would bring an untagged bullet back, so
+    deleting it would discard content on no evidence.
+
+    Returns the pruned content, or ``None`` when the section carries no
+    tagged bullets at all and is therefore not a tracked section to prune.
+    """
+    lines = [line.strip() for line in content.splitlines() if line.strip()]
+    if not lines or any(not line.startswith("- ") for line in lines):
+        return None
+    kept: list[str] = []
+    saw_tracked_item = False
+    for line in lines:
+        id_match = _PATTERN_ID_PATTERN.search(line)
+        if id_match is None:
+            kept.append(line)
+            continue
+        saw_tracked_item = True
+        if id_match.group(1) in active_item_ids:
+            kept.append(line)
+    if not saw_tracked_item:
+        return None
+    return "\n".join(kept)
 
 
 def extract_marker_block(file_content: str) -> str | None:
@@ -170,15 +287,52 @@ def _merge_recommendations(
     whose headings do not reappear in the new run are carried forward so
     a re-run doesn't silently drop accumulated learnings. To fully rebuild
     the block, delete it manually and re-run.
+
+    Recommendations that opt into ``preserve_prior_items`` are merged at the
+    item level instead, bounded by their ``active_item_ids`` lifecycle signal
+    so prior items can still expire out of the file.
+
+    That signal also reaches the carried-forward sections. A category whose
+    last item expires stops producing a recommendation at all, so its
+    heading never enters the same-section merge; without pruning the carry
+    path too, those bullets would be pinned in the file forever. Sections
+    holding no id-tagged items are carried untouched, as before.
     """
     if not file_path.exists():
         return new_recommendations
     prior = _parse_prior_recommendations(_read_text_tolerant(file_path))
     if not prior:
         return new_recommendations
-    new_sections = {r.section for r in new_recommendations}
-    carried = [p for p in prior if p.section not in new_sections]
-    return list(new_recommendations) + carried
+    prior_by_section = {r.section: r for r in prior}
+    merged_new: list[Recommendation] = []
+    for recommendation in new_recommendations:
+        prior_recommendation = prior_by_section.get(recommendation.section)
+        if recommendation.preserve_prior_items and prior_recommendation is not None:
+            merged_content = _merge_markdown_items(
+                recommendation.content,
+                prior_recommendation.content,
+                recommendation.active_item_ids,
+            )
+            if merged_content is not None:
+                recommendation = replace(recommendation, content=merged_content)
+        merged_new.append(recommendation)
+
+    new_sections = {r.section for r in merged_new}
+    run_active_item_ids = _authoritative_item_ids(merged_new)
+    carried: list[Recommendation] = []
+    for prior_recommendation in prior:
+        if prior_recommendation.section in new_sections:
+            continue
+        if run_active_item_ids is not None:
+            pruned = _prune_carried_section(prior_recommendation.content, run_active_item_ids)
+            if pruned is not None:
+                if not pruned:
+                    # Every tracked item under this heading is gone; the
+                    # heading goes with them rather than outliving them.
+                    continue
+                prior_recommendation = replace(prior_recommendation, content=pruned)
+        carried.append(prior_recommendation)
+    return merged_new + carried
 
 
 def _merge_into_file(file_path: Path, new_recommendations: list[Recommendation]) -> str:

headroom/memory/traffic_learner.py

@@ -1748,6 +1748,14 @@ def _patterns_to_recommendations(patterns: list[ExtractedPattern]) -> list:
     """
     from headroom.learn.models import Recommendation, RecommendationTarget
 
+    # Authoritative lifecycle signal for the item-level merge in the writer:
+    # every pattern the learner still holds for this project, captured before
+    # any per-category ranking or capping so that an item omitted from a
+    # rendered section is not mistaken for an expired one. `_collect_all_patterns`
+    # already dropped rows that no longer exist in memory.db, so an id missing
+    # here means the pattern is gone, not merely unrendered.
+    active_item_ids = frozenset(p.content_hash for p in patterns if p.content_hash)
+
     by_category: dict[PatternCategory, list[ExtractedPattern]] = {}
     for p in patterns:
         by_category.setdefault(p.category, []).append(p)
@@ -1769,7 +1777,15 @@ def _patterns_to_recommendations(patterns: list[ExtractedPattern]) -> list:
             items.sort(key=lambda p: p.evidence_count, reverse=True)
         if not items:
             continue
-        bullets = "\n".join(f"- {p.content}" for p in items)
+        preserve_prior_items = category is not PatternCategory.ERROR_RECOVERY
+        bullets = "\n".join(
+            (
+                f"- {p.content} <!-- headroom:pattern-id:{p.content_hash} -->"
+                if preserve_prior_items
+                else f"- {p.content}"
+            )
+            for p in items
+        )
         recs.append(
             Recommendation(
                 target=target,
@@ -1777,6 +1793,12 @@ def _patterns_to_recommendations(patterns: list[ExtractedPattern]) -> list:
                 content=bullets,
                 confidence=max((p.importance for p in items), default=0.5),
                 evidence_count=sum(p.evidence_count for p in items),
+                preserve_prior_items=preserve_prior_items,
+                # error_recovery is rebuilt from scratch on every render and
+                # `_refine_error_recovery` deliberately drops rows, so the
+                # pre-refine set above is not its lifecycle signal; it must
+                # stay unset while that section replaces rather than merges.
+                active_item_ids=active_item_ids if preserve_prior_items else None,
             )
         )
     return recs

tests/test_learn/test_section_lifecycle.py

@@ -0,0 +1,218 @@
+"""Removal invariant for merged sections — the deletion half of #2293.
+
+``test_writer.py`` covers preservation inside a section that the new run
+re-emits. The cases here cover the other transition: a section the new run
+does not emit at all, which never reaches the same-section merge and so has
+to be pruned on the writer's carry-forward path instead. Without that, the
+last item of a category can expire and its heading still be pinned in the
+file forever.
+
+The learner-level test lives here rather than in ``test_traffic_learner.py``
+because it asserts on writer behaviour; it drives the real, unmodified
+``_patterns_to_recommendations`` to get there.
+"""
+
+from headroom.learn.models import Recommendation, RecommendationTarget
+from headroom.learn.writer import _merge_into_file
+from headroom.memory.traffic_learner import (
+    ExtractedPattern,
+    PatternCategory,
+    _patterns_to_recommendations,
+)
+
+
+def _rec(section: str, content: str) -> Recommendation:
+    return Recommendation(
+        target=RecommendationTarget.CONTEXT_FILE,
+        section=section,
+        content=content,
+        confidence=0.8,
+        evidence_count=5,
+    )
+
+
+def _block(*sections: tuple[str, str]) -> str:
+    parts = ["<!-- headroom:learn:start -->", "## Headroom Learned Patterns", ""]
+    for heading, body in sections:
+        parts += [f"### {heading}", body, ""]
+    parts.append("<!-- headroom:learn:end -->")
+    return "\n".join(parts) + "\n"
+
+
+class TestCarriedSectionLifecycle:
+    """The new run's lifecycle signal also governs the sections it omits."""
+
+    def test_absent_tracked_section_is_removed_once_its_items_expire(self, tmp_path):
+        """A heading outlives its items unless the carry path is pruned too.
+
+        When a category loses its last pattern the learner stops emitting a
+        recommendation for that heading at all, so the same-section merge
+        never runs on it. The signal published by the rest of the run has to
+        reach the carry path, or the section is pinned forever.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            _block(
+                (
+                    "Learned: architecture",
+                    "- Handlers live under proxy/handlers <!-- headroom:pattern-id:handlers -->",
+                ),
+                (
+                    "Learned: preference",
+                    "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+                ),
+            ),
+            encoding="utf-8",
+        )
+        recommendation = _rec(
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        recommendation.active_item_ids = frozenset({"reviews"})
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert "### Learned: architecture" not in final
+        assert "Handlers live under proxy/handlers" not in final
+        assert final.count("Keep local reviews") == 1
+
+    def test_absent_tracked_section_keeps_its_still_active_items(self, tmp_path):
+        """Pruning a carried section is per item, not all-or-nothing.
+
+        A category can be missing from a render because of batching while
+        still holding live items, so only the ids the run no longer claims
+        may go.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            _block(
+                (
+                    "Learned: architecture",
+                    "- Handlers live under proxy/handlers <!-- headroom:pattern-id:handlers -->\n"
+                    "- Build against the vendored SDK <!-- headroom:pattern-id:vendored -->",
+                ),
+                (
+                    "Learned: preference",
+                    "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+                ),
+            ),
+            encoding="utf-8",
+        )
+        recommendation = _rec(
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        recommendation.active_item_ids = frozenset({"reviews", "handlers"})
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert "### Learned: architecture" in final
+        assert "Handlers live under proxy/handlers" in final
+        assert "Build against the vendored SDK" not in final
+
+    def test_absent_untracked_sections_are_carried_untouched(self, tmp_path):
+        """Only id-tagged sections are ours to delete.
+
+        Hand-written headings and prose bodies carry no lifecycle signal,
+        and — unlike the same-section merge, where a still-active legacy
+        item is re-emitted with an id by the same run — nothing on this path
+        would bring them back, so they must survive the prune.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            _block(
+                ("Hand-written notes", "- Deploy from the release branch"),
+                ("Project shape", "Prose, deliberately not a markdown list."),
+                (
+                    "Learned: preference",
+                    "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+                ),
+            ),
+            encoding="utf-8",
+        )
+        recommendation = _rec(
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        recommendation.active_item_ids = frozenset({"reviews"})
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert "Deploy from the release branch" in final
+        assert "Prose, deliberately not a markdown list." in final
+
+    def test_absent_tracked_section_survives_a_run_without_a_lifecycle_signal(self, tmp_path):
+        """No signal, no deletion.
+
+        A run whose recommendations carry no ``active_item_ids`` cannot
+        speak to what is still alive, so the carry path stays conservative
+        rather than reading silence as expiry.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            _block(
+                (
+                    "Learned: architecture",
+                    "- Handlers live under proxy/handlers <!-- headroom:pattern-id:handlers -->",
+                ),
+                ("Learned: error recovery", "- Search before reading a guessed path"),
+            ),
+            encoding="utf-8",
+        )
+        recommendation = _rec(
+            "Learned: error recovery",
+            "- Search before reading a guessed path",
+        )
+        assert recommendation.active_item_ids is None
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert "### Learned: architecture" in final
+        assert "Handlers live under proxy/handlers" in final
+
+
+class TestTrafficLearnerCategoryLifecycle:
+    """End-to-end through the real learner rendering."""
+
+    def test_category_losing_its_last_pattern_drops_its_heading(self, tmp_path):
+        """An emptied category leaves the file entirely.
+
+        This is the transition the per-item merge cannot see. Once
+        ``architecture`` holds nothing, ``_patterns_to_recommendations``
+        emits no recommendation for that heading, so the removal has to
+        happen on the writer's carry-forward path instead. The existing
+        end-to-end case keeps a second item in the same category, so it
+        never reaches this transition.
+        """
+        keep = ExtractedPattern(
+            category=PatternCategory.PREFERENCE,
+            content="User prefers terse output",
+            importance=0.8,
+            evidence_count=3,
+        )
+        sole_architecture_pattern = ExtractedPattern(
+            category=PatternCategory.ARCHITECTURE,
+            content="Handlers live under headroom/proxy/handlers",
+            importance=0.8,
+            evidence_count=3,
+        )
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            _merge_into_file(
+                context_file,
+                _patterns_to_recommendations([keep, sole_architecture_pattern]),
+            ),
+            encoding="utf-8",
+        )
+        assert "### Learned: architecture" in context_file.read_text()
+
+        # Next render: architecture has no live pattern left, so the whole
+        # category stops being rendered rather than rendering fewer bullets.
+        final = _merge_into_file(context_file, _patterns_to_recommendations([keep]))
+
+        assert "User prefers terse output" in final
+        assert "### Learned: architecture" not in final
+        assert "Handlers live under headroom/proxy/handlers" not in final

tests/test_learn/test_writer.py

@@ -157,6 +157,117 @@ class TestClaudeCodeWriter:
         # Only one Environment section in the final block
         assert content.count("### Environment") == 1
 
+    def test_opt_in_same_section_merge_preserves_prior_pattern_items(self, tmp_path):
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            "<!-- headroom:learn:start -->\n"
+            "## Headroom Learned Patterns\n\n"
+            "### Learned: preference\n"
+            "- Keep the established queue <!-- headroom:pattern-id:queue -->\n"
+            "- Keep local reviews\n\n"
+            "<!-- headroom:learn:end -->\n"
+        )
+        recommendation = _rec(
+            RecommendationTarget.CONTEXT_FILE,
+            "Learned: preference",
+            "- Use the updated queue <!-- headroom:pattern-id:queue -->\n"
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert "Use the updated queue" in final
+        assert "Keep the established queue" not in final
+        assert final.count("Keep local reviews") == 1
+
+    def test_active_ids_keep_unbatched_items_and_drop_expired_ones(self, tmp_path):
+        """The removal invariant: preservation is bounded by the active id set.
+
+        ``ripgrep`` is active but was left out of this batch (ranking/top-N),
+        so it must survive. ``vendored`` is no longer active, so it must be
+        deleted rather than pinned into the file forever.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            "<!-- headroom:learn:start -->\n"
+            "## Headroom Learned Patterns\n\n"
+            "### Learned: preference\n"
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->\n"
+            "- Prefer ripgrep over grep <!-- headroom:pattern-id:ripgrep -->\n"
+            "- Build against the vendored SDK <!-- headroom:pattern-id:vendored -->\n\n"
+            "<!-- headroom:learn:end -->\n"
+        )
+        recommendation = _rec(
+            RecommendationTarget.CONTEXT_FILE,
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        recommendation.active_item_ids = frozenset({"reviews", "ripgrep"})
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert final.count("Keep local reviews") == 1
+        assert "Prefer ripgrep over grep" in final
+        assert "Build against the vendored SDK" not in final
+        assert "headroom:pattern-id:vendored" not in final
+
+    def test_active_ids_drop_untagged_legacy_items(self, tmp_path):
+        """Untagged prior items pre-date id tagging and are not a lifecycle signal.
+
+        The still-active one comes back with an id from the current run and
+        collapses into a single bullet; the one the learner dropped goes away.
+        """
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            "<!-- headroom:learn:start -->\n"
+            "## Headroom Learned Patterns\n\n"
+            "### Learned: preference\n"
+            "- Keep local reviews\n"
+            "- Build against the vendored SDK\n\n"
+            "<!-- headroom:learn:end -->\n"
+        )
+        recommendation = _rec(
+            RecommendationTarget.CONTEXT_FILE,
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        recommendation.active_item_ids = frozenset({"reviews"})
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert final.count("Keep local reviews") == 1
+        assert "headroom:pattern-id:reviews" in final
+        assert "Build against the vendored SDK" not in final
+
+    def test_without_active_ids_prior_items_are_unioned(self, tmp_path):
+        """No lifecycle signal — every producer that predates it keeps the union."""
+        context_file = tmp_path / "AGENTS.md"
+        context_file.write_text(
+            "<!-- headroom:learn:start -->\n"
+            "## Headroom Learned Patterns\n\n"
+            "### Learned: preference\n"
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->\n"
+            "- Build against the vendored SDK <!-- headroom:pattern-id:vendored -->\n"
+            "- Untagged leftover\n\n"
+            "<!-- headroom:learn:end -->\n"
+        )
+        recommendation = _rec(
+            RecommendationTarget.CONTEXT_FILE,
+            "Learned: preference",
+            "- Keep local reviews <!-- headroom:pattern-id:reviews -->",
+        )
+        recommendation.preserve_prior_items = True
+        assert recommendation.active_item_ids is None
+
+        final = _merge_into_file(context_file, [recommendation])
+
+        assert final.count("Keep local reviews") == 1
+        assert "Build against the vendored SDK" in final
+        assert "Untagged leftover" in final
+
     def test_replacing_existing_block_handles_literal_backslash_escapes(self, tmp_path):
         """LLM text with backslash escapes must not be interpreted as a regex replacement."""
         proj = _project(tmp_path)

tests/test_memory/test_traffic_learner.py

@@ -808,6 +808,8 @@ class TestPatternsToRecommendations:
         assert len(recs) == 1
         assert recs[0].target == RecommendationTarget.MEMORY_FILE
         assert "User prefers terse output" in recs[0].content
+        assert "<!-- headroom:pattern-id:" in recs[0].content
+        assert recs[0].preserve_prior_items is True
 
     def test_routes_environment_to_context_file(self):
         from headroom.learn.models import RecommendationTarget
@@ -846,6 +848,77 @@ class TestPatternsToRecommendations:
         assert lines[0] == "- B"
         assert lines[1] == "- A"
         assert recs[0].evidence_count == 7
+        assert recs[0].preserve_prior_items is False
+
+    def test_active_item_ids_span_every_live_pattern(self):
+        """The lifecycle signal covers the whole live set, not just rendered bullets."""
+        preference = ExtractedPattern(
+            category=PatternCategory.PREFERENCE,
+            content="User prefers terse output",
+            importance=0.8,
+            evidence_count=3,
+        )
+        environment = ExtractedPattern(
+            category=PatternCategory.ENVIRONMENT,
+            content="Use uv run python",
+            importance=0.7,
+            evidence_count=4,
+        )
+
+        recs = _patterns_to_recommendations([preference, environment])
+
+        assert len(recs) == 2
+        expected = frozenset({preference.content_hash, environment.content_hash})
+        for rec in recs:
+            # Each section renders one bullet but claims both ids as active, so
+            # a prior item this batch left out is not read as expired.
+            assert len(rec.content.splitlines()) == 1
+            assert rec.active_item_ids == expected
+
+    def test_error_recovery_carries_no_active_item_ids(self):
+        """error_recovery replaces its section, so it exposes no preservation signal."""
+        recs = _patterns_to_recommendations(
+            [
+                ExtractedPattern(
+                    category=PatternCategory.ERROR_RECOVERY,
+                    content="A",
+                    importance=0.5,
+                    evidence_count=2,
+                ),
+            ]
+        )
+
+        assert len(recs) == 1
+        assert recs[0].active_item_ids is None
+
+    def test_pattern_dropped_from_live_set_is_removed_from_the_file(self, tmp_path):
+        """End-to-end removal invariant: an expired pattern leaves the memory file."""
+        from headroom.learn.writer import _merge_into_file
+
+        keep = ExtractedPattern(
+            category=PatternCategory.PREFERENCE,
+            content="User prefers terse output",
+            importance=0.8,
+            evidence_count=3,
+        )
+        expired = ExtractedPattern(
+            category=PatternCategory.PREFERENCE,
+            content="User prefers the legacy migration script",
+            importance=0.8,
+            evidence_count=3,
+        )
+        memory_file = tmp_path / "MEMORY.md"
+        memory_file.write_text(
+            _merge_into_file(memory_file, _patterns_to_recommendations([keep, expired])),
+            encoding="utf-8",
+        )
+        assert "User prefers the legacy migration script" in memory_file.read_text()
+
+        # Next render: the learner no longer holds the expired pattern.
+        final = _merge_into_file(memory_file, _patterns_to_recommendations([keep]))
+
+        assert "User prefers terse output" in final
+        assert "User prefers the legacy migration script" not in final
 
 
 # =============================================================================