feat(v2): T-M2 采纳 pi branch-summarization——折叠摘要携带文件清单 + rollup 行带产物锚点

- AgentService._history_from_session：从全部轮次的 write_file/edit_file
  工具调用提取已读写文件清单（去重/单路径裁 120 字符/封顶 20 个），以摘要轮
  注入对话上下文——轮次滑出窗口后事实清单仍在（丢上下文不丢事实）
- Workspace.rollup：done 折叠行追加（产出: <artifact 锚点>），上下文折叠后
  产物事实在 a:// 锚点体系内仍可寻址；既有子串断言全部兼容
- 新增 tests/test_branch_manifest.py 4 项

gateway/agent.py

@@ -71,6 +71,9 @@ _REVIEW_SCHEMA_HINT = {
 
 DEFAULT_MAX_HANDOFFS = 2   # 规划者<->执行者交接轮数上限
 
+# 会写文件的工具名（T-M2 折叠摘要的文件清单来源）
+_FILE_TOOLS = {"write_file", "edit_file"}
+
 
 def _parse_json_loose(content: str) -> Dict[str, Any]:
     """宽松解析规划者的 JSON 输出（剥围栏/取首个对象）；失败返回 {}。"""
@@ -445,12 +448,31 @@ class AgentService:
     def _history_from_session(session: Optional["AgentSession"],
                               max_turns: int = 6,
                               max_chars: int = 1500) -> List[Dict[str, Any]]:
-        """把会话既往轮次折叠成对话上下文（不含工具细节）。"""
+        """把会话既往轮次折叠成对话上下文（不含工具细节）。
+
+        T-M2（采纳 pi branch-summarization 思想）：折叠摘要携带本会话
+        已写入/编辑的文件清单——轮次滑出窗口后，事实清单仍随首条摘要
+        进入对话，丢上下文不丢事实。
+        """
         if session is None:
             return []
         turns = [t for t in session.data.get("turns", [])
                  if t.get("state") == STATE_DONE and t.get("response")]
+        files = sorted({
+            str((a.get("arguments") or {}).get("path") or "").strip()
+            for t in session.data.get("turns", [])
+            for a in (t.get("tool_calls") or [])
+            if isinstance(a, dict)
+            and str(a.get("name") or "") in _FILE_TOOLS
+            and (a.get("arguments") or {}).get("path")
+        })
         out: List[Dict[str, Any]] = []
+        if files:
+            shown = "、".join(f[:120] for f in files[:20])
+            more = f"（等共 {len(files)} 个）" if len(files) > 20 else ""
+            out.append({"role": "user",
+                        "content": f"（上下文摘要）本会话此前已写入/编辑的文件：{shown}{more}"})
+            out.append({"role": "assistant", "content": "已了解上述文件背景，将继续任务。"})
         for t in turns[-max_turns:]:
             out.append({"role": "user", "content": str(t["task"])[:max_chars]})
             out.append({"role": "assistant", "content": str(t["response"])[:max_chars]})

router_system/workspace.py

@@ -397,7 +397,12 @@ class Workspace:
         new_progress: List[Dict[str, Any]] = []
         for entry in self._data.get("progress", []):
             if entry.get("status") == "done" and entry.get("step"):
-                line = _clip(f"{entry['step']}: {entry.get('summary', '')}", LIMITS["archive"])
+                # T-M2（采纳 pi branch-summarization 思想）：折叠行携带产物锚点，
+                # 上下文折叠后产物事实仍在 archive 中可寻址（a:// 锚点体系）
+                base = f"{entry['step']}: {entry.get('summary', '')}"
+                if entry.get("artifact"):
+                    base += f"（产出: {entry['artifact']}）"
+                line = _clip(base, LIMITS["archive"])
                 if line not in self._data["archive"]:
                     self._data["archive"].append(line)
                 folded += 1

tests/test_branch_manifest.py

@@ -0,0 +1,73 @@
+"""折叠摘要携带文件清单（T-M2，采纳 pi branch-summarization 设计）。"""
+import gateway.agent as ag
+from gateway.agent import AgentSession, AgentService
+from router_system.workspace import Workspace
+
+
+def _mk_turn(rid, task, tool_calls=None, state="done", response="好"):
+    return {"request_id": rid, "task": task, "response": response,
+            "state": state, "tool_calls": tool_calls or [], "tokens": 10,
+            "error": None, "ts": 0.0}
+
+
+def _sess(turns):
+    return AgentSession({"id": "as-x", "title": "s", "turns": turns})
+
+
+def test_history_carries_file_manifest_from_out_of_window_turns():
+    """文件清单来自全部轮次：写文件轮滑出窗口后清单仍在摘要里。"""
+    turns = [
+        _mk_turn("t1", "建笔记", tool_calls=[
+            {"id": "c1", "name": "write_file", "arguments": {"path": "notes.md"}},
+        ]),
+        _mk_turn("t2", "改笔记", tool_calls=[
+            {"id": "c2", "name": "edit_file", "arguments": {"path": "notes.md"}},
+        ]),
+        _mk_turn("t3", "闲聊"),
+    ]
+    hist = AgentService._history_from_session(_sess(turns), max_turns=1)
+    flat = "".join(str(m.get("content")) for m in hist)
+    # 窗口只剩 t3，但文件清单（t1/t2 产生）随摘要进入上下文
+    assert "notes.md" in flat
+    assert "上下文摘要" in flat
+    # 重复路径去重
+    assert flat.count("notes.md") >= 1
+
+
+def test_history_without_file_tools_has_no_manifest():
+    """无写文件工具调用 -> 不注入摘要轮（零噪音）。"""
+    turns = [_mk_turn("t1", "问个问题", tool_calls=[
+        {"id": "c1", "name": "web_fetch", "arguments": {"url": "https://x"}},
+    ])]
+    hist = AgentService._history_from_session(_sess(turns), max_turns=6)
+    assert all("上下文摘要" not in str(m.get("content")) for m in hist)
+
+
+def test_manifest_caps_path_length_and_count():
+    """路径裁剪 120 字符、清单封顶 20 个，防摘要膨胀。"""
+    turns = [_mk_turn(f"t{i}", "写", tool_calls=[
+        {"id": f"c{i}", "name": "write_file", "arguments": {"path": f"很长的路径{i}" * 30}},
+    ]) for i in range(25)]
+    hist = AgentService._history_from_session(_sess(turns), max_turns=0)
+    manifest = next(m for m in hist if "上下文摘要" in str(m.get("content")))
+    content = str(manifest["content"])
+    assert "等共 25 个" in content
+    body = content.split("：", 1)[1]  # 剥掉固定前缀后逐段校验单路径裁剪
+    for seg in body.split("（等共")[0].split("、"):
+        assert len(seg) <= 120
+
+
+def test_rollup_line_carries_artifact_anchor():
+    """workspace rollup 折叠行携带产物锚点（a:// 体系内可寻址）。"""
+    ws = Workspace.new(request_id="abc123def456", query="写个快排",
+                       api_token_cap=8000, rounds_cap=6)
+    ws.apply_brief({
+        "goal": "快排", "constraints": [], "tags": ["code"],
+        "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
+        "plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "过"}],
+    })
+    ws.add_progress("s1", "done", "完成", "a://s1_main.py")
+    ws.rollup()
+    archive = "".join(ws["archive"])
+    assert "s1" in archive
+    assert "（产出: a://s1_main.py）" in archive