diff --git a/examples/10-agentscope-compat-single-agent/DESIGN.md b/examples/10-agentscope-compat-single-agent/DESIGN.md
index b292693e..47af4126 100644
--- a/examples/10-agentscope-compat-single-agent/DESIGN.md
+++ b/examples/10-agentscope-compat-single-agent/DESIGN.md
@@ -48,7 +48,7 @@ PlanNoteBook / SubTask / TruncatedFormatterBase / Knowledge / HttpStatefulClient
 - Memory: `dare_framework/memory/in_memory_stm.py`
 - Knowledge: `dare_framework/knowledge/kernel.py`
 - Plan: `dare_framework/plan_v2/types.py`, `dare_framework/plan_v2/tools.py`
-- Compression: `dare_framework/compression/core.py`
+- Compression: `dare_framework/compression/moving_compression.py`
 - MCP: `dare_framework/mcp/client.py`, `dare_framework/mcp/transports/http.py`
 
 ## 4. 能力差异矩阵（详细版）
@@ -60,7 +60,7 @@ PlanNoteBook / SubTask / TruncatedFormatterBase / Knowledge / HttpStatefulClient
 | 循环结构 | `_reasoning() → _acting() → repeat` | `assemble() → generate() → tool calls → repeat` | 等价 |
 | 最大迭代 | `max_iterations=20` | `max_tool_rounds=10` | 等价（值不同） |
 | 并行 tool 执行 | `parallel_tool_calls=True` → `asyncio.gather` | 仅串行 | **Gap-R1** |
-| 自动内存压缩 | `_compress_memory_if_needed()` 每轮触发 | 无自动压缩 | **Gap-R5** |
+| 自动内存压缩 | `_compress_memory_if_needed()` 每轮触发 | 仅有 moving compression，未接入 ReAct 自动触发 | **Gap-R5** |
 | 超时 fallback | 超 max_iterations 做 summarization | 返回"未收敛"文本 | Gap-R2 |
 | Plan 注入 | `plan_to_hint()` 生成 `<system-hint>` | `critical_block` 注入 | 接近等价 |
 | Hook 粒度 | pre/post_reasoning, pre/post_acting | session/milestone/plan/tool 级 | Gap-R4 |
@@ -162,8 +162,8 @@ PlanNoteBook / SubTask / TruncatedFormatterBase / Knowledge / HttpStatefulClient
 | 维度 | AgentScope | DARE | 差距 |
 |------|-----------|------|------|
 | 截断单位 | Token 数 | 消息条数 | **Gap-F2** |
-| Tool pair 安全 | 成对删除 | 无保护 | **Gap-F1** |
-| 自动触发 | 每次 `_reasoning()` 前 | 手动调用 | **Gap-F4** |
+| Tool pair 安全 | 成对删除 | framework 无 formatter 级保护，由 Example 兼容层补齐 | **Gap-F1** |
+| 自动触发 | 每次 `_reasoning()` 前 | moving compression 需显式接线，无 formatter 自动触发 | **Gap-F4** |
 | Provider 格式化 | OpenAI/Anthropic/Gemini/... formatter 子类 | 无 | Gap-F3 |
 | 标签感知 | 跳过 important 消息 | 无标签概念 | 依赖 Gap-M1 |
 
@@ -218,7 +218,7 @@ PlanNoteBook / SubTask / TruncatedFormatterBase / Knowledge / HttpStatefulClient
 ### P1（高优先）
 - **Gap-M1**: Message 无 tag/mark
 - **Gap-Mem1/2/5**: InMemorySTM 无 mark/summary/tool-pair-safe compress
-- **Gap-F1/F4**: compress_context 无 tool pair 安全/无自动触发
+- **Gap-F1/F4**: framework 仅提供 moving compression；无 formatter 级 tool pair 安全/无自动触发
 - **Gap-R5**: ReactAgent 无自动内存压缩
 - **Gap-LM4**: Usage 不规范化 reasoning_tokens
 - **Gap-S1/S2**: 无 StateModule/ISessionStore
diff --git a/examples/10-agentscope-compat-single-agent/README.md b/examples/10-agentscope-compat-single-agent/README.md
index ae901d8d..0bfb98f4 100644
--- a/examples/10-agentscope-compat-single-agent/README.md
+++ b/examples/10-agentscope-compat-single-agent/README.md
@@ -14,7 +14,7 @@
 | 6 | `ChatModelBase` | `CompatFormattedModelAdapter` | E0/E2 | **Gap-LM1(P0)**(thinking), Gap-LM2(stream) |
 | 7 | `PlanNoteBook` | `CompatPlanNotebook` + 6 tools | E1 | Gap-P1(status), Gap-P5(序列化) |
 | 8 | `SubTask` | `CompatSubTask` | E1 | 合并于 Gap-P1 |
-| 9 | `TruncatedFormatterBase` | `CompatTruncatedFormatter` | E1 | **Gap-F1**(tool pair safe), Gap-F2(token) |
+| 9 | `TruncatedFormatterBase` | `CompatTruncatedFormatter` | E1 | **Gap-F1**(example 层补齐 tool pair safe), Gap-F2(token) |
 | 10 | `Knowledge` | `create_knowledge(rawdata)` | E0 | Gap-K1(embedding adapter) |
 | 11 | `HttpStatefulClient` | `HttpStatefulClientShim` | E1 | Gap-H3(缓存) |
 | 12 | `Session` | `JsonSessionBridge` | E2 | **Gap-S1**(StateModule), **Gap-S2**(ISessionStore) |
