From 954c91fe73ebc6c95c74fbfbb383038a243d7cd4 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Mon, 2 Mar 2026 11:54:11 +0800
Subject: [PATCH 01/11] feat(agentscope): deliver D5 safe compression slice

Implement the AgentScope D5 gap slice for context compression safety and budget convergence, with docs-first/OpenSpec artifacts and evidence synced.

Key changes:\n- route Context.compress through dare_framework.compression.core.compress_context for unified behavior\n- add token-aware trimming (target_tokens), tool pair safety enforcement, and compression strategy metadata annotation\n- add pre-model auto-compression trigger in ReactAgent with configurable ratios and preserved default backward-compatible behavior\n- add D5 regression tests for tool pair safety, token-budget trimming, and auto-compress trigger timing\n- add OpenSpec change artifacts and feature evidence doc; update project/domain claim ledgers to owner zts212653 with D5 execution evidence

Rationale:\nAgentScope compatibility requires compression that does not orphan tool call/result pairs and can converge by token budget before model invocation. This commit closes that gap while keeping legacy execution unchanged when auto-compress is disabled.
---
 dare_framework/agent/react_agent.py           | 112 +++++++++++---
 dare_framework/compression/core.py            | 145 +++++++++++++++++-
 dare_framework/context/context.py             |   5 +-
 .../agentscope-d5-safe-compression.md         |  61 ++++++++
 .../agentscope_domain_execution_todos.md      |  32 ++--
 docs/todos/project_overall_todos.md           |   8 +-
 .../.openspec.yaml                            |   2 +
 .../agentscope-d5-safe-compression/design.md  |  50 ++++++
 .../proposal.md                               |  33 ++++
 .../specs/agentscope-safe-compression/spec.md |  26 ++++
 .../specs/chat-runtime/spec.md                |  12 ++
 .../agentscope-d5-safe-compression/tasks.md   |  21 +++
 tests/unit/test_context_compression.py        |  74 +++++++++
 .../test_react_agent_gateway_injection.py     |  56 +++++++
 14 files changed, 591 insertions(+), 46 deletions(-)
 create mode 100644 docs/features/agentscope-d5-safe-compression.md
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/.openspec.yaml
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/design.md
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/proposal.md
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/specs/agentscope-safe-compression/spec.md
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/specs/chat-runtime/spec.md
 create mode 100644 openspec/changes/agentscope-d5-safe-compression/tasks.md
 create mode 100644 tests/unit/test_context_compression.py

diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index 20884391..f42ab0bb 100644
--- a/dare_framework/agent/react_agent.py
+++ b/dare_framework/agent/react_agent.py
@@ -113,6 +113,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 +126,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 +176,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,
@@ -571,6 +563,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 +749,24 @@ 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 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..9e5e168a 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -92,6 +92,125 @@ 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:
+        trimmed.pop(0)
+        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()
+    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())
+
+        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,
+                )
+            )
+        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 and tool_id not in retained_call_ids:
+                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,
+    )
+    return messages
+
+
 def compress_context(
     context: IContext,
     *,
@@ -114,8 +233,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 +256,12 @@ def compress_context(
     if not messages:
         return
 
+    if max_messages is None:
+        max_messages = len(messages)
+
     # strategy 默认为 "truncate"，后续可扩展更多策略。
     strategy = options.get("strategy", "truncate")
+    tool_pair_safe = bool(options.get("tool_pair_safe", False))
 
     removed_total = 0
 
@@ -153,10 +284,20 @@ def compress_context(
         removed_total += len(messages) - max_messages
         messages = messages[-max_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..85ff4e8e 100644
--- a/dare_framework/context/context.py
+++ b/dare_framework/context/context.py
@@ -178,8 +178,9 @@ 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
+
+        compress_context(self, **options)
 
 
 class DefaultAssembledContext(IAssembleContext):
diff --git a/docs/features/agentscope-d5-safe-compression.md b/docs/features/agentscope-d5-safe-compression.md
new file mode 100644
index 00000000..8e433512
--- /dev/null
+++ b/docs/features/agentscope-d5-safe-compression.md
@@ -0,0 +1,61 @@
+---
+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: draft
+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）。
+- 待完成：提交评审与合并门禁记录补充。
+
+## 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
+- 待 D5 切片提交 PR 后补充。
diff --git a/docs/todos/agentscope_domain_execution_todos.md b/docs/todos/agentscope_domain_execution_todos.md
index d2f37c9a..e95b3439 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。 |
+| 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..c2cf8fdf 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）。 |
+| 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..fb95cfd0
--- /dev/null
+++ b/tests/unit/test_context_compression.py
@@ -0,0 +1,74 @@
+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
+
+
+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_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
diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index da56c571..c2da0af4 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -151,6 +151,22 @@ 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 _FinalOnlyModel:
+    async def generate(self, model_input: ModelInput, *, options: Any | None = None) -> ModelResponse:
+        _ = (model_input, options)
+        return ModelResponse(content="final", tool_calls=[])
+
+
 @pytest.mark.asyncio
 async def test_react_agent_prefers_injected_gateway_over_context_gateway() -> None:
     context_gateway = _RecordingGateway("context")
@@ -317,3 +333,43 @@ 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", ""))
+
+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_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 == []

From ca8b061f01b7c168f916d6488382e320d82f1ae1 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Mon, 2 Mar 2026 11:55:52 +0800
Subject: [PATCH 02/11] docs(agentscope): sync D5 PR review ledger links

Synchronize D5 documentation state after PR creation so TODO claims and feature evidence stay consistent with repository workflow governance.

Key changes:\n- update D5 claim notes in both overall and domain TODO ledgers from 'pending PR' to 'PR #136 in review'\n- move feature aggregation status to in_review and mark PR submission complete\n- add explicit PR link and merge-gate placeholder in evidence section

Rationale:\nThe claim-ledger policy requires explicit ownership and current execution state to avoid parallel conflict. These updates keep docs as the canonical source of truth while PR #136 is under review.
---
 docs/features/agentscope-d5-safe-compression.md | 8 +++++---
 docs/todos/agentscope_domain_execution_todos.md | 2 +-
 docs/todos/project_overall_todos.md             | 2 +-
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/docs/features/agentscope-d5-safe-compression.md b/docs/features/agentscope-d5-safe-compression.md
index 8e433512..512516e5 100644
--- a/docs/features/agentscope-d5-safe-compression.md
+++ b/docs/features/agentscope-d5-safe-compression.md
@@ -4,7 +4,7 @@ doc_kind: feature
 topics: ["agentscope", "compression", "context", "react-agent", "budget"]
 created: 2026-03-02
 updated: 2026-03-02
-status: draft
+status: in_review
 mode: openspec
 ---
 
@@ -24,7 +24,8 @@ mode: openspec
 ## Progress
 - 已完成：D5 代码实现（tool-pair-safe、token-aware compression、ReAct pre-model auto-compress）。
 - 已完成：OpenSpec tasks 全部打勾（10/10）。
-- 待完成：提交评审与合并门禁记录补充。
+- 已完成：提交 PR #136，进入评审阶段。
+- 待完成：评审反馈处理与合并门禁记录闭环。
 
 ## Evidence
 
@@ -58,4 +59,5 @@ mode: openspec
 - 回滚：保持自动压缩开关可控，必要时回退到手动压缩路径。
 
 ### Review and Merge Gate Links
-- 待 D5 切片提交 PR 后补充。
+- 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 e95b3439..7d2e6f97 100644
--- a/docs/todos/agentscope_domain_execution_todos.md
+++ b/docs/todos/agentscope_domain_execution_todos.md
@@ -22,7 +22,7 @@
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
 | 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。 |
+| 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。 |
 
diff --git a/docs/todos/project_overall_todos.md b/docs/todos/project_overall_todos.md
index c2cf8fdf..46307e1d 100644
--- a/docs/todos/project_overall_todos.md
+++ b/docs/todos/project_overall_todos.md
@@ -16,7 +16,7 @@
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
 | 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）。 |
+| 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。 |
 

From fb52b56a007462d75efdbdb025d192de64ad6073 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Mon, 2 Mar 2026 15:16:23 +0800
Subject: [PATCH 03/11] fix(compression,react-agent): preserve unbounded trim
 semantics and emit terminal events on non-converged exits

Address new Codex review findings on PR #136.\n\n1) compression max_messages sentinel:\n accepted negative max_messages as an unbounded sentinel, but token-aware compression path later reused the negative value in list slicing (), dropping history unexpectedly. The fix normalizes negative max_messages to current message count before count-based trim.\n\n2) ReactAgent non-converged terminal transport signaling:\nDirect  emitted terminal MESSAGE only on normal no-tool completion, not on repeated-tool guard or max-round fallback. Stream consumers could receive intermediate events without a completion signal. The fix introduces a shared terminal MESSAGE helper and uses it for all terminal exits while still suppressing duplicates under transport-loop execution.\n\nTests added to prove both regressions and lock behavior.
---
 dare_framework/agent/react_agent.py           | 18 ++++++
 dare_framework/compression/core.py            |  3 +
 tests/unit/test_context_compression.py        | 19 ++++++
 .../test_react_agent_gateway_injection.py     | 61 +++++++++++++++++++
 4 files changed, 101 insertions(+)

diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index f42ab0bb..68c1b98a 100644
--- a/dare_framework/agent/react_agent.py
+++ b/dare_framework/agent/react_agent.py
@@ -563,6 +563,24 @@ async def _execute_with_smart_context(
             output_text=final_message,
         )
 
+    async def _emit_terminal_transport_message(
+        self,
+        *,
+        transport: AgentChannel | None,
+        output: str,
+    ) -> None:
+        """Emit terminal MESSAGE for direct execute() transport calls."""
+        # BaseAgent emits a terminal RESULT envelope for transport-loop executions,
+        # so ReactAgent should not emit an extra terminal MESSAGE event in that path.
+        if self._is_transport_loop_execution(transport=transport):
+            return
+        await self._emit_transport_success(
+            transport=transport,
+            event_type=TransportEventType.MESSAGE.value,
+            target="prompt",
+            resp={"output": output},
+        )
+
     def _build_model_messages(self, assembled: Any) -> list[Message]:
         """Build model-facing messages including system prompt and plan state injection."""
         messages = list(assembled.messages)
diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index 9e5e168a..5a3eb7a9 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -258,6 +258,9 @@ def compress_context(
 
     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")
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
index fb95cfd0..ac387855 100644
--- a/tests/unit/test_context_compression.py
+++ b/tests/unit/test_context_compression.py
@@ -72,3 +72,22 @@ def test_compress_context_target_tokens_trims_long_history() -> None:
 
     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
+    ]
diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index c2da0af4..43ab6b8a 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -167,6 +167,25 @@ async def generate(self, model_input: ModelInput, *, options: Any | None = None)
         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")
@@ -373,3 +392,45 @@ async def test_react_agent_without_auto_compress_keeps_legacy_behavior() -> None
 
     assert result.success is True
     assert context.compress_calls == []
+
+
+@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", "")

From f00abe76c7fdc15e15d012819b4e8fe0a92ea536 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Mon, 2 Mar 2026 15:43:57 +0800
Subject: [PATCH 04/11] fix(compression): preserve id-less tool context and
 status phase rendering

Address unresolved PR #136 review findings that affect compression safety and\ntransport status output.\n\nKey changes:\n- Keep assistant tool calls that do not provide an id during tool-pair safety\n  filtering, and avoid dropping tool messages when id-less calls exist.\n- Update status rendering so non-approval structured payloads in resp (e.g.\n  {"phase": ...}) are rendered as phase/event text instead of forced\n  "approval update".\n- Add regression tests for id-less tool call/tool result preservation under\n  tool_pair_safe mode.\n- Add regression test for stdio status rendering from resp.phase payloads.\n\nRationale:\nOpenAI-compatible and other adapters may omit tool call ids while still emitting\nvalid tool call/result context. Compression should not erase that context.\nLikewise, status envelopes should display canonical progress fields instead of\nbeing coerced to approval-only text.
---
 dare_framework/compression/core.py            | 10 ++++++++
 .../transport/_internal/adapters.py           |  9 +++++++-
 tests/unit/test_context_compression.py        | 21 +++++++++++++++++
 tests/unit/test_transport_adapters.py         | 23 +++++++++++++++++++
 4 files changed, 62 insertions(+), 1 deletion(-)

diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index 5a3eb7a9..fb5c35ec 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -144,6 +144,7 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
 
     updated_messages: list[Message] = []
     retained_call_ids: set[str] = set()
+    has_idless_tool_calls = False
     changes = 0
     for message in messages:
         if message.role != "assistant":
@@ -162,6 +163,12 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
             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
+                # preserve tool messages instead of dropping context pairs heuristically.
+                filtered_calls.append(call)
+                has_idless_tool_calls = True
 
         if len(filtered_calls) != len(raw_calls):
             changes += len(raw_calls) - len(filtered_calls)
@@ -182,6 +189,9 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
     final_messages: list[Message] = []
     for message in updated_messages:
         if message.role == "tool":
+            if has_idless_tool_calls:
+                final_messages.append(message)
+                continue
             tool_id = message.name.strip() if isinstance(message.name, str) else ""
             if tool_id and tool_id not in retained_call_ids:
                 changes += 1
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/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
index ac387855..7d026de2 100644
--- a/tests/unit/test_context_compression.py
+++ b/tests/unit/test_context_compression.py
@@ -61,6 +61,27 @@ def test_compress_context_tool_pair_safe_removes_unmatched_tool_call_ids() -> No
     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_target_tokens_trims_long_history() -> None:
     ctx = Context(config=Config())
     for idx in range(8):
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

From 09d6a314befade936d9d78b016473d11cd110b5a Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Mon, 2 Mar 2026 17:02:15 +0800
Subject: [PATCH 05/11] fix: prune orphan tool results in mixed id tool
 histories

Address the new PR #136 review finding on tool-pair safety behavior in mixed id/id-less histories.\n\nKey changes:\n- Updated _enforce_tool_pair_safety to retain tool results by explicit call-id matches and by id-less call tool names, instead of preserving all tool results whenever any id-less call exists.\n- This keeps valid id-less tool context while removing stale orphan tool results from unrelated turns.\n\nRegression coverage:\n- Added a compression regression test for mixed assistant tool_calls (id-based + id-less) verifying matched results are preserved and orphan tool results are pruned.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_context_compression.py tests/unit/test_openai_model_adapter.py tests/unit/test_transport_adapters.py
---
 dare_framework/compression/core.py     | 17 +++++++++++------
 tests/unit/test_context_compression.py | 26 ++++++++++++++++++++++++++
 2 files changed, 37 insertions(+), 6 deletions(-)

diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index fb5c35ec..b908cc75 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -144,7 +144,7 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
 
     updated_messages: list[Message] = []
     retained_call_ids: set[str] = set()
-    has_idless_tool_calls = False
+    retained_idless_tool_names: set[str] = set()
     changes = 0
     for message in messages:
         if message.role != "assistant":
@@ -166,9 +166,11 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
                 continue
             if not isinstance(tool_id, str) or not tool_id.strip():
                 # Some providers emit tool calls without stable ids. Keep these calls and
-                # preserve tool messages instead of dropping context pairs heuristically.
+                # retain matching tool results by tool name.
                 filtered_calls.append(call)
-                has_idless_tool_calls = True
+                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)
@@ -189,11 +191,14 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
     final_messages: list[Message] = []
     for message in updated_messages:
         if message.role == "tool":
-            if has_idless_tool_calls:
+            tool_id = message.name.strip() if isinstance(message.name, str) else ""
+            if tool_id in retained_call_ids:
                 final_messages.append(message)
                 continue
-            tool_id = message.name.strip() if isinstance(message.name, str) else ""
-            if tool_id and tool_id not in retained_call_ids:
+            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)
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
index 7d026de2..c1cc1400 100644
--- a/tests/unit/test_context_compression.py
+++ b/tests/unit/test_context_compression.py
@@ -82,6 +82,32 @@ def test_compress_context_tool_pair_safe_keeps_idless_tool_context() -> None:
     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_target_tokens_trims_long_history() -> None:
     ctx = Context(config=Config())
     for idx in range(8):

From 2cfbcbbf3c269b61133576c45489c2ea92ca523c Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 10:32:33 +0800
Subject: [PATCH 06/11] test: restore asyncio marker after rebase conflict
 resolution

During the #136 rebase onto latest main, conflict resolution in test_react_agent_gateway_injection.py dropped the @pytest.mark.asyncio decorator on test_react_agent_auto_compress_triggers_before_model_call.\n\nThis commit restores the decorator so pytest executes the async test correctly under the existing asyncio plugin configuration.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_context_compression.py tests/unit/test_openai_model_adapter.py tests/unit/test_transport_adapters.py tests/unit/test_react_agent_gateway_injection.py\n- Result: 33 passed.
---
 tests/unit/test_react_agent_gateway_injection.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index 43ab6b8a..3401c26a 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -353,6 +353,8 @@ async def test_react_agent_emits_terminal_message_for_max_round_exit() -> None:
     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

From 11647cae80176be4a733ca5cf1b2f415ccd4ca93 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 10:58:05 +0800
Subject: [PATCH 07/11] fix(context): preserve smart compression semantics and
 smart-context auto compress

Address new PR #136 review feedback around compression identity semantics, STM backend behavior, and SmartContext parity.\n\nKey changes:\n- Preserve message identity on compression annotation by retaining message id/mark when rewriting the head message metadata.\n- Keep IMMUTABLE/PERSISTENT messages during max_messages and token trimming; when only protected messages remain, stop trimming safely.\n- Fix the keep_temporary=0 edge case in max_messages truncation (avoid accidental temporary retention from list[-0:]).\n- Update Context.compress to preserve backend STM compress semantics for max_messages-only calls and to pre-apply backend count-based compression before advanced strategy compression.\n- Add auto-compress trigger in ReactAgent SmartContext execution path and rebuild ordered model messages post-compression while retaining injected reflection prompts.\n\nTests:\n- Added regression coverage for identity preservation, protected mark retention, Context.compress backend semantics, and SmartContext auto-compress behavior.\n- Verified via targeted pytest run for compression/context/react-agent suites.
---
 dare_framework/agent/react_agent.py           | 21 ++++++
 dare_framework/compression/core.py            | 50 ++++++++++++--
 dare_framework/context/context.py             | 20 ++++++
 tests/unit/test_context_compression.py        | 53 ++++++++++++++-
 tests/unit/test_context_implementation.py     | 68 +++++++++++++++++++
 .../test_react_agent_gateway_injection.py     | 33 +++++++++
 6 files changed, 240 insertions(+), 5 deletions(-)

diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index 68c1b98a..a582967d 100644
--- a/dare_framework/agent/react_agent.py
+++ b/dare_framework/agent/react_agent.py
@@ -395,10 +395,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:
diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index b908cc75..02877239 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:
@@ -111,7 +111,19 @@ def _trim_to_target_tokens(messages: List[Message], target_tokens: int | None) -
     trimmed = list(messages)
     removed = 0
     while len(trimmed) > 1 and _estimate_tokens(trimmed) > target_tokens:
-        trimmed.pop(0)
+        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
 
@@ -222,6 +234,8 @@ def _annotate_strategy(messages: List[Message], strategy: str) -> List[Message]:
         content=head.content,
         name=head.name,
         metadata=metadata,
+        mark=getattr(head, "mark", MessageMark.TEMPORARY),
+        id=getattr(head, "id", None),
     )
     return messages
 
@@ -299,8 +313,36 @@ 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)
diff --git a/dare_framework/context/context.py b/dare_framework/context/context.py
index 85ff4e8e..ac2b6985 100644
--- a/dare_framework/context/context.py
+++ b/dare_framework/context/context.py
@@ -180,6 +180,26 @@ def compress(self, **options: Any) -> None:
         """Compress context to fit within budget."""
         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):
