chore(v2): .gitignore 登记 extra/ 参考项目目录（不入库）

- Workspace.render_for_architect：超 1200 token 时对 archive 行按与 query+goal
  的字符 2-gram Dice 相似度分档——sim<0.25 先丢、sim<0.55 截断为 40% 摘要、
  仍超限从低到高继续丢；保留行维持原插入序（前缀稳定），最近 4 行下限不变
  （相比旧'从最旧整段丢'，高相关事实在预算内留存更久）
- 顺带消除两项旧债：压缩态与首次渲染共用同一 parts 结构（格式漂移）、
  裁剪循环不再每轮 deepcopy 全文档
- 修复本轮引入的缺陷：会话轮次 tool_calls 为 int 计数时清单提取迭代崩溃
  （test_session_multi_turn 抓出，已加类型防御）
- 新增 tests/test_pinch_trim.py 4 项；全量 244 passed ×2（基线 230）

.gitignore

@@ -38,3 +38,6 @@ agent_workspace/
 
 # 安全扫描器工作目录（不入库）
 .mimosa/
+
+# 参考项目目录（不入库，仅作设计吸收来源）
+extra/

gateway/agent.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import json
 import time
 import uuid
@@ -70,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 输出（剥围栏/取首个对象）；失败返回 {}。"""
@@ -312,6 +316,13 @@ class AgentService:
         self.run_dir = Path(run_dir)
         self._runs: Dict[str, AgentRunInfo] = {}
         self.max_running = 5
+        self._running = 0  # O(1) 运行计数（原 register 每次全量扫描 _runs）
+
+    def _transition_state(self, info: AgentRunInfo, new_state: str) -> None:
+        """状态迁移的单一入口：离开 running 时同步递减计数。"""
+        if info.state == STATE_RUNNING and new_state != STATE_RUNNING:
+            self._running -= 1
+        info.state = new_state
 
     # ---------- 路径 ----------
     def _dir(self, request_id: str) -> Path:
@@ -327,14 +338,14 @@ class AgentService:
     def register(self, request_id: str, task: str, model: str, pool_id: str,
                  workspace: str = "", executor_model: str = "",
                  mode: str = "single") -> Optional[AgentRunInfo]:
-        running = [r for r in self._runs.values() if r.state == STATE_RUNNING]
+        if self._running >= self.max_running:
-        if len(running) >= self.max_running:
             return None
         info = AgentRunInfo(request_id=request_id, task=task, model=model,
                             pool_id=pool_id, workspace=workspace,
                             executor_model=executor_model, mode=mode,
                             started_at=time.time())
         self._runs[request_id] = info
+        self._running += 1
         self._dir(request_id).mkdir(parents=True, exist_ok=True)
         self._write_status(info)
         return info
@@ -413,7 +424,7 @@ class AgentService:
                 throttle.flush("executor")
             self._apply_result(info, result)
         except Exception as exc:  # pragma: no cover
-            info.state = STATE_FAILED
+            self._transition_state(info, STATE_FAILED)
             info.error = f"{type(exc).__name__}: {exc}"
             self._append_event(info, {"type": "final", "round": info.rounds,
                                       "reason": "error", "error": info.error})
@@ -437,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") if isinstance(t.get("tool_calls"), list) else [])
+            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]})
@@ -455,14 +485,14 @@ class AgentService:
         info.prompt_tokens = int(result.get("prompt_tokens", 0))
         info.completion_tokens = int(result.get("completion_tokens", 0))
         if result.get("reason") == "error":
-            info.state = STATE_FAILED
+            self._transition_state(info, STATE_FAILED)
             info.error = result.get("error")
         elif result.get("reason") in ("token_cap", "max_rounds", "max_handoffs"):
             # 触顶属于护栏行为：结果仍交付，但标记部分完成信息
-            info.state = STATE_DONE
+            self._transition_state(info, STATE_DONE)
             info.error = result.get("error")
         else:
-            info.state = STATE_DONE
+            self._transition_state(info, STATE_DONE)
 
     # ---------- 两级模式（D7）：规划者 + 执行者 ----------
     async def run_dual(self, info: AgentRunInfo, planner_chat: Any, executor_chat: Any,
@@ -882,6 +912,36 @@ class SessionStore:
         self.save(sess)
         return sess
 
+    def fork(self, sid: str, turn_index: Optional[int] = None,
+             title: str = "") -> Optional[AgentSession]:
+        """从既有会话分叉新会话（T-M1，采纳 pi 会话树 id/parentId 设计）。
+
+        值拷贝 turns[:turn_index+1]（缺省 = 全部已完结轮次；-1 = 空会话壳），
+        原会话不变；新会话记录 parent_id / fork_point 保留血统。采用拷贝式
+        而非 pi 的共享存储引用——规避并发写冲突，代价是 fork 后磁盘冗余
+        （会话体量小，可接受）。turn_index 越界返回 None。
+        """
+        src = self.get(sid)
+        if src is None:
+            return None
+        turns = src.data.get("turns", []) or []
+        n = len(turns) - 1 if turn_index is None else int(turn_index)
+        if n < -1 or n >= len(turns):
+            return None
+        data = copy.deepcopy(src.data)
+        data["id"] = "as" + uuid.uuid4().hex[:10]
+        base_title = (title or "").strip() or (src.data.get("title") or "会话") + "·分叉"
+        data["title"] = base_title[:24]
+        data["parent_id"] = sid
+        data["fork_point"] = n
+        data["created_at"] = time.time()
+        data["busy"] = False
+        data["turns"] = copy.deepcopy(turns[:n + 1]) if n >= 0 else []
+        sess = AgentSession(data)
+        self._cache[data["id"]] = sess
+        self._save(sess)
+        return sess
+
     def save(self, sess: AgentSession) -> None:
         self._cache[sess.data["id"]] = sess
         self._save(sess)

gateway/api.py

@@ -647,7 +647,7 @@ try:
             except Exception as exc:
                 import traceback
                 traceback.print_exc()
-                info.state = "failed"
+                service._transition_state(info, "failed")
                 info.error = str(exc)
                 info.finished_at = __import__("time").time()
                 service._write_status(info)
@@ -674,7 +674,7 @@ try:
             return {"ok": False, "detail": f"任务已结束（{info.state}）"}
         if info.asyncio_task is not None:
             info.asyncio_task.cancel()
-        info.state = "failed"
+        service._transition_state(info, "failed")
         info.error = "cancelled_by_user"
         info.finished_at = __import__("time").time()
         service._write_status(info)
@@ -740,6 +740,24 @@ try:
             raise HTTPException(status_code=404, detail=f"会话不存在: {sid}")
         return sess.view()
 
+    @app.post("/agent/sessions/{sid}/fork", tags=["agent"])
+    async def agent_session_fork(sid: str, req: dict = None):
+        """从会话分叉新会话：{"turn_index": 截止轮下标(缺省=全部轮), "title"}。
+
+        T-M1（采纳 pi 会话树设计）：值拷贝历史轮次，原会话不变；
+        新会话带 parent_id/fork_point 血统字段。
+        """
+        _check_id(sid, "会话 ID")
+        from gateway.agent import get_session_store
+        body = req or {}
+        turn_index = body.get("turn_index")
+        sess = get_session_store().fork(sid, turn_index,
+                                        str(body.get("title") or ""))
+        if sess is None:
+            raise HTTPException(status_code=404,
+                                detail=f"会话不存在或分叉点越界: {sid}")
+        return sess.view()
+
     @app.delete("/agent/sessions/{sid}", tags=["agent"])
     async def agent_session_delete(sid: str):
         _check_id(sid, "会话 ID")

router_system/inference.py

@@ -53,20 +53,26 @@ class InferenceEngine:
     # ---------------------------------------------------------------
     def run(self, query: str, domain: str, memory: WorkingMemory,
             max_steps: Optional[int] = None) -> List[str]:
-        """前向链主循环。返回触发规则 id 列表（按触发顺序）。"""
+        """前向链主循环。返回触发规则 id 列表（按触发顺序）。
+
+        循环不变量外提：query/domain 在循环内不变，kb.match 结果只算一次
+        （原实现每步全量重扫+重排序，最坏 O(steps × rules × patterns)）。
+        """
         steps = max_steps or self.max_steps
+        rules = self.kb.match(query, domain=domain)
         fired: List[str] = []
+        fired_set: set = set()  # O(1) 查重（fired 保持列表维护触发顺序）
         for _ in range(steps):
-            rules = self.kb.match(query, domain=domain)
             # 选第一个"未触发过"的规则
             target: Optional[Rule] = None
             for r in rules:
-                if r.id not in fired:
+                if r.id not in fired_set:
                     target = r
                     break
             if target is None:
                 break  # 无新规则可触发 → 终止
             fired.append(target.id)
+            fired_set.add(target.id)
             self._fire(target, query, memory)
         return fired
 

router_system/knowledge.py

@@ -51,13 +51,25 @@ class Rule:
     actions: List[str] = field(default_factory=list)  # 保留字段：动作扩展
     subdomain: Optional[str] = None     # 二级子领域（如 investing/labor/calculus）
     subdomain2: Optional[str] = None    # 三级子领域（如 fund/overtime/sorting）
+    _patterns_lower: Optional[tuple] = field(default=None, repr=False, compare=False)
+
+    def _lowered(self) -> tuple:
+        """patterns 的小写缓存（注册后规则视为不可变；懒计算一次）。"""
+        if self._patterns_lower is None:
+            self._patterns_lower = tuple(p.lower() for p in self.patterns)
+        return self._patterns_lower
 
     def matches(self, text: str) -> bool:
         """任一 pattern 是 text 的子串即命中（大小写不敏感）。"""
         if not self.patterns:
             return False
-        q = text.lower()
+        return self._match_lower(text.lower())
-        return any(p.lower() in q for p in self.patterns)
+
+    def _match_lower(self, q: str) -> bool:
+        """已 lowercase 文本的快速匹配（避免每条规则重复 lower 同一文本）。"""
+        if not self.patterns:
+            return False
+        return any(p in q for p in self._lowered())
 
 
 # ---------------------------------------------------------------
@@ -420,12 +432,13 @@ class KnowledgeBase:
         self._facts = {d: [dict(f) for f in facts] for d, facts in BUILTIN_FACTS.items()}
 
         if self.rules_dir.is_dir():
+            yaml_names = {p.name for p in self.rules_dir.glob("*.yaml")}  # 一次遍历，避免逐文件重扫
             for f in sorted(self.rules_dir.glob("*.yaml")):
                 data = _try_load_yaml(f)
                 if data is not None:
                     self._load_file_data(f, data)
             for f in sorted(self.rules_dir.glob("*.json")):
-                if f.name not in {p.name for p in self.rules_dir.glob("*.yaml")}:
+                if f.name not in yaml_names:
                     data = _try_load_json(f)
                     if data is not None:
                         self._load_file_data(f, data)
@@ -461,12 +474,16 @@ class KnowledgeBase:
 
     # ---- 查询 ----
     def match(self, text: str, domain: Optional[str] = None) -> List[Rule]:
-        """返回命中的规则，按优先级降序。domain 为空则全领域匹配。"""
+        """返回命中的规则，按优先级降序。domain 为空则全领域匹配。
+
+        文本只 lowercase 一次（原实现每条规则各 lower 一遍）。
+        """
+        q = text.lower()
         hits = []
         for rule in self._rules.values():
             if domain is not None and rule.domain != domain:
                 continue
-            if rule.matches(text):
+            if rule._match_lower(q):
                 hits.append(rule)
         hits.sort(key=lambda r: r.priority, reverse=True)
         return hits

router_system/pipeline.py

@@ -110,20 +110,24 @@ class CollaborativePipeline:
         plan = brief.get("plan") or []
         pending = [p.get("id") for p in plan]
         plan_ids = {p.get("id") for p in plan}          # 用于 _deps_done 过滤
+        plan_by_id = {p.get("id"): p for p in plan}     # O(1) 步定义查找（原每轮线性扫描）
+        done_ids = self._done_ids(ws)                   # 增量维护的完成集（原每轮重建）
+        base_domain = self._guess_domain(query)         # ws["query"] 恒定，领域只解析一次
         while pending and not ws.exhausted():
             progressed = False
             for sid in list(pending):
-                step = next((p for p in plan if p.get("id") == sid), {})
+                step = plan_by_id.get(sid, {})
                 deps = step.get("deps") or []
                 # 只检查在 plan 中的依赖；不在 plan 的 ID 视为"不存在"→自动满足
                 relevant = [d for d in deps if d in plan_ids]
-                if not self._deps_done(ws, relevant):
+                if not all(d in done_ids for d in relevant):
                     continue
-                existing = self._read_artifact(request_id, self._artifact_name(sid, ws))
+                existing = self._read_artifact(request_id, self._artifact_name(sid, ws, domain=base_domain))
                 outcome = await self.worker.run_step(ws, sid, existing_artifact=existing,
                                                      hint=self._last_decision_for(ws, sid))
                 if outcome.status == "done":
                     pending.remove(sid)
+                    done_ids.add(sid)
                     self._save_artifact(request_id, outcome.artifact_name, outcome.artifact_text)
                     ws.rollup()
                     route.append(f"step:{sid}:done")
@@ -230,9 +234,10 @@ class CollaborativePipeline:
                 return hits[0].domain
         return "general"
 
-    def _artifact_name(self, sid: str, ws: Workspace) -> str:
+    def _artifact_name(self, sid: str, ws: Workspace, domain: Optional[str] = None) -> str:
         from .worker import artifact_name_for
-        domain = self._guess_domain(ws["query"])
+        if domain is None:
+            domain = self._guess_domain(ws["query"])
         # 用 brief.tags 优先
         tags = (ws.get("brief") or {}).get("tags") or []
         for t in tags:
@@ -241,8 +246,10 @@ class CollaborativePipeline:
                 break
         return artifact_name_for(sid, domain)
 
-    def _deps_done(self, ws: Workspace, deps: List[str]) -> bool:
+    @staticmethod
-        # progress 中 done 的条目；done 条目 rollup 后移到 archive（字符串格式如 "s1: ..."）
+    def _done_ids(ws: Workspace) -> set:
+        """收集已完成 step id 集合：progress 中 done 的条目 + archive 折叠行
+        （done 条目 rollup 后移到 archive，字符串格式如 "s1: ..."）。"""
         done = {p["step"] for p in ws.get("progress", []) if p.get("status") == "done"}
         for entry in ws.get("archive", []):
             if isinstance(entry, dict):
@@ -251,9 +258,13 @@ class CollaborativePipeline:
                 sid = entry.split(":")[0].strip() if ":" in entry else ""
             else:
                 sid = ""
-            if sid in deps:
+            done.add(sid)
-                done.add(sid)
+        return done
-        return all(d in done for d in deps)
+
+    def _deps_done(self, ws: Workspace, deps: List[str], done_ids: Optional[set] = None) -> bool:
+        if done_ids is None:
+            done_ids = self._done_ids(ws)
+        return all(d in done_ids for d in deps)
 
     def _last_decision_for(self, ws: Workspace, sid: str) -> str:
         """取最近一条针对该 step 的决策 reply，作为 worker hint。"""

router_system/v2stats.py

@@ -2,6 +2,9 @@
 
 T9：每请求 API token 记账；聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。
 配合 /metrics 对外透出（论文 E1 token 经济学数据来源之一）。
+
+性能设计（2026-09 优化）：回合数分布以"和/最大值/分桶计数"增量维护，
+summary() 从每次全量重算 O(n) 降为 O(桶数)，且不再持有无界 list。
 """
 from __future__ import annotations
 