diff --git a/examples/10-agentscope-compat-single-agent/compat_agent.py b/examples/10-agentscope-compat-single-agent/compat_agent.py
index 0ad6bbee..6e1825b0 100644
--- a/examples/10-agentscope-compat-single-agent/compat_agent.py
+++ b/examples/10-agentscope-compat-single-agent/compat_agent.py
@@ -221,7 +221,7 @@ def to_framework_message(self) -> Message:
 # ===========================================================================
 # Capability 9: TruncatedFormatterBase — 截断格式化器
 # AgentScope: token 级截断 + tool pair 安全 + provider-specific 格式化
-# DARE:       compress_context() 按消息条数截断，无 tool pair 安全 [Gap-F1]
+# DARE:       仅提供 moving compression；formatter 级 tool pair 安全需 Example 补齐 [Gap-F1]
 # ===========================================================================
 
 
@@ -290,9 +290,10 @@ def _drop_with_tool_pair(
     ) -> int:
         """删除消息时保护 tool call/result 配对完整性。
 
-        这是 Gap-F1 的 Example 层补齐：框架的 compress_context() 不具备此能力。
-        当框架补齐 Gap-F1 (compress_context(tool_pair_safe=True)) 后，
-        此方法应迁移到框架层。
+        这是 Gap-F1 的 Example 层补齐：当前框架只有 moving compression，
+        并没有 formatter 级 tool pair 安全截断入口。
+        如果后续框架在公开 formatter/compression API 中补齐这层能力，
+        此方法再考虑下沉到框架层。
         """
         removed_count = 0
         removed_tool_ids: set[str] = set()
diff --git a/tests/unit/test_base_agent_transport_contract.py b/tests/unit/test_base_agent_transport_contract.py
index a3bceb9f..a51e4407 100644
--- a/tests/unit/test_base_agent_transport_contract.py
+++ b/tests/unit/test_base_agent_transport_contract.py
@@ -431,14 +431,14 @@ async def test_transport_loop_flag_is_task_local_for_concurrent_execute_calls()
 
     polled_task = asyncio.create_task(
         agent._execute_polled_message(
-            "loop-task",
+            Message(role="user", text="loop-task"),
             channel=channel,
             envelope_id="req_1",
         )
     )
     await agent.loop_execution_started.wait()
 