+            if not has_advanced_options:
+                compress_impl(max_messages=max_messages)
+                return
+            if max_messages is not None:
+                compress_impl(max_messages=max_messages)
+
         compress_context(self, **options)
 
 
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
index c1cc1400..c17c45d0 100644
--- a/tests/unit/test_context_compression.py
+++ b/tests/unit/test_context_compression.py
@@ -2,7 +2,7 @@
 
 from dare_framework.compression.core import compress_context
 from dare_framework.config import Config
-from dare_framework.context import Context, Message
+from dare_framework.context import Context, Message, MessageMark
 
 
 def _tool_ids(message: Message) -> list[str]:
@@ -138,3 +138,54 @@ def test_compress_context_negative_max_messages_keeps_unbounded_semantics() -> N
     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..d9996ab8 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,57 @@ 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]] = []
+
+    def _record_compress_context(context: Context, **kwargs: object) -> None:
+        _ = context
+        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 [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 3401c26a..4839dd6f 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
@@ -161,6 +162,16 @@ def compress(self, **options: Any) -> None:
         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)
@@ -396,6 +407,28 @@ async def test_react_agent_without_auto_compress_keeps_legacy_behavior() -> None
     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())

From b918558b1b275562c876c9994561f843181d7051 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 11:48:59 +0800
Subject: [PATCH 08/11] fix(compression): preserve assistant id/mark during
 tool-pair filtering

