Python: Add agent-triggered in-run compaction

python/packages/core/AGENTS.md

@@ -165,6 +165,21 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
   Shared serialization normalizes stored data; codecs and connector overrides remain trusted Python code
 - **`SupportsVectorUpsert`** / **`SupportsVectorSearch`** - Structural protocols for vector store capabilities
 
+### Compaction (`_compaction.py`)
+
+- **`AgentTriggeredCompactionStrategy`** - Wraps one existing compaction strategy and invokes it only after its
+  explicitly registered trigger tool completes in the latest message group. Trigger consumption is stored on that
+  group rather than on the strategy instance so shared agents remain safe across concurrent runs. The trigger group
+  is temporarily excluded while the inner strategy runs, so it does not consume retention counts, then restored by
+  default; set `exclude_trigger=True` to remove a trigger-only group from the next model call. Mixed groups that
+  contain unrelated tool calls are always restored atomically. The agent-triggered compaction APIs are experimental
+  under the shared `AGENT_TRIGGERED_COMPACTION` feature ID and require client-managed history (`store=False` with no
+  service-managed conversation ID).
+- **`UnloadSkillsCompactionStrategy`** - Intended as an inner agent-triggered strategy. It supplies an
+  `unload_skill` trigger-tool factory that `AgentTriggeredCompactionStrategy` adopts automatically, then removes
+  complete `load_skill` and `read_skill_resource` groups for one named skill, or all skills when no name is
+  supplied. Mixed tool groups and `run_skill_script` results remain intact.
+
 ### Middleware (`_middleware.py`)
 
 - **`AgentMiddleware`** - Intercepts agent `run()` calls

python/packages/core/agent_framework/__init__.py

@@ -69,6 +69,7 @@
         "SUMMARIZED_BY_SUMMARY_ID_KEY",
         "SUMMARY_OF_GROUP_IDS_KEY",
         "SUMMARY_OF_MESSAGE_IDS_KEY",
+        "AgentTriggeredCompactionStrategy",
         "CharacterEstimatorTokenizer",
         "CompactionProvider",
         "CompactionStrategy",
@@ -80,6 +81,7 @@
         "TokenizerProtocol",
         "ToolResultCompactionStrategy",
         "TruncationStrategy",
+        "UnloadSkillsCompactionStrategy",
         "annotate_message_groups",
         "apply_compaction",
         "included_messages",
@@ -452,6 +454,7 @@
     "AgentResponseUpdate",
     "AgentRunInputs",
     "AgentSession",
+    "AgentTriggeredCompactionStrategy",
     "AggregatingSkillsSource",
     "Annotation",
     "BackgroundAgentsProvider",
@@ -645,6 +648,7 @@
     "ToolTypes",
     "TruncationStrategy",
     "TypeCompatibilityError",
+    "UnloadSkillsCompactionStrategy",
     "UpdateT",
     "UsageDetails",
     "UserInputRequiredException",

python/packages/core/agent_framework/__init__.pyi

@@ -31,6 +31,7 @@ from ._compaction import (
     SUMMARIZED_BY_SUMMARY_ID_KEY,
     SUMMARY_OF_GROUP_IDS_KEY,
     SUMMARY_OF_MESSAGE_IDS_KEY,
+    AgentTriggeredCompactionStrategy,
     CharacterEstimatorTokenizer,
     CompactionProvider,
     CompactionStrategy,
@@ -42,6 +43,7 @@ from ._compaction import (
     TokenizerProtocol,
     ToolResultCompactionStrategy,
     TruncationStrategy,
+    UnloadSkillsCompactionStrategy,
     annotate_message_groups,
     apply_compaction,
     included_messages,
@@ -407,6 +409,7 @@ __all__ = [
     "AgentResponseUpdate",
     "AgentRunInputs",
     "AgentSession",
+    "AgentTriggeredCompactionStrategy",
     "AggregatingSkillsSource",
     "Annotation",
     "BackgroundAgentsProvider",
@@ -600,6 +603,7 @@ __all__ = [
     "ToolTypes",
     "TruncationStrategy",
     "TypeCompatibilityError",
+    "UnloadSkillsCompactionStrategy",
     "UpdateT",
     "UsageDetails",
     "UserInputRequiredException",

python/packages/core/agent_framework/_clients.py

@@ -38,6 +38,7 @@
     ResponseStream,
     validate_chat_options,
 )
+from .exceptions import ChatClientInvalidRequestException
 
 if sys.version_info >= (3, 13):
     from typing import TypeVar  # pragma: no cover
@@ -373,8 +374,27 @@ async def _prepare_messages_for_model_call(
         *,
         compaction_strategy: CompactionStrategy | None = None,
         tokenizer: TokenizerProtocol | None = None,
+        options: Mapping[str, Any] | None = None,
     ) -> list[Message]:
         prepared_messages = list(messages)
+        if compaction_strategy is not None:
+            from ._compaction import _contains_agent_triggered_compaction_strategy  # pyright: ignore[reportPrivateUsage]
+            from ._sessions import is_local_history_conversation_id
+
+            if _contains_agent_triggered_compaction_strategy(compaction_strategy):
+                store = options.get("store") if options is not None else None
+                service_stores_history = (
+                    getattr(self, "STORES_BY_DEFAULT", False) if store is None else store is not False
+                )
+                conversation_id = options.get("conversation_id") if options is not None else None
+                has_service_conversation = conversation_id is not None and (
+                    not isinstance(conversation_id, str) or not is_local_history_conversation_id(conversation_id)
+                )
+                if service_stores_history or has_service_conversation:
+                    raise ChatClientInvalidRequestException(
+                        "Agent-triggered compaction requires client-managed history. "
+                        "Set store=False and do not use a service-managed conversation_id."
+                    )
         if compaction_strategy is None:
             if tokenizer is None:
                 return prepared_messages
@@ -531,6 +551,7 @@ async def _get_stream() -> ResponseStream[ChatResponseUpdate, ChatResponse[Any]]
                 prepared_messages = await self._prepare_messages_for_model_call(
                     messages,
                     **compaction_overrides,
+                    options=options,
                 )
                 stream_response = self._inner_get_response(
                     messages=prepared_messages,
@@ -551,6 +572,7 @@ async def _get_response() -> ChatResponse[Any]:
             prepared_messages = await self._prepare_messages_for_model_call(
                 messages,
                 **compaction_overrides,
+                options=options,
             )
             return await self._inner_get_response(
                 messages=prepared_messages,