-    await agent.execute("direct-task", transport=channel)
+    await agent.execute(Message(role="user", text="direct-task"), transport=channel)
     agent.allow_loop_execution_finish.set()
     await polled_task
 
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
deleted file mode 100644
index 1576fdad..00000000
--- a/tests/unit/test_context_compression.py
+++ /dev/null
@@ -1,251 +0,0 @@
-from __future__ import annotations
-
-from dare_framework.compression.core import compress_context
-from dare_framework.config import Config
-from dare_framework.context import AttachmentKind, AttachmentRef, Context, Message, MessageKind, MessageMark
-
-
-def _tool_ids(message: Message) -> list[str]:
-    raw_calls = []
-    if isinstance(message.data, dict):
-        raw_calls = message.data.get("tool_calls", [])
-    if not isinstance(raw_calls, list):
-        return []
-    ids: list[str] = []
-    for item in raw_calls:
-        if not isinstance(item, dict):
-            continue
-        tool_id = item.get("id")
-        if isinstance(tool_id, str) and tool_id:
-            ids.append(tool_id)
-    return ids
-
-
-def test_compress_context_tool_pair_safe_removes_orphan_tool_result() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="tool call",
-            data={"tool_calls": [{"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}}]},
-        )
-    )
-    ctx.stm_add(Message(role="tool", name="tc_1", text='{"success": true}'))
-    ctx.stm_add(Message(role="tool", name="tc_orphan", text='{"success": true}'))
-
-    compress_context(ctx, strategy="truncate", max_messages=10, tool_pair_safe=True)
-
-    messages = ctx.stm_get()
-    tool_names = [message.name for message in messages if message.role == "tool"]
-    assert "tc_1" in tool_names
-    assert "tc_orphan" not in tool_names
-
-
-def test_compress_context_tool_pair_safe_removes_unmatched_tool_call_ids() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="tool call",
-            data={
-                "tool_calls": [
-                    {"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}},
-                    {"id": "tc_2", "name": "missing_tool", "arguments": {"x": 2}},
-                ]
-            },
-        )
-    )
-    ctx.stm_add(Message(role="tool", name="tc_1", text='{"success": true}'))
-
-    compress_context(ctx, strategy="truncate", max_messages=10, tool_pair_safe=True)
-
-    assistant_message = next(message for message in ctx.stm_get() if message.role == "assistant")
-    assert _tool_ids(assistant_message) == ["tc_1"]
-    assert assistant_message.data == {
-        "tool_calls": [{"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}}]
-    }
-
-
-def test_compress_context_tool_pair_safe_keeps_idless_tool_context() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="tool call without id",
-            data={"tool_calls": [{"name": "demo_tool", "arguments": {"x": 1}}]},
-        )
-    )
-    ctx.stm_add(Message(role="tool", name="demo_tool", text='{"success": true}'))
-
-    compress_context(ctx, strategy="truncate", max_messages=10, tool_pair_safe=True)
-
-    messages = ctx.stm_get()
-    assistant_message = next(message for message in messages if message.role == "assistant")
-    raw_calls = assistant_message.data.get("tool_calls", []) if isinstance(assistant_message.data, dict) else []
-    assert isinstance(raw_calls, list)
-    assert len(raw_calls) == 1
-    assert any(message.role == "tool" and message.name == "demo_tool" for message in messages)
-
-
-def test_compress_context_tool_pair_safe_drops_orphan_tool_results_with_mixed_id_modes() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="mixed tool calls",
-            data={
-                "tool_calls": [
-                    {"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}},
-                    {"name": "demo_tool", "arguments": {"x": 2}},
-                ]
-            },
-        )
-    )
-    ctx.stm_add(Message(role="tool", name="tc_1", text='{"success": true}'))
-    ctx.stm_add(Message(role="tool", name="demo_tool", text='{"success": true}'))
-    ctx.stm_add(Message(role="tool", name="tc_orphan", text='{"success": true}'))
-
-    compress_context(ctx, strategy="truncate", max_messages=10, tool_pair_safe=True)
-
-    tool_names = [message.name for message in ctx.stm_get() if message.role == "tool"]
-    assert "tc_1" in tool_names
-    assert "demo_tool" in tool_names
-    assert "tc_orphan" not in tool_names
-
-
-def test_compress_context_tool_pair_safe_preserves_assistant_id_and_mark_when_filtering_calls() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="mixed tool calls",
-            id="assistant-state",
-            mark=MessageMark.PERSISTENT,
-            data={
-                "tool_calls": [
-                    {"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}},
-                    {"id": "tc_missing", "name": "demo_tool", "arguments": {"x": 2}},
-                ]
-            },
-        )
-    )
-    ctx.stm_add(Message(role="tool", name="tc_1", text='{"success": true}'))
-
-    compress_context(ctx, strategy="truncate", max_messages=10, tool_pair_safe=True)
-
-    assistant_message = next(message for message in ctx.stm_get() if message.role == "assistant")
-    assert assistant_message.id == "assistant-state"
-    assert assistant_message.mark == MessageMark.PERSISTENT
-    assert _tool_ids(assistant_message) == ["tc_1"]
-    assert assistant_message.data == {
-        "tool_calls": [{"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}}]
-    }
-
-
-def test_compress_context_target_tokens_trims_long_history() -> None:
-    ctx = Context(config=Config())
-    for idx in range(8):
-        ctx.stm_add(Message(role="user", text=f"long-message-{idx}-" + "x" * 120))
-
-    before_count = len(ctx.stm_get())
-    compress_context(ctx, strategy="truncate", max_messages=8, target_tokens=80)
-    after_messages = ctx.stm_get()
-
-    assert len(after_messages) < before_count
-    assert len(after_messages) >= 1
-
-
-def test_compress_context_negative_max_messages_keeps_unbounded_semantics() -> None:
-    ctx = Context(config=Config())
-    for idx in range(4):
-        ctx.stm_add(Message(role="user", text=f"msg-{idx}"))
-
-    before_messages = list(ctx.stm_get())
-    compress_context(
-        ctx,
-        strategy="truncate",
-        max_messages=-1,
-        target_tokens=10_000,
-    )
-    after_messages = ctx.stm_get()
-
-    assert [message.text for message in after_messages] == [
-        message.text for message in before_messages
-    ]
-
-
-def test_compress_context_annotate_preserves_message_identity_fields() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="keep identity",
-            id="assistant-1",
-            mark=MessageMark.PERSISTENT,
-        )
-    )
-    ctx.stm_add(Message(role="user", text="latest"))
-
-    compress_context(ctx, strategy="dedup_then_truncate", max_messages=1, phase="pre_tool")
-
-    head = ctx.stm_get()[0]
-    assert head.id == "assistant-1"
-    assert head.mark == MessageMark.PERSISTENT
-    assert head.metadata.get("compressed") is True
-    assert head.metadata.get("strategy") == "dedup_then_truncate"
-
-
-def test_compress_context_annotate_preserves_structured_message_fields() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            kind=MessageKind.CHAT,
-            text="keep attachment",
-            attachments=[AttachmentRef(kind=AttachmentKind.IMAGE, uri="https://example.com/a.png")],
-            data={"tool_calls": [{"id": "tc_1"}]},
-            metadata={"trace": "1"},
-            mark=MessageMark.PERSISTENT,
-        )
-    )
-    ctx.stm_add(Message(role="user", text="latest"))
-
-    compress_context(ctx, strategy="dedup_then_truncate", max_messages=1, phase="pre_tool")
-
-    head = ctx.stm_get()[0]
-    assert head.text == "keep attachment"
-    assert len(head.attachments) == 1
-    assert head.attachments[0].uri == "https://example.com/a.png"
-    assert head.data == {"tool_calls": [{"id": "tc_1"}]}
-    assert head.metadata.get("trace") == "1"
-    assert head.metadata.get("compressed") is True
-
-
-def test_compress_context_max_messages_preserves_protected_marks() -> None:
-    ctx = Context(config=Config())
-    ctx.stm_add(
-        Message(
-            role="system",
-            text="immutable",
-            id="imm-1",
-            mark=MessageMark.IMMUTABLE,
-        )
-    )
-    ctx.stm_add(
-        Message(
-            role="assistant",
-            text="persistent",
-            id="persist-1",
-            mark=MessageMark.PERSISTENT,
-        )
-    )
-    for idx in range(4):
-        ctx.stm_add(Message(role="user", text=f"temp-{idx}", id=f"tmp-{idx}"))
-
-    compress_context(ctx, strategy="truncate", max_messages=3, target_tokens=10_000)
-
-    messages = ctx.stm_get()
-    ids = [message.id for message in messages]
-    assert "imm-1" in ids
-    assert "persist-1" in ids
-    assert len(messages) == 3
diff --git a/tests/unit/test_context_implementation.py b/tests/unit/test_context_implementation.py
index ef1b1129..a7d20770 100644
--- a/tests/unit/test_context_implementation.py
+++ b/tests/unit/test_context_implementation.py
@@ -1,5 +1,7 @@
+from __future__ import annotations
 
 import pytest