Address remaining PR #136 review concern in tool_pair_safe reconstruction path.\n\nProblem:\n- _enforce_tool_pair_safety rebuilt assistant messages after filtering tool_calls but dropped id/mark.\n- In SmartContext and identity-sensitive flows this could silently break message identity, downgrade protected marks, and impact later retention/ordering behavior.\n\nChanges:\n- Preserve assistant message id and mark when rebuilding filtered assistant messages in dare_framework/compression/core.py.\n- Add regression test to ensure id/mark survive tool_call filtering while unmatched tool_call IDs are still removed.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q -p no:cacheprovider tests/unit/test_context_compression.py tests/unit/test_react_agent_gateway_injection.py\n- Result: 21 passed, 1 warning.
---
 dare_framework/compression/core.py     |  2 ++
 tests/unit/test_context_compression.py | 26 ++++++++++++++++++++++++++
 2 files changed, 28 insertions(+)

diff --git a/dare_framework/compression/core.py b/dare_framework/compression/core.py
index 02877239..a52b246c 100644
--- a/dare_framework/compression/core.py
+++ b/dare_framework/compression/core.py
@@ -194,6 +194,8 @@ def _enforce_tool_pair_safety(messages: List[Message]) -> Tuple[List[Message], i
                     content=message.content,
                     name=message.name,
                     metadata=metadata,
+                    mark=getattr(message, "mark", MessageMark.TEMPORARY),
+                    id=getattr(message, "id", None),
                 )
             )
         else:
