Python: Add agent-triggered in-run compaction

python/packages/core/AGENTS.md

@@ -171,6 +171,25 @@ 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 the
+  completed trigger call occurrence rather than on the strategy instance so shared agents remain safe across
+  concurrent runs and composed trigger tools do not suppress each other. 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). Strategy wrappers and compositions must propagate the optional
+  `requires_client_managed_history` capability; `TokenBudgetComposedStrategy` does this for its nested strategies.
+- **`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. Loaded skill identity comes from successful result provenance recorded by `SkillsProvider` after
+  resolving the effective invocation, not the original model arguments. Groups with missing/invalid provenance,
+  unresolved calls, mixed tools, or `run_skill_script` results remain intact.
+
 ### Middleware (`_middleware.py`)
 
 - **`AgentMiddleware`** - Intercepts agent `run()` calls

python/packages/core/agent_framework/__init__.py

@@ -73,6 +73,7 @@
         "SUMMARIZED_BY_SUMMARY_ID_KEY",
         "SUMMARY_OF_GROUP_IDS_KEY",
         "SUMMARY_OF_MESSAGE_IDS_KEY",
+        "AgentTriggeredCompactionStrategy",
         "CharacterEstimatorTokenizer",
         "CompactionProvider",
         "CompactionStrategy",
@@ -84,6 +85,7 @@
         "TokenizerProtocol",
         "ToolResultCompactionStrategy",
         "TruncationStrategy",
+        "UnloadSkillsCompactionStrategy",
         "annotate_message_groups",
         "apply_compaction",
         "included_messages",
@@ -456,6 +458,7 @@
     "AgentResponseUpdate",
     "AgentRunInputs",
     "AgentSession",
+    "AgentTriggeredCompactionStrategy",
     "AggregatingSkillsSource",
     "Annotation",
     "BackgroundAgentsProvider",
@@ -649,6 +652,7 @@
     "ToolTypes",
     "TruncationStrategy",
     "TypeCompatibilityError",
+    "UnloadSkillsCompactionStrategy",
     "UpdateT",
     "UsageDetails",
     "UserInputRequiredException",

python/packages/core/agent_framework/__init__.pyi

@@ -35,6 +35,7 @@ from ._compaction import (
     SUMMARIZED_BY_SUMMARY_ID_KEY,
     SUMMARY_OF_GROUP_IDS_KEY,
     SUMMARY_OF_MESSAGE_IDS_KEY,
+    AgentTriggeredCompactionStrategy,
     CharacterEstimatorTokenizer,
     CompactionProvider,
     CompactionStrategy,
@@ -46,6 +47,7 @@ from ._compaction import (
     TokenizerProtocol,
     ToolResultCompactionStrategy,
     TruncationStrategy,
+    UnloadSkillsCompactionStrategy,
     annotate_message_groups,
     apply_compaction,
     included_messages,
@@ -411,6 +413,7 @@ __all__ = [
     "AgentResponseUpdate",
     "AgentRunInputs",
     "AgentSession",
+    "AgentTriggeredCompactionStrategy",
     "AggregatingSkillsSource",
     "Annotation",
     "BackgroundAgentsProvider",
@@ -604,6 +607,7 @@ __all__ = [
     "ToolTypes",
     "TruncationStrategy",
     "TypeCompatibilityError",
+    "UnloadSkillsCompactionStrategy",
     "UpdateT",
     "UsageDetails",
     "UserInputRequiredException",

python/packages/core/agent_framework/_clients.py

@@ -40,6 +40,7 @@
     ResponseStream,
     validate_chat_options,
 )
+from .exceptions import ChatClientInvalidRequestException
 
 if sys.version_info >= (3, 13):
     from typing import TypeVar  # pragma: no cover
@@ -295,6 +296,8 @@ async def _stream():
     When ``True``, the agent skips auto-injecting ``InMemoryHistoryProvider`` unless the
     user explicitly sets ``store=False``.
     """
+    SERVICE_CONTINUATION_OPTION_NAMES: ClassVar[tuple[str, ...]] = ("conversation_id",)
+    """Option names that activate provider-managed conversation continuation."""
     # OTEL_PROVIDER_NAME is used for OTel setup, should be overridden in subclasses
 
     def __init__(
@@ -406,8 +409,37 @@ 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 _requires_client_managed_history  # pyright: ignore[reportPrivateUsage]
+            from ._sessions import is_local_history_conversation_id
+
+            if _requires_client_managed_history(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
+                )
+                has_service_conversation = False
+                if options is not None:
+                    for option_name in self.SERVICE_CONTINUATION_OPTION_NAMES:
+                        value = options.get(option_name)
+                        if value is None or value == "":
+                            continue
+                        if (
+                            option_name == "conversation_id"
+                            and isinstance(value, str)
+                            and is_local_history_conversation_id(value)
+                        ):
+                            continue
+                        has_service_conversation = True
+                        break
+                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 continuation option."
+                    )
         if compaction_strategy is None:
             if tokenizer is None:
                 return prepared_messages
@@ -564,6 +596,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,
@@ -584,6 +617,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,