+
 from dare_framework.config import Config
 from dare_framework.context.context import Context
 from dare_framework.context.types import AttachmentKind, AttachmentRef, Budget, Message
@@ -7,9 +9,11 @@
 from dare_framework.tool._internal.tools.noop_tool import NoopTool
 from dare_framework.tool.tool_manager import ToolManager
 
-def test_context_initialization():
+
+def test_context_initialization() -> None:
     config = Config()
     ctx = Context(id="test-id", config=config)
+
     assert ctx.id == "test-id"
     assert isinstance(ctx.budget, Budget)
     assert ctx.short_term_memory is not None
@@ -18,31 +22,35 @@ def test_context_initialization():
     assert ctx.config is config
     assert ctx.sys_prompt is None
 
-def test_context_stm_methods():
+
+def test_context_stm_methods() -> None:
     ctx = Context(config=Config())
     msg = Message(role="user", text="hello")
     ctx.stm_add(msg)
-    
+
     messages = ctx.stm_get()
     assert len(messages) == 1
     assert messages[0].text == "hello"
-    
+
     ctx.stm_clear()
     assert len(ctx.stm_get()) == 0
 
-def test_context_budget_methods():
+
+def test_context_budget_methods() -> None:
     ctx = Context(config=Config(), budget=Budget(max_tokens=100))
     ctx.budget_use("tokens", 50)
+
     assert ctx.budget.used_tokens == 50
     assert ctx.budget_remaining("tokens") == 50
-    
-    ctx.budget_check() # Should not raise
-    
+
+    ctx.budget_check()
+
     ctx.budget_use("tokens", 60)
     with pytest.raises(RuntimeError, match="Token budget exceeded"):
         ctx.budget_check()
 
-def test_context_assemble():
+
+def test_context_assemble() -> None:
     prompt = Prompt(
         prompt_id="test.system",
         role="system",
@@ -52,8 +60,9 @@ def test_context_assemble():
     )
     ctx = Context(config=Config(), sys_prompt=prompt)
     ctx.stm_add(Message(role="user", text="hi"))
-    
+
     assembled = ctx.assemble()
+
     assert assembled.sys_prompt is not None
     assert assembled.sys_prompt.content == "You are a helpful assistant"
     assert len(assembled.messages) == 1
@@ -85,12 +94,12 @@ def test_context_assemble_preserves_chat_attachments() -> None:
     assert assembled.messages[0].attachments[0].uri == "https://example.com/a.png"
 
 
-def test_context_requires_non_null_config():
+def test_context_requires_non_null_config() -> None:
     with pytest.raises(ValueError, match="non-null Config"):
         Context(id="missing-config", config=None)  # type: ignore[arg-type]
 
 
-def test_context_list_tools_returns_capability_descriptors_from_tool_manager():
+def test_context_list_tools_returns_capability_descriptors_from_tool_manager() -> None:
     manager = ToolManager(load_entrypoints=False)
     manager.register_tool(NoopTool())
     ctx = Context(config=Config(), tool_gateway=manager)
@@ -102,7 +111,7 @@ def test_context_list_tools_returns_capability_descriptors_from_tool_manager():
     assert tools[0].name == "noop"
 
 
-def test_context_exposes_public_tool_gateway_accessor_and_setter():
+def test_context_exposes_public_tool_gateway_accessor_and_setter() -> None:
     manager = ToolManager(load_entrypoints=False)
     ctx = Context(config=Config())
 
@@ -113,15 +122,12 @@ def test_context_exposes_public_tool_gateway_accessor_and_setter():
 
 
 class _FakeRetrieval:
-    def __init__(self, messages: list[Message], *, fail: bool = False) -> None:
+    def __init__(self, messages: list[Message]) -> None:
         self._messages = list(messages)
-        self._fail = fail
         self.calls: list[tuple[str, dict[str, object]]] = []
 
     def get(self, query: str = "", **kwargs: object) -> list[Message]:
         self.calls.append((query, dict(kwargs)))
-        if self._fail:
-            raise RuntimeError("retrieval failed")
         return list(self._messages)
 
     def add(self, message: Message) -> None:
@@ -135,21 +141,7 @@ def compress(self, **kwargs: object) -> int:
         return 0
 
 