diff --git a/tests/unit/test_context_compression.py b/tests/unit/test_context_compression.py
index c17c45d0..93f6808c 100644
--- a/tests/unit/test_context_compression.py
+++ b/tests/unit/test_context_compression.py
@@ -108,6 +108,32 @@ def test_compress_context_tool_pair_safe_drops_orphan_tool_results_with_mixed_id
     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):

From f54cb85ca864c19707a9d047e4f581c2653c455e Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 12:11:48 +0800
Subject: [PATCH 09/11] fix(context): run advanced compression before backend
 max trim

Address new PR #136 review item about advanced compression ordering.\n\nProblem:\n- Context.compress applied backend STM max_messages compression before advanced strategy compression when advanced options were present.\n- Strategies that depend on full history (e.g. summary_preview) would receive already-trimmed context and degrade to plain truncation behavior.\n\nChanges:\n- Keep basic path unchanged: max_messages-only calls still delegate directly to backend compress.\n- For advanced path: run compress_context first, then apply backend compress(max_messages=...) as a retention guardrail.\n- This preserves backend semantics while allowing advanced strategies to operate on full pre-trim history.\n\nTests:\n- Updated advanced-path unit test to assert strategy sees the full STM size before backend trimming, while backend semantics still apply after strategy execution.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q -p no:cacheprovider tests/unit/test_context_implementation.py tests/unit/test_context_compression.py tests/unit/test_react_agent_gateway_injection.py\n- Result: 42 passed, 1 warning.
---
 dare_framework/context/context.py         | 14 ++++++++------
 tests/unit/test_context_implementation.py |  4 +++-
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/dare_framework/context/context.py b/dare_framework/context/context.py
index ac2b6985..76ff575f 100644
--- a/dare_framework/context/context.py
+++ b/dare_framework/context/context.py
@@ -193,15 +193,17 @@ def compress(self, **options: Any) -> None:
             if isinstance(raw_max_messages, int) and raw_max_messages >= 0
             else None
         )
