fix(compression): mark compressed outputs with _compacted metadata to prevent re-processing

Set metadata._compacted=True on messages whose content was modified by
compression, so the already-processed fast-path recognizes them on
subsequent requests. This ensures each tool output is only logged once
(when first compressed) rather than on every request in the session.

Author: Mateusz

src/core/services/tool_output_compression_service.py

@@ -566,8 +566,12 @@ async def compress_messages(
                 )
                 continue
 
+            updated_metadata = dict(message.metadata) if message.metadata else {}
+            updated_metadata["_compacted"] = True
             updated_messages.append(
-                message.model_copy(update={"content": final_content})
+                message.model_copy(
+                    update={"content": final_content, "metadata": updated_metadata}
+                )
             )
             self._log_output_evaluation(
                 record=record,

tests/unit/core/services/test_tool_output_compression_service.py

@@ -402,6 +402,42 @@ async def test_service_skips_outputs_marked_compacted_in_metadata() -> None:
     assert "skipped_already_processed_compaction" in result.records[0].warnings
 
 
+@pytest.mark.asyncio
+async def test_service_sets_compacted_metadata_on_first_compression() -> None:
+    registry = CompressionStrategyRegistry()
+    registry.register("half_trim", _HalfTrimStrategy())
+    service = ToolOutputCompressionService(
+        strategy_registry=registry,
+        identity_resolver=ToolIdentityResolver(),
+        selector=RuleBasedStrategySelector(),
+    )
+    messages = _build_tool_messages("git status", "x" * 200)
+    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"],
+            )
+        ],
+    )
+
+    first = await service.compress_messages(messages=messages, config=cfg)
+
+    assert first.records[0].applied is True
+    assert first.messages[1].metadata == {"_compacted": True}
+
+    second = await service.compress_messages(messages=first.messages, config=cfg)
+
+    assert second.records[0].applied is False
+    assert "skipped_already_processed_compaction" in second.records[0].warnings
+
+
 @pytest.mark.asyncio
 async def test_service_skips_artifact_preview_system_reminder_outputs() -> None:
     registry = CompressionStrategyRegistry()