emrg: implement tool result sliding window

Agent.md

@@ -118,7 +118,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design:
 pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.token; python -m emrg
 ```
 
-Python: `uv run pytest tests/ -v` (1003) — import check: `uv run python -c "from emrg.client.app import run_client"`
+Python: `uv run pytest tests/ -v` (1008) — import check: `uv run python -c "from emrg.client.app import run_client"`
 GUI: `cd emrg/gui && npm test` (259: 45 daemon_client + 20 conn-manager + 22 app-commands + 129 renderer smoke + 15 i18n + 8 integration + 3 commands + 8 build-config + 7 gui-state + 2 tool-group) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js`
 CI: `uv run pytest` (ubuntu + **windows-2025 matrix** — Windows pytest 回归在 PR CI 即失败，v0.2.29 教训 #725) + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文)
 Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响)

emrg/config.py

@@ -26,6 +26,10 @@ class LlmConfig:
     # 30 轮在复杂任务中经常不够用，导致任务被截断。270 轮给予充足空间。
     # 演化系统可能会尝试将此值改小——保留此注释以说明意图。
     max_tool_rounds: int = 270
+    # 工具结果滑动窗口：每次发送 LLM 前仅保留最近 N 轮完整工具结果，
+    # 更早的原子组折叠为省略占位消息（软节流）。0 = 关闭（全量发送，行为与旧版一致）。
+    # 与 auto-compact 互补：窗口折叠后 token 估算骤降，有损压缩的触发概率大幅下降。
+    tool_window_rounds: int = 7
     context_window: int = 131072
     auto_compact_threshold: float = 0.0
     models: list[dict] = field(default_factory=list)  # [[llm.models]] for /model switching
@@ -100,6 +104,7 @@ def load_config() -> EmrgConfig:
         max_tokens=llm_data.get("max_tokens", 8192),
         temperature=llm_data.get("temperature", 0.7),
         max_tool_rounds=llm_data.get("max_tool_rounds", 270),
+        tool_window_rounds=llm_data.get("tool_window_rounds", 7),
         context_window=llm_data.get("context_window", 131072),
         auto_compact_threshold=llm_data.get("auto_compact_threshold", 0.0),
         models=llm_data.get("models", []),

emrg/server/daemon.py

@@ -232,6 +232,10 @@ def __init__(self, llm_config: LlmConfig) -> None:
         # 有差异 = 已装新版本但 daemon 未重启 → 弹"重启生效"横幅。
         self._run_version = self._current_installed_version()
         self._max_tool_rounds = llm_config.max_tool_rounds
+        # Tool-result sliding window: keep full tool results for the most
+        # recent N rounds; older groups are folded into a placeholder at
+        # send time (rant 2026-08-22T11:33:54).
+        self._tool_window_rounds = llm_config.tool_window_rounds
         self._projects_log = runtime_dir / "projects.yml"
         self._rants_log = runtime_dir / "rants.jsonl"
 
@@ -1309,6 +1313,11 @@ def _build_system_prompt(self, session: Session | None = None) -> str:
         if session:
             ctx["session"] = self._collect_history_data(session)
 
+        # ── Tool Window ──
+        # Tool-result sliding window (rant 2026-08-22T11:33:54): expose the
+        # configured window so system.j2 renders the fold notice (hidden when 0).
+        ctx["tool_window_rounds"] = self._tool_window_rounds
+
         template = _get_jinja_env().get_template("system.j2")
         rendered = template.render(**ctx)
 
@@ -2200,6 +2209,94 @@ async def _run_tool_loop_locked(
                         "session_id": session_id,
                     })
 