-        if callable(compress_impl):
-            if not has_advanced_options:
-                compress_impl(max_messages=max_messages)
-                return
-            if max_messages is not None:
-                compress_impl(max_messages=max_messages)
+        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):
     """Default context assembly strategy.
diff --git a/tests/unit/test_context_implementation.py b/tests/unit/test_context_implementation.py
index d9996ab8..f08e5b34 100644
--- a/tests/unit/test_context_implementation.py
+++ b/tests/unit/test_context_implementation.py
@@ -478,9 +478,10 @@ def test_context_compress_advanced_path_preserves_backend_semantics(monkeypatch:
     )
     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:
-        _ = context
+        stm_sizes_seen_by_strategy.append(len(context.stm_get()))
         calls.append(dict(kwargs))
 
     monkeypatch.setattr(
@@ -493,4 +494,5 @@ def _record_compress_context(context: Context, **kwargs: object) -> None:
     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"]

From dfcaf4573c0bc295c65901c2152dafeba6923691 Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 12:44:25 +0800
Subject: [PATCH 10/11] refactor(react-agent): remove duplicate terminal
 transport emitter

Address new PR #136 review comment about duplicate method definition.\n\nChanges:\n- Remove the earlier duplicate  definition in ReactAgent.\n- Keep the later implementation as the single canonical runtime path.\n\nRationale:\n- Avoid dead-code shadowing where edits to the first definition have no runtime effect.\n- Reduce maintenance risk and prevent behavior drift confusion.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q -p no:cacheprovider tests/unit/test_react_agent_gateway_injection.py tests/unit/test_context_implementation.py\n- Result: 33 passed, 1 warning.
---
 dare_framework/agent/react_agent.py | 18 ------------------
 1 file changed, 18 deletions(-)

diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index a582967d..94c28277 100644
--- a/dare_framework/agent/react_agent.py
+++ b/dare_framework/agent/react_agent.py
@@ -584,24 +584,6 @@ async def _execute_with_smart_context(
             output_text=final_message,
         )
 
-    async def _emit_terminal_transport_message(
-        self,
-        *,
-        transport: AgentChannel | None,
-        output: str,
-    ) -> None:
-        """Emit terminal MESSAGE for direct execute() transport calls."""
-        # BaseAgent emits a terminal RESULT envelope for transport-loop executions,
-        # so ReactAgent should not emit an extra terminal MESSAGE event in that path.
-        if self._is_transport_loop_execution(transport=transport):
-            return
-        await self._emit_transport_success(
-            transport=transport,
-            event_type=TransportEventType.MESSAGE.value,
-            target="prompt",
-            resp={"output": output},
-        )
-
     def _build_model_messages(self, assembled: Any) -> list[Message]:
         """Build model-facing messages including system prompt and plan state injection."""
         messages = list(assembled.messages)

From 30a1fb886a04875e81c7b3f5e28eb9e8e1d691bc Mon Sep 17 00:00:00 2001
From: lang <lang_c@outlook.com>
Date: Tue, 3 Mar 2026 12:59:34 +0800
Subject: [PATCH 11/11] fix(react-agent): guard auto-compress ratios against
 NaN/Inf

Address new PR #136 review comment about non-finite compression ratios.\n\nChanges:\n- In _clamp_ratio, reject non-finite values with math.isfinite and fall back to default ratio.\n- Prevent int(max_tokens * ratio) from raising when ratio is NaN/Inf in auto-compress path.\n- Add regression test covering NaN ratio inputs and default-ratio fallback behavior during auto compression.\n\nVerification:\n- /Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q -p no:cacheprovider tests/unit/test_react_agent_gateway_injection.py tests/unit/test_context_implementation.py\n- Result: 34 passed, 1 warning.
---
 dare_framework/agent/react_agent.py           |  3 +++
 .../test_react_agent_gateway_injection.py     | 23 +++++++++++++++++++
 2 files changed, 26 insertions(+)

diff --git a/dare_framework/agent/react_agent.py b/dare_framework/agent/react_agent.py
index 94c28277..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
 
@@ -783,6 +784,8 @@ def _clamp_ratio(value: Any, *, default: float) -> float:
         ratio = float(value)
     except (TypeError, ValueError):
         return default
+    if not math.isfinite(ratio):
+        return default
     if ratio <= 0:
         return default
     if ratio > 1:
diff --git a/tests/unit/test_react_agent_gateway_injection.py b/tests/unit/test_react_agent_gateway_injection.py
index 4839dd6f..f8851d8a 100644
--- a/tests/unit/test_react_agent_gateway_injection.py
+++ b/tests/unit/test_react_agent_gateway_injection.py
@@ -389,6 +389,29 @@ async def test_react_agent_auto_compress_triggers_before_model_call() -> None:
     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())