@@ -18,7 +21,11 @@ class V2Stats:
         self._fast_path = 0
         self._breach = 0
         self._by_status: Dict[str, int] = {}
-        self._rounds: List[int] = []
+        # 回合数增量聚合（等价于全量 list 的 sum/max/histogram，内存 O(桶数)）
+        self._rounds_count = 0
+        self._rounds_sum = 0
+        self._rounds_max = 0
+        self._rounds_hist: Dict[str, int] = {}
         self._api_input_tokens = 0
         self._api_output_tokens = 0
         self._api_cost_usd = 0.0
@@ -33,7 +40,13 @@ class V2Stats:
                 self._fast_path += 1
             status = getattr(result, "status", "?")
             self._by_status[status] = self._by_status.get(status, 0) + 1
-            self._rounds.append(getattr(result, "rounds_used", 0))
+            rounds = getattr(result, "rounds_used", 0)
+            self._rounds_count += 1
+            self._rounds_sum += rounds
+            if rounds > self._rounds_max:
+                self._rounds_max = rounds
+            bucket = str(rounds) if rounds <= 10 else ">10"
+            self._rounds_hist[bucket] = self._rounds_hist.get(bucket, 0) + 1
             if "breach" in " ".join(getattr(result, "route", [])):
                 self._breach += 1
             self._api_input_tokens += getattr(result, "api_input_tokens", 0)