-class _CompressionRecordingSTM(_FakeRetrieval):
-    def __init__(self, messages: list[Message]) -> None:
-        super().__init__(messages)
-        self.compress_calls: list[dict[str, object]] = []
-
-    def compress(self, **kwargs: object) -> int:
-        self.compress_calls.append(dict(kwargs))
-        raw_limit = kwargs.get("max_messages")
-        limit = raw_limit if isinstance(raw_limit, int) and raw_limit >= 0 else None
-        if limit is not None and len(self._messages) > limit:
-            self._messages = self._messages[-limit:]
-        return 0
-
-
-def test_context_assemble_fuses_ltm_and_knowledge_with_latest_user_query():
+def test_context_assemble_ignores_optional_retrieval_sources_in_default_strategy() -> None:
     ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")])
     knowledge = _FakeRetrieval([Message(role="assistant", text="knowledge-hit")])
     ctx = Context(
@@ -157,366 +149,69 @@ def test_context_assemble_fuses_ltm_and_knowledge_with_latest_user_query():
         long_term_memory=ltm,
         knowledge=knowledge,
     )
-    ctx.stm_add(Message(role="user", text="old request"))
-    ctx.stm_add(Message(role="assistant", text="ack"))
     ctx.stm_add(Message(role="user", text="latest request"))
 
     assembled = ctx.assemble()
 
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["old request", "ack", "latest request", "ltm-hit", "knowledge-hit"]
-    assert ltm.calls and ltm.calls[0][0] == "latest request"
-    assert knowledge.calls and knowledge.calls[0][0] == "latest request"
-    assert assembled.metadata["retrieval"]["ltm_count"] == 1
-    assert assembled.metadata["retrieval"]["knowledge_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is False
-
-
-def test_context_assemble_degrades_when_token_budget_low():
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")])
-    knowledge = _FakeRetrieval([Message(role="assistant", text="knowledge-hit")])
-    # Force a low remaining token budget so retrieval should be skipped.
-    budget = Budget(max_tokens=32)
-    ctx = Context(
-        config=Config(),
-        budget=budget,
-        long_term_memory=ltm,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="x" * 160))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["x" * 160]
-    assert assembled.metadata["retrieval"]["degraded"] is True
-    assert assembled.metadata["retrieval"]["degrade_reason"] == "token_budget_low"
-
-
-def test_context_assemble_handles_retrieval_exception_gracefully():
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")], fail=True)
-    knowledge = _FakeRetrieval([Message(role="assistant", text="knowledge-hit")])
-    ctx = Context(
-        config=Config(),
-        long_term_memory=ltm,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="query"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["query", "knowledge-hit"]
-    assert assembled.metadata["retrieval"]["degraded"] is True
-    assert assembled.metadata["retrieval"]["degrade_reason"] == "ltm_retrieval_failed"
-
-
-def test_context_assemble_single_source_uses_full_retrieval_budget():
-    ltm = _FakeRetrieval([Message(role="assistant", text="x" * 64)])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 1,
-            "assemble_reserve_tokens": 0,
-            "assemble_ratio": 0.5,
-        },
-        knowledge={
-            "assemble_top_k": 1,
-            "assemble_ratio": 0.5,
-        },
-    )
-    # Remaining retrieval budget ~= 40 tokens after STM estimate.
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=49),
-        long_term_memory=ltm,
-        knowledge=None,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "x" * 64]
-    assert assembled.metadata["retrieval"]["ltm_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is False
-
-
-def test_context_assemble_skips_oversized_retrieval_hits_and_keeps_later_candidates():
-    ltm = _FakeRetrieval(
-        [
-            Message(role="assistant", text="x" * 220),
-            Message(role="assistant", text="small-hit"),
-        ]
-    )
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 2,
-            "assemble_reserve_tokens": 0,
-            "assemble_ratio": 1.0,
-        },
-        knowledge={
-            "assemble_top_k": 0,
-            "assemble_ratio": 0.0,
-        },
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=35),
-        long_term_memory=ltm,
-        knowledge=None,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "small-hit"]
-    assert assembled.metadata["retrieval"]["ltm_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is True
-
-
-def test_context_assemble_reserve_tokens_respects_knowledge_only_config():
-    knowledge = _FakeRetrieval([Message(role="assistant", text="x" * 64)])
-    config = Config(
-        long_term_memory={"assemble_top_k": 0},
-        knowledge={
-            "assemble_top_k": 1,
-            "assemble_ratio": 1.0,
-            "assemble_reserve_tokens": 0,
-        },
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=40),
-        long_term_memory=None,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "x" * 64]
-    assert assembled.metadata["retrieval"]["knowledge_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is False
-
-
-def test_context_assemble_ignores_inactive_ltm_reserve_tokens_for_knowledge_only_retrieval():
-    knowledge = _FakeRetrieval([Message(role="assistant", text="x" * 64)])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 0,
-            "assemble_reserve_tokens": 10_000,
-        },
-        knowledge={
-            "assemble_top_k": 1,
-            "assemble_ratio": 1.0,
-            "assemble_reserve_tokens": 0,
-        },
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=40),
-        long_term_memory=None,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "x" * 64]
-    assert assembled.metadata["retrieval"]["knowledge_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is False
-
-
-def test_context_assemble_rebalances_budget_when_ltm_retrieval_fails():
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")], fail=True)
-    knowledge = _FakeRetrieval([Message(role="assistant", text="x" * 64)])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 1,
-            "assemble_ratio": 0.5,
-            "assemble_reserve_tokens": 0,
-        },
-        knowledge={
-            "assemble_top_k": 1,
-            "assemble_ratio": 0.5,
-            "assemble_reserve_tokens": 0,
-        },
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=34),
-        long_term_memory=ltm,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "x" * 64]
-    assert assembled.metadata["retrieval"]["ltm_count"] == 0
-    assert assembled.metadata["retrieval"]["knowledge_count"] == 1
-    assert assembled.metadata["retrieval"]["degraded"] is True
-    assert assembled.metadata["retrieval"]["degrade_reason"] == "ltm_retrieval_failed"
-
-
-def test_context_assemble_skips_zero_budget_source_retrieval_call() -> None:
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")])
-    knowledge = _FakeRetrieval([Message(role="assistant", text="knowledge-hit")])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 1,
-            "assemble_ratio": 0.0,
-            "assemble_reserve_tokens": 0,
-        },
-        knowledge={
-            "assemble_top_k": 1,
-            "assemble_ratio": 1.0,
-            "assemble_reserve_tokens": 0,
-        },
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=60),
-        long_term_memory=ltm,
-        knowledge=knowledge,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
-
-    assembled = ctx.assemble()
-
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q", "knowledge-hit"]
+    assert [message.text for message in assembled.messages] == ["latest request"]
     assert ltm.calls == []
