fix(compression): detect previously compressed outputs via [COMPRESSED] content marker

The _compacted metadata flag only survives within a single request pass
since clients don't echo back ChatMessage.metadata. Outputs dynamically
compressed in prior requests now carry a [COMPRESSED...] prefix in their
content, so checking for it in _already_processed_skip_warning ensures
cross-request detection without any memory overhead.

Author: Mateusz

src/core/services/tool_output_compression_service.py

@@ -6,6 +6,7 @@
 import hashlib
 import json
 import logging
+import re
 import time
 from collections.abc import Sequence
 
@@ -65,6 +66,7 @@
 _TIME_BUDGET_EXCEEDED_REASON = "time_budget_exceeded"
 _DYNAMIC_CONFIG_RUNTIME_TUNABLE_ATTR = "__dynamic_config_runtime_tunable__"
 _COMPACTED_STUB_MARKER = "[COMPACTED]"
+_COMPRESSED_MARKER_RE = re.compile(r"^\[COMPRESSED[^\]]*\]", re.MULTILINE)
 _SYSTEM_REMINDER_MARKER = "<system-reminder>"
 _NOISY_NOOP_DECISION_REASONS = frozenset(
     {
@@ -813,6 +815,8 @@ def _already_processed_skip_warning(message: ChatMessage) -> str | None:
             return None
         if _COMPACTED_STUB_MARKER in message.content:
             return "skipped_already_processed_compaction"
+        if _COMPRESSED_MARKER_RE.match(message.content):
+            return "skipped_already_processed_compression"
         if (
             _SYSTEM_REMINDER_MARKER in message.content
             and "artifact" in message.content.lower()

tests/unit/core/services/test_tool_output_compression_service.py

@@ -438,6 +438,42 @@ async def test_service_sets_compacted_metadata_on_first_compression() -> None:
     assert "skipped_already_processed_compaction" in second.records[0].warnings
 
 
+@pytest.mark.asyncio
+async def test_service_skips_outputs_with_compressed_marker_already_processed() -> None:
+    registry = CompressionStrategyRegistry()
+    registry.register("half_trim", _HalfTrimStrategy())
+    service = ToolOutputCompressionService(
+        strategy_registry=registry,
+        identity_resolver=ToolIdentityResolver(),
+        selector=RuleBasedStrategySelector(),
+    )
+    previously_compressed = (
+        "[COMPRESSED level=balanced methods=ansi_normalize,diff_compact saved=2048B]\n"
+        "diff --git a/foo.py b/foo.py\n--- a/foo.py\n+++ b/foo.py\n"
+    )
+    messages = _build_tool_messages("git diff", previously_compressed)
+    cfg = DynamicCompressionConfig(
+        enabled=True,
+        min_bytes=0,
+        marker=CompressionMarkerConfig(enabled=False),
+        methods={"half_trim": True},
+        rules=[
+            CompressionRule(
+                name="default",
+                priority=1,
+                when=CompressionRulePredicate(command_signature="git"),
+                pipeline=["half_trim"],
+            )
+        ],
+    )
+
+    result = await service.compress_messages(messages=messages, config=cfg)
+
+    assert result.messages[1].content == previously_compressed
+    assert result.records[0].applied is False
+    assert "skipped_already_processed_compression" in result.records[0].warnings
+
+
 @pytest.mark.asyncio
 async def test_service_skips_artifact_preview_system_reminder_outputs() -> None:
     registry = CompressionStrategyRegistry()