Fit over-budget views without weakening required content protection

README.md

@@ -106,10 +106,17 @@ Compaction triggers when token usage reaches the configured threshold (default:
 - **System messages**: All system messages are always preserved
 - **First human prompt**: The original human task/request is protected, using message metadata rather than treating every user-role message as human input
 - **Last human prompt**: The most recent human input is protected by the same metadata-based classification
-- **Recent messages**: Last N% of messages (configurable via `protected_recent`)
 - **Recent tool results**: The last `protected_tool_results` results (default 5) are protected from both truncation and removal; a protected sibling also prevents removal of its owning call group
 - **Tool pairs**: Tool_use and tool_result messages are treated as atomic units
 
+Recent history also receives a soft preference (`protected_recent`). The normal
+compaction ladder reduces this fraction progressively. If its final view still
+exceeds the effective input budget, one final pass relaxes the fraction to zero
+and targets the budget, not the lower compaction target. Required request
+injections, human boundaries, loaded-tool state, and protected tool-result
+groups keep their protections. A required floor that cannot fit still fails;
+canonical history is never rewritten by this pass.
+
 ### Request-scoped retention
 
 The optional `context.request_retention` capability lets an orchestrator name

amplifier_module_context_simple/__init__.py

@@ -2557,6 +2557,25 @@ async def _compact_ephemeral(
                 f"Tokens: {old_tokens:,} → {current_tokens:,}"
             )
 
+        # Recency is a quality preference, not an irreducible requirement.
+        # A small recent slice can contain several large historical machine
+        # reminders. After the normal ladder, relax only that soft preference
+        # if necessary to fit the actual budget (not merely the lower target).
+        # The shared remover still protects current retained injections, human
+        # boundaries, loaded-tool state, and the last tool-result groups.
+        if self._estimate_tokens(working_messages) + system_tokens > budget:
+            working_messages, removed, stubbed, current_tokens = (
+                self._remove_messages_with_protection(
+                    working_messages,
+                    budget,
+                    protected_recent=0,
+                    system_tokens=system_tokens,
+                    additional_protected_seqs=additional_protected_seqs,
+                )
+            )
+            total_removed += removed
+            total_stubbed += stubbed
+
         return await self._finalize_compaction_with_stats(
             working_messages,
             system_messages,

tests/test_soft_recent_budget.py

@@ -0,0 +1,70 @@
+"""Historical reminder density must not turn soft recency into a hard floor."""
+
+import copy
+
+import pytest
+from amplifier_core.llm_errors import ContextLengthError
+
+from amplifier_module_context_simple import SimpleContextManager
+
+
+async def dense_history(*, tool_size=100, required_size=1000, tool_count=5):
+    context = SimpleContextManager(max_tokens=10000, compaction_notice_enabled=False)
+    await context.add_message({'role': 'system', 'content': 'Required system instructions.'})
+    await context.add_message({'role': 'user', 'content': 'Original human requirements.'})
+    for i in range(100):
+        await context.add_message({
+            'role': 'user', 'content': f'Historical reminder {i}: ' + 'r' * 16000,
+            'metadata': {'ephemeral': True, 'persisted': True},
+        })
+        await context.add_message({'role': 'assistant', 'content': 'Earlier work ' + 'a' * 400})
+    # An older tool-state carrier must survive even after soft recency relaxes.
+    await context.add_message({
+        'role': 'assistant', 'content': 'Loaded tools must remain available.',
+        'metadata': {'openai:tool_search_items': [{'id': 'loaded-read-file'}]},
+    })
+    for i in range(tool_count):
+        await context.add_message({
+            'role': 'assistant', 'content': '',
+            'tool_calls': [{'id': f'call-{i}', 'name': 'read_file', 'arguments': {}}],
+        })
+        await context.add_message({
+            'role': 'tool', 'tool_call_id': f'call-{i}', 'content': 't' * tool_size,
+        })
+    required = 'Current required reminder: ' + 'c' * required_size
+    await context.add_message({
+        'role': 'user', 'content': required,
+        'metadata': {'ephemeral': True, 'persisted': True},
+    })
+    await context.add_message({'role': 'user', 'content': 'Latest human correction.'})
+    return context, required
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('tool_count', [0, 5])
+async def test_dense_historical_reminders_fit_without_losing_requirements(tool_count):
+    context, required = await dense_history(tool_count=tool_count)
+    canonical = copy.deepcopy(await context.get_messages())
+    view = await context.get_messages_for_request_retaining(retain_contents=[required])
+    assert context._estimate_tokens(view) <= 10000
+    for body in ('Required system instructions.', 'Original human requirements.',
+                 'Latest human correction.', required, 'Loaded tools must remain available.'):
+        assert any(message.get('content') == body for message in view)
+    assert [m['tool_call_id'] for m in view if m['role'] == 'tool'] == [f'call-{i}' for i in range(tool_count)]
+    assert all(m['content'] == 't' * 100 for m in view if m['role'] == 'tool')
+    assert {c['id'] for m in view for c in m.get('tool_calls', [])} == {f'call-{i}' for i in range(tool_count)}
+    assert await context.get_messages() == canonical
+    assert await context.get_messages_for_request_retaining(retain_contents=[required]) == view
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('tool_size,required_size', [(10000, 1000), (100, 44000)])
+async def test_required_floor_still_fails_without_history_or_decision_changes(tool_size, required_size):
+    context, required = await dense_history(tool_size=tool_size, required_size=required_size)
+    canonical = copy.deepcopy(await context.get_messages())
+    with pytest.raises(ContextLengthError):
+        await context.get_messages_for_request_retaining(retain_contents=[required])
+    assert await context.get_messages() == canonical
+    assert not context._removed_seqs
+    assert not context._truncated_seqs
+    assert not context._stubbed_seqs