-    assert len(knowledge.calls) == 1
-    assert assembled.metadata["retrieval"]["degraded"] is False
-
-
-def test_context_assemble_handles_overflowing_numeric_retrieval_config() -> None:
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": float("inf"),
-            "assemble_reserve_tokens": float("inf"),
-        },
-        knowledge={"assemble_top_k": 0},
-    )
-    ctx = Context(
-        config=config,
-        long_term_memory=ltm,
-        knowledge=None,
-    )
-    ctx.stm_add(Message(role="user", text="query"))
-
-    assembled = ctx.assemble()
+    assert knowledge.calls == []
+    assert assembled.metadata == {"context_id": ctx.id}
 
-    assert assembled.metadata["retrieval"]["ltm_requested"] == 3
 
+class _RecordingMovingCompressor:
+    def __init__(self) -> None:
+        self.calls: list[dict[str, object]] = []
 
-def test_context_assemble_rejects_infinite_ratio_and_keeps_budget_guardrails() -> None:
-    ltm = _FakeRetrieval([Message(role="assistant", text="x" * 220)])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 1,
-            "assemble_ratio": float("inf"),
-            "assemble_reserve_tokens": 0,
-        },
-        knowledge={"assemble_top_k": 0},
-    )
-    ctx = Context(
-        config=config,
-        budget=Budget(max_tokens=40),
-        long_term_memory=ltm,
-        knowledge=None,
-    )
-    ctx.stm_add(Message(role="user", text="q"))
+    async def prune(self, context: Context, **options: object) -> None:
+        self.calls.append({"context": context, "options": dict(options)})
 
-    assembled = ctx.assemble()
 
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["q"]
-    assert assembled.metadata["retrieval"]["ltm_count"] == 0
-    assert assembled.metadata["retrieval"]["degraded"] is True
-
-
-def test_context_assemble_handles_overflowing_numeric_ratio_config() -> None:
-    ltm = _FakeRetrieval([Message(role="assistant", text="ltm-hit")])
-    config = Config(
-        long_term_memory={
-            "assemble_top_k": 1,
-            "assemble_ratio": 10**10000,
-            "assemble_reserve_tokens": 0,
-        },
-        knowledge={"assemble_top_k": 0},
-    )
-    ctx = Context(
-        config=config,
-        long_term_memory=ltm,
-        knowledge=None,
-    )
-    ctx.stm_add(Message(role="user", text="query"))
+@pytest.mark.asyncio
+async def test_context_compress_without_moving_compressor_is_noop() -> None:
+    ctx = Context(config=Config())
+    ctx.stm_add(Message(role="user", text="hello"))
 
-    assembled = ctx.assemble()
+    await ctx.compress(max_context_tokens=128)
 
-    contents = [message.text for message in assembled.messages]
-    assert contents == ["query", "ltm-hit"]
-    assert assembled.metadata["retrieval"]["ltm_count"] == 1
+    assert [message.text for message in ctx.stm_get()] == ["hello"]
 
 
-def test_context_compress_max_messages_uses_backend_compress_only(monkeypatch: pytest.MonkeyPatch) -> None:
-    stm = _CompressionRecordingSTM(
-        [
-            Message(role="user", text="m0"),
-            Message(role="assistant", text="m1"),
-            Message(role="user", text="m2"),
-        ]
-    )
-    ctx = Context(config=Config(), short_term_memory=stm)
+@pytest.mark.asyncio
+async def test_context_compress_uses_context_window_tokens_when_present() -> None:
+    ctx = Context(config=Config(), context_window_tokens=256)
+    compressor = _RecordingMovingCompressor()
+    ctx.set_moving_compressor(compressor)
 
-    def _unexpected_compress_context(*args: object, **kwargs: object) -> None:
-        _ = (args, kwargs)
-        raise AssertionError("compress_context should not be called for basic max_messages compression")
+    await ctx.compress()
 
