diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index 20884391..64a6e09d 100644
--- a/dare_framework/agent/react_agent.py
+++ b/dare_framework/agent/react_agent.py
@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import math
 import json
 from typing import Any
 
@@ -113,6 +114,11 @@ def __init__(
         tool_gateway: IToolGateway,
         plan_provider: IToolProvider | None = None,
         max_tool_rounds: int = 10,
+        auto_compress: bool = False,
+        compress_trigger_ratio: float = 0.9,
+        compress_target_ratio: float = 0.75,
+        compress_max_messages: int | None = None,
+        compress_strategy: str = "dedup_then_truncate",
         agent_channel: AgentChannel | None = None,
     ) -> None:
         super().__init__(name, agent_channel=agent_channel)
@@ -121,6 +127,11 @@ def __init__(
         self._context = context
         self._tool_gateway = tool_gateway
         self._plan_provider = plan_provider
+        self._auto_compress = bool(auto_compress)
+        self._compress_trigger_ratio = _clamp_ratio(compress_trigger_ratio, default=0.9)
+        self._compress_target_ratio = _clamp_ratio(compress_target_ratio, default=0.75)
+        self._compress_max_messages = compress_max_messages if isinstance(compress_max_messages, int) and compress_max_messages > 0 else None
+        self._compress_strategy = compress_strategy.strip() if isinstance(compress_strategy, str) and compress_strategy.strip() else "dedup_then_truncate"
         self._context.set_tool_gateway(self._tool_gateway)
 
         # 运行时检测是否可以启用 SmartContext 能力
@@ -166,28 +177,10 @@ async def _execute_basic(
         for round_idx in range(self._max_tool_rounds):
             print(f"[{self.name}] Round {round_idx + 1}/{self._max_tool_rounds}: 调用模型中...", flush=True)
             assembled = self._context.assemble()
-            messages = list(assembled.messages)
-            prompt_def = getattr(assembled, "sys_prompt", None)
-            if prompt_def is not None:
-                messages = [
-                    Message(
-                        role=prompt_def.role,
-                        content=prompt_def.content,
-                        name=prompt_def.name,
-                        metadata=dict(prompt_def.metadata),
-                    ),
-                    *messages,
-                ]
-            # Inject critical_block from plan_provider (maintained by plan tools)
-            if self._plan_provider is not None:
-                state = getattr(self._plan_provider, "state", None)
-                critical_block = getattr(state, "critical_block", "") if state else ""
-                if critical_block:
-                    print("\n--- [Plan State] (injected) ---\n" + critical_block + "\n---\n", flush=True)
-                    messages.insert(
-                        1,
-                        Message(role="system", content=critical_block, name="plan_state"),
-                    )
+            messages = self._build_model_messages(assembled)
+            if self._maybe_auto_compress(messages):
+                assembled = self._context.assemble()
+                messages = self._build_model_messages(assembled)
 
             model_input = ModelInput(
                 messages=messages,
@@ -403,10 +396,31 @@ async def _execute_with_smart_context(
             messages = self._context.order_messages_for_llm(messages, sys_prompt_message)
 
             # 注入 _next_round_reflection_prompt：普通工具轮后提示，仅本次 LLM 调用传入，不写 STM
+            injected_reflection_prompt = self._next_round_reflection_prompt
             if self._next_round_reflection_prompt is not None:
                 messages.append(self._next_round_reflection_prompt)
                 self._next_round_reflection_prompt = None
 
+            if self._maybe_auto_compress(messages):
+                assembled = self._context.assemble()
+                messages = list(assembled.messages)
+                prompt_def = getattr(assembled, "sys_prompt", None)
+                sys_prompt_message = (
+                    Message(
+                        role=prompt_def.role,
+                        content=prompt_def.content,
+                        name=prompt_def.name,
+                        metadata=dict(prompt_def.metadata),
+                        mark=MessageMark.IMMUTABLE,
+                        id="sys_prompt",
+                    )
+                    if prompt_def is not None
+                    else None
+                )
+                messages = self._context.order_messages_for_llm(messages, sys_prompt_message)
+                if injected_reflection_prompt is not None:
+                    messages.append(injected_reflection_prompt)
+
             # Inject critical_block from plan_provider (maintained by plan tools)
             # Disabled: skip injection to observe plan agent behavior without it
             if False and self._plan_provider is not None:
@@ -571,6 +585,62 @@ async def _execute_with_smart_context(
             output_text=final_message,
         )
 
+    def _build_model_messages(self, assembled: Any) -> list[Message]:
+        """Build model-facing messages including system prompt and plan state injection."""
+        messages = list(assembled.messages)
+        prompt_def = getattr(assembled, "sys_prompt", None)
+        if prompt_def is not None:
+            messages = [
+                Message(
+                    role=prompt_def.role,
+                    content=prompt_def.content,
+                    name=prompt_def.name,
+                    metadata=dict(prompt_def.metadata),
+                ),
+                *messages,
+            ]
+        if self._plan_provider is not None:
+            state = getattr(self._plan_provider, "state", None)
+            critical_block = getattr(state, "critical_block", "") if state else ""
+            if critical_block:
+                print("\n--- [Plan State] (injected) ---\n" + critical_block + "\n---\n", flush=True)
+                messages.insert(
+                    1,
+                    Message(role="system", content=critical_block, name="plan_state"),
+                )
+        return messages
+
+    def _maybe_auto_compress(self, model_messages: list[Message]) -> bool:
+        """Auto-compress context before model invocation when token estimate is near budget."""
+        if not self._auto_compress:
+            return False
+        max_tokens = self._context.budget.max_tokens
+        if max_tokens is None or max_tokens <= 0:
+            return False
+
+        estimated_tokens = _estimate_messages_tokens(model_messages)
+        trigger_tokens = max(1, int(max_tokens * self._compress_trigger_ratio))
+        if estimated_tokens < trigger_tokens:
+            return False
+
+        stm_messages = self._context.stm_get()
+        if not stm_messages:
+            return False
+        max_messages = self._compress_max_messages
+        if max_messages is None:
+            max_messages = max(1, int(len(stm_messages) * self._compress_target_ratio))
+            if max_messages >= len(stm_messages):
+                max_messages = max(1, len(stm_messages) - 1)
+
+        target_tokens = max(1, int(max_tokens * self._compress_target_ratio))
+        self._context.compress(
+            strategy=self._compress_strategy,
+            max_messages=max_messages,
+            target_tokens=target_tokens,
+            tool_pair_safe=True,
+        )
+        return True
+
     async def _emit_terminal_transport_message(
         self,
         *,
@@ -701,4 +771,26 @@ def _tool_calls_signature(tool_calls: list[dict[str, Any]]) -> tuple[str, ...]:
     return tuple(signature)
 
 
+def _estimate_messages_tokens(messages: list[Message]) -> int:
+    total = 0
+    for message in messages:
+        content = (message.content or "").strip()
+        total += max(1, len(content) // 4) + 8
+    return total
+
+
+def _clamp_ratio(value: Any, *, default: float) -> float:
+    try:
+        ratio = float(value)
+    except (TypeError, ValueError):
+        return default
+    if not math.isfinite(ratio):
+        return default
+    if ratio <= 0:
+        return default
+    if ratio > 1:
+        return 1.0
+    return ratio
+
+
 __all__ = ["ReactAgent"]
diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index a7bd9af5..a52b246c 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -11,7 +11,7 @@
 
 from typing import TYPE_CHECKING, Any, List, Tuple
 
-from dare_framework.context.types import Message as CtxMessage
+from dare_framework.context.types import Message as CtxMessage, MessageMark
 from dare_framework.model import ModelInput
 
 if TYPE_CHECKING:
@@ -92,6 +92,156 @@ def _build_summary_preview(
     return new_messages, removed
 
 
+def _estimate_tokens(messages: List[Message]) -> int:
+    """Rough token estimate using a cheap character-based heuristic."""
+    total = 0
+    for msg in messages:
+        content = (msg.content or "").strip()
+        total += max(1, len(content) // 4) + 8
+    return total
+
+
+def _trim_to_target_tokens(messages: List[Message], target_tokens: int | None) -> Tuple[List[Message], int]:
+    """Trim oldest messages until estimated token size fits target_tokens."""
+    if target_tokens is None or target_tokens <= 0:
+        return messages, 0
+    if _estimate_tokens(messages) <= target_tokens:
+        return messages, 0
+
+    trimmed = list(messages)
+    removed = 0
+    while len(trimmed) > 1 and _estimate_tokens(trimmed) > target_tokens:
+        removable_idx = next(
+            (
+                idx
+                for idx, message in enumerate(trimmed)
+                if getattr(message, "mark", MessageMark.TEMPORARY)
+                not in (MessageMark.IMMUTABLE, MessageMark.PERSISTENT)
+            ),
+            None,
+        )
+        if removable_idx is None:
+            # Nothing removable left (all protected by mark semantics).
+            break
+        trimmed.pop(removable_idx)
+        removed += 1
+    return trimmed, removed
+
+
+def _extract_tool_call_ids(message: Message) -> list[str]:
+    """Collect tool call ids declared on an assistant message."""
+    if message.role != "assistant":
+        return []
+    tool_calls = message.metadata.get("tool_calls", [])
+    if not isinstance(tool_calls, list):
+        return []
+
+    ids: list[str] = []
+    for call in tool_calls:
+        if not isinstance(call, dict):
+            continue
+        tool_id = call.get("id")
+        if isinstance(tool_id, str) and tool_id.strip():
+            ids.append(tool_id.strip())
+    return ids
+
+
+def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], int]:
+    """Keep tool_call/tool_result in sync so compression never leaves orphan pairs."""
+    tool_result_ids = {
+        message.name.strip()
+        for message in messages
+        if message.role == "tool" and isinstance(message.name, str) and message.name.strip()
+    }
+
+    updated_messages: list[Message] = []
+    retained_call_ids: set[str] = set()
+    retained_idless_tool_names: set[str] = set()
+    changes = 0
+    for message in messages:
+        if message.role != "assistant":
+            updated_messages.append(message)
+            continue
+        raw_calls = message.metadata.get("tool_calls", [])
+        if not isinstance(raw_calls, list):
+            updated_messages.append(message)
+            continue
+
+        filtered_calls = []
+        for call in raw_calls:
+            if not isinstance(call, dict):
+                continue
+            tool_id = call.get("id")
+            if isinstance(tool_id, str) and tool_id.strip() and tool_id.strip() in tool_result_ids:
+                filtered_calls.append(call)
+                retained_call_ids.add(tool_id.strip())
+                continue
+            if not isinstance(tool_id, str) or not tool_id.strip():
+                # Some providers emit tool calls without stable ids. Keep these calls and
+                # retain matching tool results by tool name.
+                filtered_calls.append(call)
+                tool_name = call.get("name")
+                if isinstance(tool_name, str) and tool_name.strip():
+                    retained_idless_tool_names.add(tool_name.strip())
+
+        if len(filtered_calls) != len(raw_calls):
+            changes += len(raw_calls) - len(filtered_calls)
+            metadata = dict(message.metadata)
+            metadata["tool_calls"] = filtered_calls
+            updated_messages.append(
+                CtxMessage(
+                    role=message.role,
+                    content=message.content,
+                    name=message.name,
+                    metadata=metadata,
+                    mark=getattr(message, "mark", MessageMark.TEMPORARY),
+                    id=getattr(message, "id", None),
+                )
+            )
+        else:
+            retained_call_ids.update(_extract_tool_call_ids(message))
+            updated_messages.append(message)
+
+    final_messages: list[Message] = []
+    for message in updated_messages:
+        if message.role == "tool":
+            tool_id = message.name.strip() if isinstance(message.name, str) else ""
+            if tool_id in retained_call_ids:
+                final_messages.append(message)
+                continue
+            if tool_id and tool_id in retained_idless_tool_names:
+                final_messages.append(message)
+                continue
+            if tool_id:
+                changes += 1
+                continue
+        final_messages.append(message)
+    return final_messages, changes
+
+
+def _annotate_strategy(messages: List[Message], strategy: str) -> List[Message]:
+    """Attach strategy metadata to the first message when compression rewrites context."""
+    if not messages:
+        return messages
+    for message in messages:
+        if message.metadata.get("compressed") is True:
+            return messages
+
+    head = messages[0]
+    metadata = dict(head.metadata)
+    metadata["compressed"] = True
+    metadata.setdefault("strategy", strategy)
+    messages[0] = CtxMessage(
+        role=head.role,
+        content=head.content,
+        name=head.name,
+        metadata=metadata,
+        mark=getattr(head, "mark", MessageMark.TEMPORARY),
+        id=getattr(head, "id", None),
+    )
+    return messages
+
+
 def compress_context(
     context: IContext,
     *,
@@ -114,8 +264,16 @@ def compress_context(
     NOTE:
         - phase 参数暂未参与决策，仅为未来差异化调用预留。
     """
-    # 未提供 max_messages 时，不做任何压缩（由调用方决定是否传入）。
-    if max_messages is None or max_messages < 0:
+    # 未提供任何压缩上限时，不做压缩（由调用方决定是否传入）。
+    target_tokens_raw = options.get("target_tokens")
+    target_tokens: int | None = None
+    if target_tokens_raw is not None:
+        try:
+            target_tokens = int(target_tokens_raw)
+        except (TypeError, ValueError):
+            target_tokens = None
+
+    if (max_messages is None or max_messages < 0) and (target_tokens is None or target_tokens <= 0):
         return
 
     # 通过 IContext 的标准接口访问 STM，避免绑定具体实现细节。
@@ -129,8 +287,15 @@ def compress_context(
     if not messages:
         return
 
+    if max_messages is None:
+        max_messages = len(messages)
+    elif max_messages < 0:
+        # Keep historical sentinel semantics: negative means "no message cap".
+        max_messages = len(messages)
+
     # strategy 默认为 "truncate"，后续可扩展更多策略。
     strategy = options.get("strategy", "truncate")
+    tool_pair_safe = bool(options.get("tool_pair_safe", False))
 
     removed_total = 0
 
@@ -150,13 +315,51 @@ def compress_context(
         removed_total += len(messages)
         messages = []
     elif len(messages) > max_messages:
-        removed_total += len(messages) - max_messages
-        messages = messages[-max_messages:]
+        protected = [
+            message
+            for message in messages
+            if getattr(message, "mark", MessageMark.TEMPORARY)
+            in (MessageMark.IMMUTABLE, MessageMark.PERSISTENT)
+        ]
+        temporary = [
+            message
+            for message in messages
+            if getattr(message, "mark", MessageMark.TEMPORARY)
+            not in (MessageMark.IMMUTABLE, MessageMark.PERSISTENT)
+        ]
+        keep_temporary = max(max_messages - len(protected), 0)
+        if keep_temporary <= 0:
+            kept_tail = []
+        elif keep_temporary < len(temporary):
+            kept_tail = temporary[-keep_temporary:]
+        else:
+            kept_tail = temporary
+        kept_tail_refs = {id(message) for message in kept_tail}
+        messages = [
+            message
+            for message in messages
+            if (
+                getattr(message, "mark", MessageMark.TEMPORARY)
+                in (MessageMark.IMMUTABLE, MessageMark.PERSISTENT)
+            )
+            or id(message) in kept_tail_refs
+        ]
+        removed_total += len(protected) + len(temporary) - len(messages)
+
+    # Step 4: token-aware 截断（按估算 token 控制）
+    messages, removed = _trim_to_target_tokens(messages, target_tokens)
+    removed_total += removed
+
+    # Step 5: 工具调用对齐保护（可选）
+    if tool_pair_safe:
+        messages, changes = _enforce_tool_pair_safety(messages)
+        removed_total += changes
 
     # 如无任何压缩，不改写 STM，避免无意义写操作。
     if removed_total == 0:
         return
 
+    messages = _annotate_strategy(messages, str(strategy))
     stm_clear()
     for msg in messages:
         stm_add(msg)
diff --git a/dare_framework/context/context.py b/dare_framework/context/context.py
index cd631d21..76ff575f 100644
--- a/dare_framework/context/context.py
+++ b/dare_framework/context/context.py
@@ -178,8 +178,31 @@ def assemble(self) -> AssembledContext:
 
     def compress(self, **options: Any) -> None:
         """Compress context to fit within budget."""
-        if self._short_term_memory is not None:
-            self._short_term_memory.compress(**options)
+        from dare_framework.compression.core import compress_context
+
+        # Preserve backend STM semantics (for example SmartSTM mark-based retention)
+        # before applying advanced compression strategies.
+        compress_impl = getattr(self._short_term_memory, "compress", None)
+        has_advanced_options = any(
+            key in options
+            for key in ("target_tokens", "tool_pair_safe", "strategy", "phase")
+        )
+        raw_max_messages = options.get("max_messages")
+        max_messages = (
+            raw_max_messages
+            if isinstance(raw_max_messages, int) and raw_max_messages >= 0
+            else None
+        )
+        if callable(compress_impl) and not has_advanced_options:
+            compress_impl(max_messages=max_messages)
+            return
+
+        compress_context(self, **options)
+
+        # For advanced compression, run backend max-message retention after strategy
+        # execution so strategy implementations can inspect full pre-trim history.
+        if callable(compress_impl) and has_advanced_options and max_messages is not None:
+            compress_impl(max_messages=max_messages)
 
 
 class DefaultAssembledContext(IAssembleContext):
diff --git a/dare_framework/transport/_internal/adapters.py b/dare_framework/transport/_internal/adapters.py
index 68aa2a50..f2dc4afc 100644
--- a/dare_framework/transport/_internal/adapters.py
+++ b/dare_framework/transport/_internal/adapters.py
@@ -308,7 +308,14 @@ def _render_status_output(payload: dict[str, Any]) -> Any:
             return f"approval resolved: request_id={request_id} decision={decision}"
         if request_id:
             return f"approval pending: request_id={request_id}"
-        return "approval update"
+        if any(key in resp for key in ("request", "request_id", "decision")):
+            return "approval update"
+        if "phase" in resp:
+            return resp.get("phase")
+        if "event" in resp:
+            return resp.get("event")
+        return resp
+
     if "phase" in payload:
         return payload.get("phase")
     if "event" in payload:
diff --git a/docs/features/agentscope-d5-safe-compression.md b/docs/features/agentscope-d5-safe-compression.md
new file mode 100644
index 00000000..512516e5
--- /dev/null
+++ b/docs/features/agentscope-d5-safe-compression.md
@@ -0,0 +1,63 @@
+---
+change_ids: ["agentscope-d5-safe-compression"]
+doc_kind: feature
+topics: ["agentscope", "compression", "context", "react-agent", "budget"]
+created: 2026-03-02
+updated: 2026-03-02
+status: in_review
+mode: openspec
+---
+
+# Feature: agentscope-d5-safe-compression
+
+## Scope
+补齐 AgentScope 迁移 D5：上下文压缩安全性与预算收敛能力，包括 tool-pair-safe、token-aware 触发与 ReAct 调用前自动压缩。
+
+## OpenSpec Artifacts
+- Proposal: `openspec/changes/agentscope-d5-safe-compression/proposal.md`
+- Design: `openspec/changes/agentscope-d5-safe-compression/design.md`
+- Specs:
+  - `openspec/changes/agentscope-d5-safe-compression/specs/agentscope-safe-compression/spec.md`
+  - `openspec/changes/agentscope-d5-safe-compression/specs/chat-runtime/spec.md`
+- Tasks: `openspec/changes/agentscope-d5-safe-compression/tasks.md`
+
+## Progress
+- 已完成：D5 代码实现（tool-pair-safe、token-aware compression、ReAct pre-model auto-compress）。
+- 已完成：OpenSpec tasks 全部打勾（10/10）。
+- 已完成：提交 PR #136，进入评审阶段。
+- 待完成：评审反馈处理与合并门禁记录闭环。
+
+## Evidence
+
+### Commands
+- `openspec new change "agentscope-d5-safe-compression"`
+- `openspec status --change "agentscope-d5-safe-compression" --json`
+- `openspec instructions apply --change "agentscope-d5-safe-compression" --json`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_context_compression.py tests/unit/test_react_agent_gateway_injection.py`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_context_implementation.py tests/unit/test_agent_output_envelope.py tests/unit/test_example_10_agentscope_compat.py`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q`
+
+### Results
+- OpenSpec change 创建成功（schema: spec-driven）。
+- OpenSpec apply status：`10/10 tasks complete`（state=`all_done`）。
+- D5 定向新增回归：`9 passed, 1 warning`。
+- 受影响面回归：`38 passed, 1 warning`。
+- 全量回归：`533 passed, 12 skipped, 1 warning`。
+
+### Behavior Verification
+- Happy path:
+  - `compress_context(..., tool_pair_safe=True)` 可消除孤儿 `tool_result` 并修正无匹配的 `tool_call`；
+  - `target_tokens` 生效时可在保留至少一条消息前提下收敛历史消息；
+  - `ReactAgent(auto_compress=True)` 在模型调用前按预算阈值触发压缩。
+- Error branch:
+  - `ReactAgent(auto_compress=False)` 保持旧行为，不触发压缩调用；
+  - 压缩后仍保留结构化 metadata 标记（`compressed + strategy`），便于诊断回滚。
+
+### Risks and Rollback
+- 风险：压缩触发时机变化可能影响部分场景回答质量。
+- 风险：tool pair 保护逻辑若实现不严谨可能导致消息遗漏。
+- 回滚：保持自动压缩开关可控，必要时回退到手动压缩路径。
+
+### Review and Merge Gate Links
+- PR：`https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/136`
+- Merge Gate：待评审通过后补充（approval + merge commit）。
diff --git a/docs/todos/agentscope_domain_execution_todos.md b/docs/todos/agentscope_domain_execution_todos.md
index d2f37c9a..7d2e6f97 100644
--- a/docs/todos/agentscope_domain_execution_todos.md
+++ b/docs/todos/agentscope_domain_execution_todos.md
@@ -21,10 +21,10 @@
 
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
-| CLM-20260302-D2D4 | D2-1~D2-4, D4-1~D4-4 | mindfn | active | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 先处理 P0/P1 的 thinking 与 transport 协议统一。 |
-| CLM-20260302-D5 | D5-1~D5-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 压缩链路：tool pair safe + token-aware + auto trigger。 |
-| CLM-20260302-D7 | D7-1~D7-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | plan 状态机与 finish/revise 原生工具补齐。 |
-| CLM-20260302-D1D3 | D1-1~D1-4, D3-1~D3-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 多模态输入 schema 与 assemble normalize。 |
+| CLM-20260302-D2D4 | D2-1~D2-4, D4-1~D4-4 | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 先处理 P0/P1 的 thinking 与 transport 协议统一（PR #134 review 中）。 |
+| CLM-20260302-D5 | D5-1~D5-4 | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | D5 实现与回归已完成，PR #136 待审。 |
+| CLM-20260302-D7 | D7-1~D7-4 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | plan 状态机与 finish/revise 原生工具补齐。 |
+| CLM-20260302-D1D3 | D1-1~D1-4, D3-1~D3-4 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 多模态输入 schema 与 assemble normalize。 |
 
 ---
 
@@ -94,10 +94,10 @@
 
 | ID | 任务 | 主要代码改动 | 支持能力 | 依赖 | 状态 | 输出证据 |
 |---|---|---|---|---|---|---|
-| D2-1 | 定义消息类型枚举 | enum + 常量统一 | thinking/tool 事件标准化 | D1-1 | todo | 枚举被全链路复用 |
-| D2-2 | 定义 payload 协议 | envelope schema + serializer | transport 语义一致 | D2-1 | todo | 往返一致性测试 |
-| D2-3 | 错误码标准化 | error payload model | 可观测错误治理 | D2-2 | todo | CLI/transport 输出一致 |
-| D2-4 | 协议测试矩阵 | e2e + contract test | 端到端稳定性 | D2-1/2/3 | todo | message/tool/thinking/error 全覆盖 |
+| D2-1 | 定义消息类型枚举 | enum + 常量统一 | thinking/tool 事件标准化 | D1-1 | done | `tests/unit/test_transport_types.py` |
+| D2-2 | 定义 payload 协议 | envelope schema + serializer | transport 语义一致 | D2-1 | done | `tests/unit/test_transport_adapters.py` |
+| D2-3 | 错误码标准化 | error payload model | 可观测错误治理 | D2-2 | done | `tests/unit/test_transport_channel.py` |
+| D2-4 | 协议测试矩阵 | e2e + contract test | 端到端稳定性 | D2-1/2/3 | done | `tests/unit/test_transport_types.py`, `tests/unit/test_transport_adapters.py` |
 
 ---
 
@@ -145,10 +145,10 @@
 
 | ID | 任务 | 主要代码改动 | 支持能力 | 依赖 | 状态 | 输出证据 |
 |---|---|---|---|---|---|---|
-| D4-1 | thinking_content | ModelResponse + adapter 提取 | ChatModelBase thinking | D2-1/2 | todo | 单测：thinking 不丢失 |
-| D4-2 | reasoning_tokens | usage parser 标准化 | 预算计量准确性 | D4-1 | todo | usage 测试覆盖 |
-| D4-3 | 中间态事件 | loop 内发送 thinking/tool 事件 | 执行可观测 | D2-1/2/3 | todo | 端到端事件序列测试 |
-| D4-4 | 回归验证 | model+loop 集成测试 | 模型输出链路稳定 | D4-1/2/3 | todo | 无回归 |
+| D4-1 | thinking_content | ModelResponse + adapter 提取 | ChatModelBase thinking | D2-1/2 | done | `tests/unit/test_openai_model_adapter.py`, `tests/unit/test_openrouter_adapter.py` |
+| D4-2 | reasoning_tokens | usage parser 标准化 | 预算计量准确性 | D4-1 | done | `tests/unit/test_openai_model_adapter.py`, `tests/unit/test_openrouter_adapter.py` |
+| D4-3 | 中间态事件 | loop 内发送 thinking/tool 事件 | 执行可观测 | D2-1/2/3 | done | `tests/unit/test_react_agent_gateway_injection.py` |
+| D4-4 | 回归验证 | model+loop 集成测试 | 模型输出链路稳定 | D4-1/2/3 | done | `pytest -q`（528 passed, 12 skipped, 1 warning） |
 
 ---
 
@@ -170,10 +170,10 @@
 
 | ID | 任务 | 主要代码改动 | 支持能力 | 依赖 | 状态 | 输出证据 |
 |---|---|---|---|---|---|---|
-| D5-1 | tool pair safe | compression 截断算法 | F1/Mem5 | D3-1 + D4-3 | todo | 工具对完整性测试 |
-| D5-2 | token 压缩 | token-aware 策略 | F2 | D4-2 | todo | token 预算测试 |
-| D5-3 | 自动触发 | model 调用前压缩 hook | F4/R5 | D5-1/2 | todo | 超阈值自动收敛 |
-| D5-4 | 压缩矩阵测试 | truncate/summary/pair-safe | 压缩质量与稳定 | D5-1/2/3 | todo | 策略矩阵全通过 |
+| D5-1 | tool pair safe | compression 截断算法 | F1/Mem5 | D3-1 + D4-3 | done | `tests/unit/test_context_compression.py` |
+| D5-2 | token 压缩 | token-aware 策略 | F2 | D4-2 | done | `tests/unit/test_context_compression.py` |
+| D5-3 | 自动触发 | model 调用前压缩 hook | F4/R5 | D5-1/2 | done | `tests/unit/test_react_agent_gateway_injection.py` |
+| D5-4 | 压缩矩阵测试 | truncate/summary/pair-safe | 压缩质量与稳定 | D5-1/2/3 | done | `pytest -q`（533 passed, 12 skipped, 1 warning） |
 
 ---
 
diff --git a/docs/todos/project_overall_todos.md b/docs/todos/project_overall_todos.md
index 6dedf554..46307e1d 100644
--- a/docs/todos/project_overall_todos.md
+++ b/docs/todos/project_overall_todos.md
@@ -15,10 +15,10 @@
 
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
-| CLM-20260302-AG1 | T5-2 | mindfn | active | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 对齐 D2/D4：thinking + transport 事件链路。 |
-| CLM-20260302-AG2 | T2-1 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 对齐 D5：安全压缩与预算收敛。 |
-| CLM-20260302-AG3 | D7-1~D7-4（关联 T5-5） | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | 先按 AgentScope gap 切片推进 plan 状态机能力。 |
-| CLM-20260302-AG4 | T5-3 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 对齐 D1/D3：多模态输入 schema + normalize。 |
+| CLM-20260302-AG1 | T5-2 | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 对齐 D2/D4：thinking + transport 事件链路（PR #134 review 中）。 |
+| CLM-20260302-AG2 | T2-1 | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 对齐 D5：安全压缩与预算收敛（PR #136 待审）。 |
+| CLM-20260302-AG3 | D7-1~D7-4（关联 T5-5） | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | 先按 AgentScope gap 切片推进 plan 状态机能力。 |
+| CLM-20260302-AG4 | T5-3 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 对齐 D1/D3：多模态输入 schema + normalize。 |
 
 ## 2. 当前基线
 
diff --git a/openspec/changes/agentscope-d5-safe-compression/.openspec.yaml b/openspec/changes/agentscope-d5-safe-compression/.openspec.yaml
new file mode 100644
index 00000000..fd79bfc5
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/.openspec.yaml
@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-03-02
diff --git a/openspec/changes/agentscope-d5-safe-compression/design.md b/openspec/changes/agentscope-d5-safe-compression/design.md
new file mode 100644
index 00000000..0509fa59
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/design.md
@@ -0,0 +1,50 @@
+## Context
+
+当前 `compress_context` 主要支持按条数截断、去重截断与启发式摘要；`ReactAgent` 仅做预算累计与检查，并未在调用模型前自动压缩。对于包含工具调用的长链路，这会导致历史中 `assistant(tool_calls)` 与 `tool` 结果被拆断，降低后续轮次可解释性与稳定性。
+
+## Goals / Non-Goals
+
+**Goals:**
+- 保证压缩后不会产生孤儿 `tool_call` 或 `tool_result`。
+- 在 token 预算紧张时自动触发压缩并继续执行。
+- 给压缩结果打上稳定策略标记，支持后续日志/审计。
+
+**Non-Goals:**
+- 不在本切片引入新的 LLM provider。
+- 不在本切片实现跨 session 持久化压缩策略。
+- 不在本切片实现完整可配置策略中心（仅补最小必要参数）。
+
+## Decisions
+
+1. **tool pair safe 作为压缩后置保护**
+- Decision: 在压缩得到候选消息序列后，执行一次成对校验与修复，确保 `assistant.tool_calls` 与对应 `tool` 消息同留或同删。
+- Rationale: 兼容现有压缩策略，实现侵入最小。
+
+2. **token-aware 触发采用轻量估算优先**
+- Decision: 复用现有 token 估算启发式，在 `max_tokens` 预算接近阈值时触发压缩；后续可扩展为 provider 真实 tokenizer。
+- Rationale: 本轮优先闭环能力与稳定性，避免引入额外依赖。
+
+3. **自动触发放在 ReAct 模型调用前**
+- Decision: 在 `ReactAgent.execute` 每轮组装消息后、`model.generate` 前判断并触发 `context.compress(...)`。
+- Rationale: 时序明确，且能覆盖多轮工具调用场景。
+
+4. **验证采用“策略单测 + 执行链路回归”**
+- Decision: 压缩核心做策略级单测；ReAct 做触发时序与非回归测试。
+- Rationale: 同时保障算法正确性与运行时行为。
+
+## Risks / Trade-offs
+
+- [Risk] 压缩策略更积极可能影响模型回答质量。  
+  Mitigation: 先默认保守阈值，保留回滚开关。
+- [Risk] tool-pair 保护增加实现复杂度。  
+  Mitigation: 仅覆盖单轮内标准 `tool_call_id` 匹配语义并补回归。
+
+## Migration Plan
+
+1. 扩展压缩 core：pair-safe + token-aware 策略。
+2. 接入 ReAct 自动触发点并补保护参数。
+3. 补齐 D5 回归测试矩阵并跑全量测试。
+4. 回写 feature evidence 与 TODO 状态。
+
+Rollback:
+- 如出现行为回归，可关闭自动触发，仅保留手动压缩路径。
diff --git a/openspec/changes/agentscope-d5-safe-compression/proposal.md b/openspec/changes/agentscope-d5-safe-compression/proposal.md
new file mode 100644
index 00000000..7752380b
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/proposal.md
@@ -0,0 +1,33 @@
+## Why
+
+AgentScope 迁移里，D5 压缩链路仍未闭环：当前压缩以消息条数为主，缺少 `tool_call/tool_result` 成对保护，也未在模型调用前按 token 预算自动触发。结果是长对话下易出现上下文溢出、工具语义断裂与行为不稳定。
+
+## What Changes
+
+- 增加 D5 切片能力：
+  - `tool pair safe` 压缩（禁止孤儿化 tool 调用或结果）；
+  - token-aware 压缩阈值与预算收敛；
+  - ReAct 模型调用前自动触发压缩。
+- 统一压缩策略输入参数与输出标记，便于后续 observability 追踪。
+- 增加覆盖 `truncate/summary/tool-pair-safe/auto-trigger` 的回归测试。
+
+## Capabilities
+
+### New Capabilities
+- `agentscope-safe-compression`: 在 AgentScope 对齐路径中提供可预测、可验证的安全压缩能力。
+
+### Modified Capabilities
+- `chat-runtime`: 模型调用前可按预算自动压缩上下文。
+- `context-memory`: 压缩策略对工具消息对保持完整性并支持 token-aware 策略。
+
+## Impact
+
+- Affected code:
+  - `dare_framework/compression/core.py`
+  - `dare_framework/context/context.py`
+  - `dare_framework/agent/react_agent.py`
+- Affected tests:
+  - `tests/unit/test_context_implementation.py`
+  - `tests/unit/test_agent_output_envelope.py`
+  - `tests/unit/test_example_10_agentscope_compat.py`
+- No external dependency change.
diff --git a/openspec/changes/agentscope-d5-safe-compression/specs/agentscope-safe-compression/spec.md b/openspec/changes/agentscope-d5-safe-compression/specs/agentscope-safe-compression/spec.md
new file mode 100644
index 00000000..ac7fb202
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/specs/agentscope-safe-compression/spec.md
@@ -0,0 +1,26 @@
+## ADDED Requirements
+
+### Requirement: Tool pair safe compression
+The runtime SHALL preserve tool invocation/result integrity during compression.  
+When a compressed context retains an assistant tool call, it SHALL retain matching tool result messages for the retained call ids.
+
+#### Scenario: Compression keeps tool call and result together
+- **GIVEN** context messages include `assistant(tool_calls=[id=tc-1])` and `tool(name=tc-1)`
+- **WHEN** compression is triggered
+- **THEN** both messages are retained together, or both are removed together
+- **AND** no orphan `tool` message remains without a matching retained tool call
+
+### Requirement: Token-aware compression trigger
+The runtime SHALL support token-aware compression decisions before model generation.
+
+#### Scenario: Compression is triggered when token budget is near limit
+- **GIVEN** current context token estimate is near/exceeds configured token threshold
+- **WHEN** the execute loop is about to call the model
+- **THEN** compression is triggered automatically before model invocation
+
+### Requirement: Compression strategy observability metadata
+Compressed messages SHALL include strategy metadata for downstream observability.
+
+#### Scenario: Compressed summary carries strategy marker
+- **WHEN** runtime applies summary or pair-safe compression
+- **THEN** compressed system message metadata includes a stable strategy identifier
diff --git a/openspec/changes/agentscope-d5-safe-compression/specs/chat-runtime/spec.md b/openspec/changes/agentscope-d5-safe-compression/specs/chat-runtime/spec.md
new file mode 100644
index 00000000..47342da1
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/specs/chat-runtime/spec.md
@@ -0,0 +1,12 @@
+## MODIFIED Requirements
+
+### Requirement: LLM-driven execute loop
+The runtime SHALL invoke the configured `IModelAdapter` during the execute loop and iterate over tool calls until the model returns a final response.
+
+Before each model invocation, the execute loop SHALL perform token-aware context compression when configured thresholds are exceeded.
+
+#### Scenario: Execute loop auto-compresses before model call
+- **GIVEN** a ReAct run with context token usage above configured threshold
+- **WHEN** the loop prepares to call `model.generate`
+- **THEN** context compression runs before model invocation
+- **AND** resulting model call sees compressed context messages
diff --git a/openspec/changes/agentscope-d5-safe-compression/tasks.md b/openspec/changes/agentscope-d5-safe-compression/tasks.md
new file mode 100644
index 00000000..eaaeb080
--- /dev/null
+++ b/openspec/changes/agentscope-d5-safe-compression/tasks.md
@@ -0,0 +1,21 @@
+## 1. Compression strategy alignment (D5 core)
+
+- [x] 1.1 Add tool-pair-safe guard in compression flow to preserve assistant tool_call and tool_result integrity.
+- [x] 1.2 Add token-aware compression threshold handling for runtime pre-model invocation decisions.
+- [x] 1.3 Ensure compression outputs include stable strategy metadata markers for observability.
+
+## 2. ReAct execute loop integration
+
+- [x] 2.1 Integrate pre-model auto-compression trigger in `ReactAgent.execute`.
+- [x] 2.2 Keep non-compression paths backward compatible for existing direct-chat behavior.
+
+## 3. Verification and regression
+
+- [x] 3.1 Add unit tests for tool pair safe behavior and token-aware compression trigger.
+- [x] 3.2 Add execute-loop regression tests validating compression trigger timing and no final-output regression.
+- [x] 3.3 Run targeted tests and full regression (`pytest -q`) and record outputs.
+
+## 4. Documentation and ledger sync
+
+- [x] 4.1 Update TODO claim/evidence entries for D5 from `active` to execution evidence.
+- [x] 4.2 Update feature aggregation evidence with commands, behavior checks, and risks/rollback.
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
new file mode 100644
index 00000000..93f6808c
--- /dev/null
+++ b/tests/unit/test_context_compression.py
@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+from dare_framework.compression.core import compress_context
+from dare_framework.config import Config
+from dare_framework.context import Context, Message, MessageMark
+
+
+def _tool_ids(message: Message) -> list[str]:
+    raw_calls = message.metadata.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",
+            content="tool call",
+            metadata={"tool_calls": [{"id": "tc_1", "name": "demo_tool", "arguments": {"x": 1}}]},
+        )
+    )
+    ctx.stm_add(Message(role="tool", name="tc_1", content='{"success": true}'))
+    ctx.stm_add(Message(role="tool", name="tc_orphan", content='{"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",
+            content="tool call",
+            metadata={
+                "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", content='{"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"]
+
+
+def test_compress_context_tool_pair_safe_keeps_idless_tool_context() -> None:
+    ctx = Context(config=Config())
+    ctx.stm_add(
+        Message(
+            role="assistant",
+            content="tool call without id",
+            metadata={"tool_calls": [{"name": "demo_tool", "arguments": {"x": 1}}]},
+        )
+    )
+    ctx.stm_add(Message(role="tool", name="demo_tool", content='{"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.metadata.get("tool_calls", [])
+    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",
+            content="mixed tool calls",
+            metadata={
+                "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", content='{"success": true}'))
+    ctx.stm_add(Message(role="tool", name="demo_tool", content='{"success": true}'))
+    ctx.stm_add(Message(role="tool", name="tc_orphan", content='{"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",
+            content="mixed tool calls",
+            id="assistant-state",
+            mark=MessageMark.PERSISTENT,
+            metadata={
+                "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", content='{"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"]
+
+
+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", content=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", content=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.content for message in after_messages] == [
+        message.content 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",
+            content="keep identity",
+            id="assistant-1",
+            mark=MessageMark.PERSISTENT,
+        )
+    )
+    ctx.stm_add(Message(role="user", content="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_max_messages_preserves_protected_marks() -> None:
+    ctx = Context(config=Config())
+    ctx.stm_add(
+        Message(
+            role="system",
+            content="immutable",
+            id="imm-1",
+            mark=MessageMark.IMMUTABLE,
+        )
+    )
+    ctx.stm_add(
+        Message(
+            role="assistant",
+            content="persistent",
+            id="persist-1",
+            mark=MessageMark.PERSISTENT,
+        )
+    )
+    for idx in range(4):
+        ctx.stm_add(Message(role="user", content=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 6fdc258d..f08e5b34 100644
--- a/tests/unit/test_context_implementation.py
+++ b/tests/unit/test_context_implementation.py
@@ -111,6 +111,20 @@ 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():
     ltm = _FakeRetrieval([Message(role="assistant", content="ltm-hit")])
     knowledge = _FakeRetrieval([Message(role="assistant", content="knowledge-hit")])
@@ -426,3 +440,59 @@ def test_context_assemble_handles_overflowing_numeric_ratio_config() -> None:
     contents = [message.content for message in assembled.messages]
     assert contents == ["query", "ltm-hit"]
     assert assembled.metadata["retrieval"]["ltm_count"] == 1
+
+
+def test_context_compress_max_messages_uses_backend_compress_only(monkeypatch: pytest.MonkeyPatch) -> None:
+    stm = _CompressionRecordingSTM(
+        [
+            Message(role="user", content="m0"),
+            Message(role="assistant", content="m1"),
+            Message(role="user", content="m2"),
+        ]
+    )
+    ctx = Context(config=Config(), short_term_memory=stm)
+
+    def _unexpected_compress_context(*args: object, **kwargs: object) -> None:
+        _ = (args, kwargs)
+        raise AssertionError("compress_context should not be called for basic max_messages compression")
+
+    monkeypatch.setattr(
+        "dare_framework.compression.core.compress_context",
+        _unexpected_compress_context,
+    )
+
+    ctx.compress(max_messages=2)
+
+    assert len(stm.compress_calls) == 1
+    assert stm.compress_calls[0].get("max_messages") == 2
+    assert [message.content for message in ctx.stm_get()] == ["m1", "m2"]
+
+
+def test_context_compress_advanced_path_preserves_backend_semantics(monkeypatch: pytest.MonkeyPatch) -> None:
+    stm = _CompressionRecordingSTM(
+        [
+            Message(role="user", content="m0"),
+            Message(role="assistant", content="m1"),
+            Message(role="user", content="m2"),
+        ]
+    )
+    ctx = Context(config=Config(), short_term_memory=stm)
+    calls: list[dict[str, object]] = []
+    stm_sizes_seen_by_strategy: list[int] = []
+
+    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,
+    )
+
+    ctx.compress(max_messages=2, target_tokens=100, strategy="truncate", tool_pair_safe=True)
+
+    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.content for message in ctx.stm_get()] == ["m1", "m2"]
diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index da56c571..f8851d8a 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -7,6 +7,7 @@
 from dare_framework.agent.react_agent import ReactAgent
 from dare_framework.config import Config
 from dare_framework.context import Context
+from dare_framework.context.smartcontext import SmartContext
 from dare_framework.model.types import ModelInput, ModelResponse
 from dare_framework.tool.types import CapabilityDescriptor, CapabilityType, ToolResult
 from dare_framework.transport import TransportEventType
@@ -151,6 +152,51 @@ 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
+
+    async def generate(self, model_input: ModelInput, *, options: Any | None = None) -> ModelResponse:
+        _ = (model_input, options)
+        self._idx += 1
+        return ModelResponse(
+            content="keep going",
+            tool_calls=[
+                {
+                    "id": f"tc_{self._idx}",
+                    "name": "tool:echo",
+                    "arguments": {"value": f"ping-{self._idx}"},
+                }
+            ],
+        )
+
+
 @pytest.mark.asyncio
 async def test_react_agent_prefers_injected_gateway_over_context_gateway() -> None:
     context_gateway = _RecordingGateway("context")
@@ -317,3 +363,132 @@ async def test_react_agent_emits_terminal_message_for_max_round_exit() -> None:
     last_envelope = transport.sent[-1]
     assert getattr(last_envelope, "event_type", None) == TransportEventType.MESSAGE.value
     assert "达到最大轮次" in str(getattr(last_envelope, "payload", {}).get("resp", {}).get("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())
+    gateway = _RecordingGateway("injected")
+    transport = _RecordingTransport()
+    agent = ReactAgent(
+        name="react-test-loop-guard-terminal-message",
+        model=_RepeatingToolModel(),
+        context=context,
+        tool_gateway=gateway,
+        max_tool_rounds=10,
+    )
+
+    result = await agent.execute("test loop guard", transport=transport)
+
+    assert result.success is True
+    assert getattr(transport.sent[-1], "event_type", None) == TransportEventType.MESSAGE.value
+    terminal_payload = getattr(transport.sent[-1], "payload", {})
+    assert "连续重复调用相同工具" in terminal_payload.get("resp", {}).get("output", "")
+
+
+@pytest.mark.asyncio
+async def test_react_agent_max_round_exit_emits_terminal_message_event() -> None:
+    context = Context(config=Config())
+    gateway = _RecordingGateway("injected")
+    transport = _RecordingTransport()
+    agent = ReactAgent(
+        name="react-test-max-round-terminal-message",
+        model=_NonConvergingToolModel(),
+        context=context,
+        tool_gateway=gateway,
+        max_tool_rounds=2,
+    )
+
+    result = await agent.execute("test max rounds", transport=transport)
+
+    assert result.success is True
+    assert getattr(transport.sent[-1], "event_type", None) == TransportEventType.MESSAGE.value
+    terminal_payload = getattr(transport.sent[-1], "payload", {})
+    assert "未收敛" in terminal_payload.get("resp", {}).get("output", "")
diff --git a/tests/unit/test_transport_adapters.py b/tests/unit/test_transport_adapters.py
index da521088..9a529fae 100644
--- a/tests/unit/test_transport_adapters.py
+++ b/tests/unit/test_transport_adapters.py
@@ -254,3 +254,26 @@ async def test_stdio_receiver_handles_canonical_thinking_event(capsys) -> None:
 
     captured = capsys.readouterr()
     assert "Assistant: need tool data" in captured.out
+
+
+@pytest.mark.asyncio
+async def test_stdio_receiver_renders_structured_status_phase_from_resp(capsys) -> None:
+    channel = StdioClientChannel()
+    receiver = channel.agent_envelope_receiver()
+
+    await receiver(
+        TransportEnvelope(
+            id="evt-status",
+            kind=EnvelopeKind.MESSAGE,
+            event_type=TransportEventType.STATUS.value,
+            payload={
+                "kind": "message",
+                "target": "agent",
+                "ok": True,
+                "resp": {"phase": "before_tool"},
+            },
+        )
+    )
+
+    captured = capsys.readouterr()
+    assert "Assistant: before_tool" in captured.out