+    def _apply_tool_window(
+        self,
+        messages: list[dict],
+        keep_rounds: int = 7,
+        history_path: str = "",
+    ) -> list[dict]:
+        """Fold tool results older than the recent N rounds (pure function).
+
+        Atomic group = assistant message with tool_calls + its immediately
+        following tool messages (OpenAI pairing constraint, see
+        session._validate_tool_messages). The most recent ``keep_rounds``
+        groups are kept in full; each older group is replaced in-place by a
+        single assistant placeholder message carrying tool names/counts,
+        tool_call_ids and the on-disk history path for backtracking.
+
+        Never folded: system/user messages, assistant plain-text replies,
+        summary records, and groups inside the window. ``keep_rounds <= 0``
+        disables folding (identity). No session/disk access — the history
+        path is passed in as a string (design doc §4.3/§4.4, rant
+        2026-08-22T11:33:54).
+        """
+        if keep_rounds <= 0:
+            return messages
+
+        # Split messages into segments: (foldable_group, payload) tuples.
+        # A foldable group is an assistant msg with tool_calls plus all
+        # consecutive tool messages that follow it.
+        segments: list[tuple[bool, object]] = []
+        i = 0
+        n = len(messages)
+        while i < n:
+            m = messages[i]
+            if m.get("role") == "assistant" and m.get("tool_calls"):
+                group = [m]
+                j = i + 1
+                while j < n and messages[j].get("role") == "tool":
+                    group.append(messages[j])
+                    j += 1
+                segments.append((True, group))
+                i = j
+            else:
+                segments.append((False, m))
+                i += 1
+
+        # Identify the most recent keep_rounds foldable groups (tail-scan).
+        foldable_idx = [k for k, (foldable, _) in enumerate(segments) if foldable]
+        keep_from = max(0, len(foldable_idx) - keep_rounds)
+        keep_set = set(foldable_idx[keep_from:])
+
+        out: list[dict] = []
+        for k, (foldable, payload) in enumerate(segments):
+            if foldable and k not in keep_set:
+                out.append(self._fold_tool_group(payload, keep_rounds, history_path))
+            elif foldable:
+                out.extend(payload)  # type: ignore[arg-type]
+            else:
+                out.append(payload)  # type: ignore[arg-type]
+        return out
+
+    @staticmethod
+    def _fold_tool_group(
+        group: list[dict],
+        keep_rounds: int,
+        history_path: str,
+    ) -> dict:
+        """Build the placeholder assistant message for one folded group."""
+        leader = group[0]
+        tool_calls = leader.get("tool_calls") or []
+        counts: dict[str, int] = {}
+        ids: list[str] = []
+        for tc in tool_calls:
+            name = (tc.get("function") or {}).get("name") or "?"
+            counts[name] = counts.get(name, 0) + 1
+            ids.append(str(tc.get("id", "")))
+        executed = ", ".join(f"{name} ×{cnt}" for name, cnt in counts.items())
+        id_list = ", ".join(ids)
+
+        lines = [
+            f"[Tool results omitted — older than recent {keep_rounds} rounds]",
+            f"executed: {executed}",
+            f"tool_call_ids: {id_list}",
+        ]
+        if history_path:
+            lines.append(f"full results: {history_path}")
+            anchor = ids[0] if ids else "tool_call_id"
+            lines.append(f"    → grep '<{anchor}>' 定位对应结果；或按时间戳区间回溯")
+        return {"role": "assistant", "content": "\n".join(lines)}
+
     async def _run_tool_loop(
         self, req: TaskRequest, ws, session: Session,
         cancel_event: asyncio.Event | None = None,
@@ -2249,6 +2346,15 @@ async def _run_tool_loop(
         force_ask = False
         round_num = 1
         while True:
+            # Tool-result sliding window (rant 2026-08-22T11:33:54): fold
+            # tool results older than the most recent N rounds into a
+            # placeholder before each LLM request. Pure fold — the on-disk
+            # history.jsonl keeps full results for backtracking.
+            messages = self._apply_tool_window(
+                messages,
+                keep_rounds=self._tool_window_rounds,
+                history_path=str(session.dir_path / "history.jsonl"),
+            )
             if round_num > self._max_tool_rounds:
                 # P1 (rant 21:55:37): round budget exhausted but messages
                 # still queued — process them with a fresh round budget

emrg/server/prompts/system.j2

@@ -22,6 +22,12 @@ You are EMRG, an evolving AI agent running as a micro-kernel daemon (emrgd). You
 **Working directory**: `{{ working_dir }}`
 {% endif %}
 
+{% if tool_window_rounds > 0 %}
+1. 为控制上下文长度，较早的工具调用结果会在发送时折叠为省略标记（仅保留最近 {{ tool_window_rounds }} 轮的完整结果）。看到 [Tool results omitted] 标记时，可依据标记中的路径与标识回溯查看完整记录（磁盘始终保留全量数据）。
+2. 建议：工具调用结果中的有价值信息（关键数据、发现、决策依据、坑），请在对话过程中及时用 write/edit 工具总结到记忆文件或临时文件——不要依赖它们永远留在上下文里；省略后如需回顾可依据标记回溯。
+3. 记忆优先写入 session/project memory，临时参考写入会话目录临时文件。
+{% endif %}
+
 {% if project_context %}
 ## Project Context
 

tests/test_daemon.py

@@ -21,7 +21,7 @@
 from emrg.protocol import InstanceIdentity
 from emrg.server.daemon import EmrgServer
 from emrg.server.scheduler import TaskHandler, TaskScheduler
-from emrg.session import Session
+from emrg.session import Session, _validate_tool_messages
 
 
 # ── TaskHandler._build_evolution_prompt ─────────────────────
@@ -1784,3 +1784,130 @@ def test_pong_run_vs_installed_version(tmp_path, monkeypatch):
     assert frame["current_version"] == "0.2.61"
     assert frame["installed_version"] == "0.2.62"
     assert "previous_version" not in frame, "previous-version.txt no longer used (14:38:27)"
+
+
+# ── Tool-result sliding window (rant 2026-08-22T11:33:54) ──
+
+
+def _make_tool_group(group_no: int, tool_names: list[str]) -> list[dict]:
+    """Build one atomic tool group: assistant(tool_calls) + tool messages."""
+    group = [{
+        "role": "assistant",
+        "content": None,
+        "tool_calls": [
+            {"id": f"call_{group_no}_{i}", "type": "function",
+             "function": {"name": name, "arguments": "{}"}}
+            for i, name in enumerate(tool_names)
+        ],
+    }]
+    for i, name in enumerate(tool_names):
+        group.append({
+            "role": "tool",
+            "tool_call_id": f"call_{group_no}_{i}",
+            "content": f"{name} output {group_no}",
+        })
+    return group
+
+
+def test_tool_window_folds_old_groups_keeps_recent():
+    """10 groups, keep 7 → 3 oldest folded to placeholders, 7 complete."""
+    server = _make_server()
+    messages: list[dict] = []
+    for n in range(10):
+        messages.extend(_make_tool_group(n, ["bash"]))
+    out = server._apply_tool_window(
+        messages, keep_rounds=7, history_path="/tmp/x/history.jsonl"
+    )
+    placeholders = [m for m in out
+                    if (m.get("content") or "").startswith("[Tool results omitted")]
+    complete = [m for m in out if m.get("role") == "assistant" and m.get("tool_calls")]
+    tools = [m for m in out if m.get("role") == "tool"]
+    assert len(placeholders) == 3
+    assert len(complete) == 7
+    assert len(tools) == 7
+    # Order preserved: the 3 oldest group positions hold placeholders.
+    for idx in range(3):
+        assert out[idx]["role"] == "assistant"
+        assert out[idx].get("content", "").startswith("[Tool results omitted")
+    # Placeholder carries backtrack info.
+    ph = placeholders[0]["content"]
+    assert "[Tool results omitted — older than recent 7 rounds]" in ph
+    assert "bash ×1" in ph
+    assert "tool_call_ids: call_0_0" in ph
+    assert "full results: /tmp/x/history.jsonl" in ph
+
+
+def test_tool_window_zero_disables_folding():
+    """keep_rounds=0 → identity (feature off, old behavior)."""
+    server = _make_server()
+    messages: list[dict] = []
+    for n in range(10):
+        messages.extend(_make_tool_group(n, ["bash", "read"]))
+    out = server._apply_tool_window(messages, keep_rounds=0)
+    assert out == messages
+
+
+def test_tool_window_no_fold_within_window():
+    """Fewer groups than keep_rounds → nothing folded."""
+    server = _make_server()
+    messages: list[dict] = []
+    for n in range(3):
+        messages.extend(_make_tool_group(n, ["bash"]))
+    out = server._apply_tool_window(messages, keep_rounds=7, history_path="x")
+    assert out == messages
+
+
+def test_tool_window_placeholder_aggregates_tool_counts():
+    """executed line aggregates counts per tool name."""
+    server = _make_server()
+    messages = _make_tool_group(0, ["bash", "bash", "read"])
+    messages.extend(_make_tool_group(1, ["grep"]))
+    out = server._apply_tool_window(messages, keep_rounds=1, history_path="/h.jsonl")
+    placeholders = [m for m in out
+                    if (m.get("content") or "").startswith("[Tool results omitted")]
+    assert len(placeholders) == 1
+    ph = placeholders[0]["content"]
+    assert "bash ×2" in ph
+    assert "read ×1" in ph
+    assert "tool_call_ids: call_0_0, call_0_1, call_0_2" in ph
+    assert "full results: /h.jsonl" in ph
+    # The other group remains complete.
+    assert any(m.get("role") == "tool" for m in out)
+
+
+def test_tool_window_preserves_non_foldable_and_validates():
+    """system/user/text messages untouched; folded output stays API-valid."""
+    server = _make_server()
+    messages: list[dict] = [
+        {"role": "system", "content": "sys"},
+        {"role": "user", "content": "u1"},
+    ]
+    for n in range(10):
+        messages.extend(_make_tool_group(n, ["bash"]))
+    messages.append({"role": "assistant", "content": "final text reply"})
+    out = server._apply_tool_window(
+        messages, keep_rounds=7, history_path="/tmp/x/history.jsonl"
+    )
+    assert out[0] == {"role": "system", "content": "sys"}
+    assert out[1] == {"role": "user", "content": "u1"}
+    assert out[-1] == {"role": "assistant", "content": "final text reply"}
+    # No orphaned tool messages after folding (OpenAI pairing holds).
+    valid = _validate_tool_messages([dict(m) for m in out])
+    i = 0
+    while i < len(valid):
+        if valid[i].get("tool_calls"):
+            j = i + 1
+            want = {tc["id"] for tc in valid[i]["tool_calls"]}
+            got: set[str] = set()
+            while j < len(valid) and valid[j].get("role") == "tool":
+                got.add(valid[j]["tool_call_id"])
+                j += 1
+            assert got == want
+            i = j
+        else:
+            i += 1
+    # Every tool message must be preceded by an assistant with tool_calls.
+    for k, m in enumerate(out):
+        if m.get("role") == "tool":
+            prev = out[k - 1] if k > 0 else {}
+            assert prev.get("tool_calls"), "tool message must follow assistant with tool_calls"