-    monkeypatch.setattr(
-        "dare_framework.compression.core.compress_context",
-        _unexpected_compress_context,
-    )
+    assert len(compressor.calls) == 1
+    assert compressor.calls[0]["context"] is ctx
+    assert compressor.calls[0]["options"] == {"max_context_tokens": 256}
 
-    ctx.compress(max_messages=2)
 
-    assert len(stm.compress_calls) == 1
-    assert stm.compress_calls[0].get("max_messages") == 2
-    assert [message.text for message in ctx.stm_get()] == ["m1", "m2"]
+@pytest.mark.asyncio
+async def test_context_compress_prefers_explicit_max_context_tokens() -> None:
+    ctx = Context(config=Config(), context_window_tokens=256)
+    compressor = _RecordingMovingCompressor()
+    ctx.set_moving_compressor(compressor)
 
+    await ctx.compress(max_context_tokens=64)
 
-def test_context_compress_advanced_path_preserves_backend_semantics(monkeypatch: pytest.MonkeyPatch) -> None:
-    stm = _CompressionRecordingSTM(
-        [
-            Message(role="user", text="m0"),
-            Message(role="assistant", text="m1"),
-            Message(role="user", text="m2"),
-        ]
-    )
-    ctx = Context(config=Config(), short_term_memory=stm)
-    calls: list[dict[str, object]] = []
-    stm_sizes_seen_by_strategy: list[int] = []
+    assert len(compressor.calls) == 1
+    assert compressor.calls[0]["options"] == {"max_context_tokens": 64}
 
-    def _record_compress_context(context: Context, **kwargs: object) -> None:
-        stm_sizes_seen_by_strategy.append(len(context.stm_get()))
-        calls.append(dict(kwargs))
 
-    monkeypatch.setattr(
-        "dare_framework.compression.core.compress_context",
-        _record_compress_context,
-    )
+@pytest.mark.asyncio
+async def test_context_assemble_for_model_runs_moving_compressor_with_context_window_tokens() -> None:
+    ctx = Context(config=Config(), context_window_tokens=256)
+    compressor = _RecordingMovingCompressor()
+    ctx.set_moving_compressor(compressor)
+    ctx.stm_add(Message(role="user", text="query"))
 
-    ctx.compress(max_messages=2, target_tokens=100, strategy="truncate", tool_pair_safe=True)
+    assembled = await ctx.assemble_for_model()
 
-    assert len(stm.compress_calls) == 1
-    assert stm.compress_calls[0].get("max_messages") == 2
-    assert calls and calls[0].get("max_messages") == 2
-    assert stm_sizes_seen_by_strategy == [3]
-    assert [message.text for message in ctx.stm_get()] == ["m1", "m2"]
+    assert len(compressor.calls) == 1
+    assert compressor.calls[0]["context"] is ctx
+    assert compressor.calls[0]["options"] == {"max_context_tokens": 256}
+    assert [message.text for message in assembled.messages] == ["query"]
diff --git a/tests/unit/test_example_10_agentscope_compat.py b/tests/unit/test_example_10_agentscope_compat.py
index 974dbf7c..a5ba4e61 100644
--- a/tests/unit/test_example_10_agentscope_compat.py
+++ b/tests/unit/test_example_10_agentscope_compat.py
@@ -263,6 +263,8 @@ def test_truncated_formatter_truncates_and_preserves_tool_pairs() -> None:
     assert tool_result_names.issubset(tool_call_ids)
 
 
+
+
 def test_json_session_bridge_roundtrip(tmp_path: Path) -> None:
     module = _load_example_module()
     notebook = module.CompatPlanNotebook()
diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index ff90142e..1304efa2 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -6,9 +6,8 @@
 
 from dare_framework.agent.react_agent import ReactAgent
 from dare_framework.config import Config
-from dare_framework.context import Context
+from dare_framework.context import Context, Message
 from dare_framework.context.types import MessageKind
-from dare_framework.context.smartcontext import SmartContext
 from dare_framework.model.types import ModelInput, ModelResponse
 from dare_framework.tool.types import CapabilityDescriptor, CapabilityType, ToolResult
 
@@ -152,32 +151,6 @@ async def generate(self, model_input: ModelInput, *, options: Any | None = None)
         return response
 
 
-class _CompressionRecordingContext(Context):
-    def __init__(self, *, config: Config) -> None:
-        super().__init__(config=config)
-        self.compress_calls: list[dict[str, Any]] = []
-
-    def compress(self, **options: Any) -> None:
-        self.compress_calls.append(dict(options))
-        super().compress(**options)
-
-
-class _CompressionRecordingSmartContext(SmartContext):
-    def __init__(self, *, config: Config) -> None:
-        super().__init__(config=config)
-        self.compress_calls: list[dict[str, Any]] = []
-
-    def compress(self, **options: Any) -> None:
-        self.compress_calls.append(dict(options))
-        super().compress(**options)
-
-
-class _FinalOnlyModel:
-    async def generate(self, model_input: ModelInput, *, options: Any | None = None) -> ModelResponse:
-        _ = (model_input, options)
-        return ModelResponse(content="final", tool_calls=[])
-
-
 class _NonConvergingToolModel:
     def __init__(self) -> None:
         self._idx = 0
@@ -274,7 +247,7 @@ async def test_react_agent_emits_intermediate_transport_events_in_order() -> Non
         tool_gateway=gateway,
     )
 