@@ -64,16 +77,16 @@ class V2Stats:
     def summary(self) -> Dict[str, Any]:
         with self._lock:
             total = self._total
-            rounds = self._rounds
+            rounds_n = self._rounds_count
             return {
                 "total_requests": total,
                 "fast_path_rate": round(self._fast_path / total, 4) if total else 0.0,
                 "status_distribution": dict(self._by_status),
                 "breach_count": self._breach,
                 "rounds_used": {
-                    "avg": round(sum(rounds) / len(rounds), 2) if rounds else 0.0,
+                    "avg": round(self._rounds_sum / rounds_n, 2) if rounds_n else 0.0,
-                    "max": max(rounds) if rounds else 0,
+                    "max": self._rounds_max if rounds_n else 0,
-                    "distribution": _histogram(rounds),
+                    "distribution": dict(self._rounds_hist),
                 },
                 "api_tokens": {
                     "input": self._api_input_tokens,

router_system/worker.py

@@ -188,25 +188,33 @@ def _mock_generate() -> Callable[[str], Awaitable[str]]:
 
 def _make_llama_generate(cfg: Dict[str, Any],
                            default_base_url: Optional[str] = None) -> Callable[[str], Awaitable[str]]:
-    """返回调用本地 OpenAI 兼容端点（llama-server / Ollama / vLLM）的生成器。"""
+    """返回调用本地 OpenAI 兼容端点（llama-server / Ollama / vLLM）的生成器。
+
+    连接复用：httpx.AsyncClient 懒建一次、跨步骤复用（与 ArchitectClient 一致），
+    避免协作循环每步重新 TCP 建连。
+    """
     if default_base_url is None:
         default_base_url = f"http://127.0.0.1:{cfg.get('port', 8901)}/v1"
     base_url = cfg.get("base_url") or default_base_url
     model = cfg.get("model") or "local"
     temperature = float(cfg.get("temperature", 0.3))
     timeout_s = float(cfg.get("per_step_timeout_s", 300))
+    client_holder: Dict[str, Any] = {"client": None}
 
     async def _gen(prompt: str) -> str:
         try:
             import httpx
-            async with httpx.AsyncClient(timeout=timeout_s) as client:
+            client = client_holder["client"]
-                resp = await client.post(
+            if client is None or client.is_closed:
-                    f"{base_url}/chat/completions",
+                client = httpx.AsyncClient(timeout=timeout_s)
-                    json={"model": model, "messages": [{"role": "user", "content": prompt}],
+                client_holder["client"] = client
-                          "temperature": temperature, "max_tokens": 4096},
+            resp = await client.post(
-                )
+                f"{base_url}/chat/completions",
-                resp.raise_for_status()
+                json={"model": model, "messages": [{"role": "user", "content": prompt}],
-                return resp.json()["choices"][0]["message"]["content"]
+                      "temperature": temperature, "max_tokens": 4096},
+            )
+            resp.raise_for_status()
+            return resp.json()["choices"][0]["message"]["content"]
         except Exception as e:  # noqa: BLE001
             # 连不上本地模型 -> 优雅降级（不抛 500），提示用户检查模型端点
             return ("（本地降级）无法连接本地模型端点，未能生成该步骤内容。"

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
@@ -445,15 +450,17 @@ class Workspace:
         for p in d.get("progress", [])[-5:]:
             parts.append(f"{p.get('step')} [{p.get('status')}] {_clip(p.get('summary', ''), 40)}")
         parts.append("== archive ==")
-        for line in d.get("archive", [])[-8:]:
+        archive = d.get("archive", []) or []
-            parts.append(line)
+        shown = list(archive[-8:])
-        # token 预算：超限先截断最旧 archive（已只保留 3 条 decisions）
+        parts.extend(shown)
+        # token 预算：超限按 pinch 三档裁剪 archive（T-M3，采纳 llmrouter pinch
+        # keep/summarize/drop 分档思想），决策见 _trim_archive_lines
         out = "\n".join(parts)
-        while estimate_tokens(out) > 1200 and len(d.get("archive", [])) > 4:
+        if estimate_tokens(out) <= 1200 or len(shown) <= 4:
-            d = copy.deepcopy(d)
+            return out  # 预算内 / 已在下限（与旧实现的 4 行下限一致）
-            d["archive"] = d["archive"][4:]
+        head = parts[:-len(shown)]
-            out = "\n".join(_rerender(self, d))
+        ref_text = d.get("query", "") + " " + (d.get("brief") or {}).get("goal", "")
-        return out
+        return _trim_archive_lines(head, archive, shown, ref_text)
 
     def render_for_worker(self, step_id: str,
                           artifact_text: Optional[str] = None) -> str:
@@ -520,24 +527,69 @@ class Workspace:
         return self.data
 
 
-def _rerender(ws: "Workspace", d: Dict[str, Any]) -> List[str]:
+def _bigrams(text: str) -> Dict[str, int]:
-    """用裁剪后的数据重建 Architect 渲染（供超限压缩内部用）。"""
+    """字符 2-gram 计数（去空白、小写；零依赖轻量相关性度量）。"""
-    parts: List[str] = []
+    t = "".join(text.lower().split())
-    m = d["meta"]
+    out: Dict[str, int] = {}
-    parts.append("== meta ==")
+    for i in range(len(t) - 1):
-    parts.append(f"status={m['status']} round={m['round']}")
+        g = t[i:i + 2]
-    parts.append("== query ==")
+        out[g] = out.get(g, 0) + 1
-    parts.append(_clip(d["query"], LIMITS["query_truncate"]))
+    return out
-    parts.append("== issues ==")
+
-    for iss in d.get("issues", []):
+
-        parts.append(f"{iss['id']} step={iss.get('step')} ask={_clip(iss.get('ask', ''), 80)}")
+def _dice(a: Dict[str, int], b: Dict[str, int]) -> float:
-    parts.append("== decisions(最近3) ==")
+    """Dice 系数：2 * 交集 / (|a| + |b|)，空集返回 0。"""
-    for dec in d.get("decisions", [])[-3:]:
+    if not a or not b:
-        parts.append(f"ref={dec.get('ref')} reply={_clip(dec.get('reply', ''), 80)}")
+        return 0.0
-    parts.append("== progress ==")
+    inter = sum(min(v, b.get(k, 0)) for k, v in a.items())
-    for p in d.get("progress", [])[-5:]:
+    return 2.0 * inter / (sum(a.values()) + sum(b.values()))
-        parts.append(f"{p.get('step')} [{p.get('status')}] {_clip(p.get('summary', ''), 40)}")
+
-    parts.append("== archive ==")
+
-    for line in d.get("archive", [])[-6:]:
+def _trim_archive_lines(head: List[str], archive: List[str],
-        parts.append(line)
+                        shown: List[str], ref_text: str) -> str:
-    return parts
+    """Architect 渲染超预算时的 archive 三档裁剪（T-M3，采纳 llmrouter pinch
+    keep/summarize/drop 分档思想）。
+
+    相关度 = 行与 query+goal 的字符 2-gram Dice 系数：
+      第一档：丢 sim<0.25 的行（最低相关先丢）；
+      第二档：把 sim<0.55 的行截断为 40% 摘要；
+      第三档：仍超限则从低相关到高相关继续丢。
+    保留行维持原插入序（前缀稳定）；任何情况下至少保留 4 行，
+    达到下限仍超限则接受溢出（与旧实现一致）。
+    相比旧"从最旧起整段丢弃"，高相关事实在预算内留存得更久；
+    且压缩态与首次渲染共用同一 parts 结构（消除格式漂移与每轮 deepcopy）。
+    """
+    ref = _bigrams(ref_text)
+    start = len(archive) - len(shown)
+    idx = list(range(start, len(archive)))
+    sims = {i: _dice(_bigrams(archive[i]), ref) for i in idx}
+    order = sorted(idx, key=lambda i: (sims[i], i))
+
+    kept = set(idx)
+    clips: Dict[int, str] = {}
+
+    def _over_budget() -> bool:
+        lines = [clips.get(i, archive[i]) for i in sorted(kept)]
+        return estimate_tokens("\n".join(head + lines)) > 1200
+
+    # 第一档：丢低相关
+    for i in order:
+        if not _over_budget() or len(kept) <= 4:
+            break
+        if sims[i] < 0.25:
+            kept.discard(i)
+    # 第二档：中相关截断为摘要（40% 长度）
+    for i in order:
+        if not _over_budget():
+            break
+        if i in kept and sims[i] < 0.55:
+            clips[i] = _clip(archive[i], max(20, int(len(archive[i]) * 0.4)))
+    # 第三档：仍超限从低到高继续丢
+    for i in order:
+        if not _over_budget() or len(kept) <= 4:
+            break
+        kept.discard(i)
+        clips.pop(i, None)
+
+    lines = [clips.get(i, archive[i]) for i in sorted(kept)]
+    return "\n".join(head + lines)

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

tests/test_pinch_trim.py

@@ -0,0 +1,60 @@
+"""Architect 渲染三档裁剪（T-M3，采纳 llmrouter pinch keep/summarize/drop 设计）。"""
+from router_system.workspace import Workspace
+
+_GOAL = "用 Python 实现快速排序并分析时间复杂度"
+
+_BRIEF = {
+    "goal": _GOAL, "constraints": ["标准库"], "tags": ["code"],
+    "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
+    "plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "过"}],
+}
+
+
+def _ws_with_archive():
+    ws = Workspace.new(request_id="abc123def456", query=_GOAL,
+                       api_token_cap=8000, rounds_cap=6)
+    ws.apply_brief(_BRIEF)
+    relevant = ("s1: 完成快速排序实现，基准选取三数取中，递归深度优化，"
+                "平均时间复杂度 O(n log n) 分析见附件。" + "快排边界处理细节" * 50)
+    lines = [f"sX{i}: 无关行{i}，今日食堂菜单有红烧肉、番茄炒蛋与清炒时蔬，"
+             f"周末计划去郊外徒步露营并整理旅行照片。" + "生活琐事记录" * 50
+             + f"独特结尾标记{i}号"  # 唯一尾串：判断"该行是否还被输出"
+             for i in range(7)]
+    # 相关行放在中间：不在最新 4 行下限保护区内，检验相关性而非新近度
+    ws._data["archive"] = lines[:3] + [relevant] + lines[3:]
+    return ws, relevant, lines
+
+
+def test_trim_keeps_relevant_line_full():
+    """高相关行全文保留，低相关行被丢弃/截断（尾段不再出现在输出）。"""
+    ws, relevant, lines = _ws_with_archive()
+    out = ws.render_for_architect()
+    # 相关行全文（含尾段独特内容）仍在
+    assert relevant[-40:] in out
+    # 至少一条低相关行的尾段消失（被丢或被截为摘要）
+    assert lines[0][-40:] not in out
+
+
+def test_trim_respects_floor_of_four_lines():
+    """任何情况下至少保留 4 行（与旧实现下限一致）。"""
+    ws, _, lines = _ws_with_archive()
+    out = ws.render_for_architect()
+    archive_shown = out.split("== archive ==")[1].strip().splitlines()
+    assert len(archive_shown) >= 4
+
+
+def test_render_deterministic():
+    """同输入两次渲染逐字节一致（裁剪决策确定性）。"""
+    ws1, _, _ = _ws_with_archive()
+    ws2, _, _ = _ws_with_archive()
+    assert ws1.render_for_architect() == ws2.render_for_architect()
+
+
+def test_under_budget_untouched():
+    """预算内不触发裁剪：archive 行原样出现、无截断省略号。"""
+    ws = Workspace.new(request_id="abc123def456", query="写个快排",
+                       api_token_cap=8000, rounds_cap=6)
+    ws.apply_brief(_BRIEF)
+    ws._data["archive"] = ["s1: 完成（产出: a://s1.py）"]
+    out = ws.render_for_architect()
+    assert "s1: 完成（产出: a://s1.py）" in out

tests/test_session_fork.py

@@ -0,0 +1,98 @@
+"""会话树 fork 单元与端点测试（T-M1，采纳 pi 会话树 id/parentId 设计）。"""
+import pytest
+
+pytest.importorskip("fastapi")
+pytest.importorskip("httpx")
+
+from fastapi.testclient import TestClient
+
+import gateway.agent as ag
+import gateway.api as ga
+from gateway.agent import AgentSession, SessionStore
+
+
+def _mk_turn(rid: str, task: str) -> dict:
+    return {"request_id": rid, "task": task, "response": f"ans-{rid}",
+            "state": "done", "tool_calls": [], "tokens": 10,
+            "error": None, "ts": 0.0}
+
+
+def test_fork_copies_turns_and_keeps_lineage(tmp_path):
+    """值拷贝历史轮次 + 记录 parent_id/fork_point 血统。"""
+    store = SessionStore(root=tmp_path / "sessions")
+    src = store.create("源会话", "ws")
+    src.data["turns"] = [_mk_turn("t1", "任务一"), _mk_turn("t2", "任务二")]
+    store.save(src)
+
+    forked = store.fork(src.data["id"], turn_index=0, title="分叉A")
+    assert forked is not None
+    assert forked.data["id"] != src.data["id"]
+    assert forked.data["parent_id"] == src.data["id"]
+    assert forked.data["fork_point"] == 0
+    assert [t["request_id"] for t in forked.data["turns"]] == ["t1"]
+    # 原会话不变
+    assert len(store.get(src.data["id"]).data["turns"]) == 2
+
+
+def test_fork_default_takes_all_turns(tmp_path):
+    """缺省 turn_index = 全部已完结轮次。"""
+    store = SessionStore(root=tmp_path / "sessions")
+    src = store.create("源", "ws")
+    src.data["turns"] = [_mk_turn("t1", "a"), _mk_turn("t2", "b")]
+    store.save(src)
+    forked = store.fork(src.data["id"])
+    assert len(forked.data["turns"]) == 2
+    assert forked.data["fork_point"] == 1
+
+
+def test_fork_out_of_range_returns_none(tmp_path):
+    """turn_index 越界 / 源会话不存在 -> None（端点映射 404）。"""
+    store = SessionStore(root=tmp_path / "sessions")
+    src = store.create("源", "ws")
+    src.data["turns"] = [_mk_turn("t1", "a")]
+    store.save(src)
+    assert store.fork(src.data["id"], turn_index=5) is None
+    assert store.fork(src.data["id"], turn_index=-2) is None
+    assert store.fork("as-not-exist") is None
+
+
+def test_fork_is_deep_copy(tmp_path):
+    """fork 后修改新会话不影响原会话（值拷贝语义）。"""
+    store = SessionStore(root=tmp_path / "sessions")
+    src = store.create("源", "ws")
+    src.data["turns"] = [_mk_turn("t1", "a")]
+    store.save(src)
+    forked = store.fork(src.data["id"])
+    forked.data["turns"][0]["task"] = "被篡改"
+    forked.data["turns"].append(_mk_turn("t2", "新轮"))
+    assert store.get(src.data["id"]).data["turns"][0]["task"] == "a"
+    assert len(store.get(src.data["id"]).data["turns"]) == 1
+
+
+def test_old_sessions_without_lineage_fields_compatible(tmp_path):
+    """旧数据无 parent_id/fork_point 字段自然兼容（dict.get 语义，零迁移）。"""
+    store = SessionStore(root=tmp_path / "sessions")
+    sess = AgentSession({"id": "as-legacy", "title": "旧", "turns": []})
+    store.save(sess)
+    assert store.get("as-legacy").data.get("parent_id", "") == ""
+
+
+def test_fork_endpoint(tmp_path, monkeypatch):
+    """POST /agent/sessions/{sid}/fork：端到端 200 + 404。"""
+    ag.reset_session_store()
+    monkeypatch.setattr(ag, "_session_store", SessionStore(root=tmp_path / "sessions"))
+    store = ag._session_store
+    src = store.create("源", "ws")
+    src.data["turns"] = [_mk_turn("t1", "a")]
+    store.save(src)
+
+    client = TestClient(ga.app)
+    r = client.post(f"/agent/sessions/{src.data['id']}/fork",
+                    json={"turn_index": 0, "title": "端点分叉"})
+    assert r.status_code == 200
+    data = r.json()
+    assert data["parent_id"] == src.data["id"]
+    assert data["title"] == "端点分叉"
+
+    r404 = client.post("/agent/sessions/as-not-exist/fork", json={})
+    assert r404.status_code == 404