-    result = await agent.execute("test", transport=transport)
+    result = await agent.execute(Message(role="user", text="test"), transport=transport)
 
     assert result.success is True
     message_kinds = [envelope.payload.message_kind for envelope in transport.sent]
@@ -306,7 +279,7 @@ async def test_react_agent_transport_loop_emits_single_terminal_result_event() -
     )
 
     await agent._execute_polled_message(
-        "test",
+        Message(role="user", text="test"),
         channel=transport,
         envelope_id="req_1",
     )
@@ -337,7 +310,7 @@ async def test_react_agent_emits_terminal_message_for_repeated_tool_guard() -> N
         tool_gateway=gateway,
     )
 
-    result = await agent.execute("test", transport=transport)
+    result = await agent.execute(Message(role="user", text="test"), transport=transport)
 
     assert result.success is True
     assert transport.sent
@@ -360,7 +333,7 @@ async def test_react_agent_emits_terminal_message_for_max_round_exit() -> None:
         max_tool_rounds=2,
     )
 
-    result = await agent.execute("test", transport=transport)
+    result = await agent.execute(Message(role="user", text="test"), transport=transport)
 
     assert result.success is True
     assert transport.sent
@@ -369,93 +342,6 @@ async def test_react_agent_emits_terminal_message_for_max_round_exit() -> None:
     assert "达到最大轮次" in str(last_envelope.payload.data["output"])
 
 
-@pytest.mark.asyncio
-async def test_react_agent_auto_compress_triggers_before_model_call() -> None:
-    context = _CompressionRecordingContext(config=Config())
-    context.budget.max_tokens = 100
-    gateway = _RecordingGateway("injected")
-    agent = ReactAgent(
-        name="react-test-auto-compress",
-        model=_FinalOnlyModel(),
-        context=context,
-        tool_gateway=gateway,
-        auto_compress=True,
-        compress_trigger_ratio=0.01,
-        compress_target_ratio=0.5,
-    )
-
-    result = await agent("test auto compress trigger")
-
-    assert result.success is True
-    assert len(context.compress_calls) >= 1
-    first_call = context.compress_calls[0]
-    assert first_call.get("tool_pair_safe") is True
-    assert first_call.get("target_tokens") is not None
-
-
-@pytest.mark.asyncio
-async def test_react_agent_auto_compress_nan_ratios_fallback_to_defaults() -> None:
-    context = _CompressionRecordingContext(config=Config())
-    context.budget.max_tokens = 100
-    gateway = _RecordingGateway("injected")
-    agent = ReactAgent(
-        name="react-test-auto-compress-nan-ratios",
-        model=_FinalOnlyModel(),
-        context=context,
-        tool_gateway=gateway,
-        auto_compress=True,
-        compress_trigger_ratio=float("nan"),
-        compress_target_ratio=float("nan"),
-    )
-
-    result = await agent("x" * 600)
-
-    assert result.success is True
-    assert len(context.compress_calls) >= 1
-    first_call = context.compress_calls[0]
-    assert first_call.get("target_tokens") == 75
-
-
-@pytest.mark.asyncio
-async def test_react_agent_without_auto_compress_keeps_legacy_behavior() -> None:
-    context = _CompressionRecordingContext(config=Config())
-    gateway = _RecordingGateway("injected")
-    agent = ReactAgent(
-        name="react-test-no-auto-compress",
-        model=_FinalOnlyModel(),
-        context=context,
-        tool_gateway=gateway,
-        auto_compress=False,
-    )
-
-    result = await agent("test no auto compress")
-
-    assert result.success is True
-    assert context.compress_calls == []
-
-
-@pytest.mark.asyncio
-async def test_react_agent_auto_compress_triggers_in_smart_context_path() -> None:
-    context = _CompressionRecordingSmartContext(config=Config())
-    context.budget.max_tokens = 100
-    gateway = _RecordingGateway("injected")
-    agent = ReactAgent(
-        name="react-test-smartcontext-auto-compress",
-        model=_FinalOnlyModel(),
-        context=context,
-        tool_gateway=gateway,
-        auto_compress=True,
-        compress_trigger_ratio=0.01,
-        compress_target_ratio=0.5,
-    )
-
-    result = await agent("smart context compress")
-
-    assert result.success is True
-    assert len(context.compress_calls) >= 1
-    assert context.compress_calls[0].get("tool_pair_safe") is True
-
-
 @pytest.mark.asyncio
 async def test_react_agent_loop_guard_emits_terminal_message_event() -> None:
     context = Context(config=Config())
@@ -469,7 +355,7 @@ async def test_react_agent_loop_guard_emits_terminal_message_event() -> None:
         max_tool_rounds=10,
     )
 
-    result = await agent.execute("test loop guard", transport=transport)
+    result = await agent.execute(Message(role="user", text="test loop guard"), transport=transport)
 
     assert result.success is True
     assert transport.sent[-1].payload.message_kind is MessageKind.CHAT
@@ -489,7 +375,7 @@ async def test_react_agent_max_round_exit_emits_terminal_message_event() -> None
         max_tool_rounds=2,
     )
 
-    result = await agent.execute("test max rounds", transport=transport)
+    result = await agent.execute(Message(role="user", text="test max rounds"), transport=transport)
 
     assert result.success is True
     assert transport.sent[-1].payload.message_kind is MessageKind.CHAT