diff --git a/config/config.yaml b/config/config.yaml
index 215957e..62a2ed1 100644
--- a/config/config.yaml
+++ b/config/config.yaml
@@ -1,49 +1,113 @@
-# ============================================================
-# 多专业小模型 + 路由模型系统 — 配置
-# 默认全 mock 模式（零依赖、离线可跑）。要接入真实模型，把对应
-# 后端 type 改为 hf / api 即可（见 README）。
-# ============================================================
-
-system:
-  name: multi-expert-router
-  version: 0.1.0
-
-router:
-  low_confidence_threshold: 0.60     # 分类置信度低于此值 -> 直接走大模型
-  judge_fallback_threshold: 0.70     # Judge 质量分低于此值 -> 升级大模型
-  default_temperature: 0.2
-
-classifier:
-  type: rule                          # rule（零依赖）| hf（transformers）
-  model: Qwen/Qwen3-0.6B
-  confidence_floor: 0.55
-
-domains:
-  - code
-  - math
-  - legal
-  - medical
-  - general
-
-experts:
-  code:     { type: mock, model: Qwen/Qwen2.5-Coder-7B-Instruct }
-  math:     { type: mock, model: Qwen/Qwen3-4B-Instruct }
-  legal:    { type: mock, model: Qwen/Qwen3-4B-Instruct }
-  medical:  { type: mock, model: Qwen/Qwen3-4B-Instruct }
-  general:  { type: mock, model: Qwen/Qwen3-1.7B-Instruct }
-
-fallback:
-  type: mock                          # mock | api（OpenAI 兼容，如 DeepSeek）
-  model: deepseek-chat
-  base_url: https://api.deepseek.com/v1
-  api_key_env: DEEPSEEK_API_KEY
-
-judge:
-  type: rule                          # rule（零依赖）| llm
-  model: Qwen/Qwen3-1.7B-Instruct
-
-cache:
-  enabled: true
-  semantic_enabled: true              # 语义缓存（字符 n-gram 相似度，零依赖）
-  similarity_threshold: 0.88
-  promote_frequency: 5                # 命中 N 次后提升为精确缓存

+# ============================================================
+# 多专业小模型 + 路由模型系统 — 配置
+# 默认全 mock 模式（零依赖、离线可跑）。要接入真实模型，把对应
+# 后端 type 改为 hf / api 即可（见 README）。
+# ============================================================
+
+system:
+  name: multi-expert-router
+  version: 0.1.0
+
+router:
+  low_confidence_threshold: 0.60     # 分类置信度低于此值 -> 直接走最后处理者
+  judge_fallback_threshold: 0.70     # Judge 质量分低于此值 -> 升级最后处理者
+  default_temperature: 0.2
+
+# 专家系统内核执行配置（L0 默认零参数）
+execution:
+  mode: rule                          # rule（默认，L0 零参数）| hybrid
+  planner: rule                       # rule（规则拆解）| hf（可选小模型拆解）
+  expert_backend: rule                # rule（规则执行器）| hf | api（本地小模型按需加载）
+  model_level: L0                     # L0 纯规则 | L1 分类/Planner增强 | L2 领域生成
+  max_plan_depth: 3                   # 任务拆解深度上限
+
+classifier:
+  type: rule                          # rule（零依赖）| hf（transformers）
+  model: Qwen/Qwen3-0.6B
+  confidence_floor: 0.55
+
+domains:
+  - code
+  - math
+  - legal
+  - medical
+  - finance
+  - life
+  - education
+  - general
+
+# 两级路由：大领域分组（用户接口指定 group → 组内路由模型 → 组内专业小模型）
+# 组内路由模型只识别本组领域，体积约为统一路由模型的 1/4
+domain_groups:
+  tech: [code, math]
+  professional: [legal, medical, finance]
+  lifestyle: [life, education]
+  general: [general]
+
+experts:
+  code:      { type: mock, model: Qwen/Qwen2.5-Coder-7B-Instruct }
+  math:      { type: mock, model: Qwen/Qwen3-4B-Instruct }
+  legal:     { type: mock, model: Qwen/Qwen3-4B-Instruct }
+  medical:   { type: mock, model: Qwen/Qwen3-4B-Instruct }
+  finance:   { type: mock, model: Qwen/Qwen3-4B-Instruct }
+  life:      { type: mock, model: Qwen/Qwen3-1.7B-Instruct }
+  education: { type: mock, model: Qwen/Qwen3-1.7B-Instruct }
+  general:   { type: mock, model: Qwen/Qwen3-1.7B-Instruct }
+
+fallback:
+  type: mock                          # none（降级模板）| mock | local（本地≤8B 按需加载）| api
+  model: deepseek-chat
+  base_url: https://api.deepseek.com/v1
+  api_key_env: DEEPSEEK_API_KEY
+
+judge:
+  type: rule                          # rule（零依赖）| llm
+  model: Qwen/Qwen3-1.7B-Instruct
+
+cache:
+  enabled: true
+  semantic_enabled: true              # 语义缓存（字符 n-gram 相似度，零依赖）
+  similarity_threshold: 0.88
+  promote_frequency: 5                # 命中 N 次后提升为精确缓存
+
+# ============================================================
+# v2：端云协同 LLM 协作系统（《实现方案_v2》）配置段
+# v1 段（上方）保留，供 legacy 路由（POST /chat/legacy）使用。
+# ============================================================
+
+runtime:
+  llama_server:
+    binary: bin/llama-server.exe        # 捆绑上游 release，不改源码（D1）
+    model: models/qwen3.5-4b-q4_k_m.gguf
+    port: 8901
+    hw_profile: auto                    # auto | gpu12 | gpu8 | cpu
+    extra_args: ["-fa", "-ctk", "q8_0", "-ctv", "q8_0", "--cache-reuse", "256"]
+  tiers:                                # 三档硬件模板（保守默认，可手动覆盖）
+    gpu12: {ngl: 99, ctx: 32768}
+    gpu8:  {ngl: 14, ctx: 16384}
+    cpu:   {ngl: 0,  ctx: 8192}
+
+architect:                              # 大模型（API）
+  model: deepseek-chat
+  base_url: https://api.deepseek.com/v1
+  api_key_env: DEEPSEEK_API_KEY
+  temperature: 0.2
+  timeout_s: 60
+
+worker:                                 # 小模型（本地）
+  backend: llama_server
+  temperature: 0.3
+  max_fix_attempts: 2
+  per_step_timeout_s: 300
+
+pipeline:
+  fast_path: true
+  rounds_cap: 6
+  api_token_cap: 8000
+  breach_policy: architect_do           # architect_do | local_only
+
+review:
+  queue_db: data/review.sqlite3
+  sample_rate: 0.10                     # 随机抽样送审
+  force_tags: [safety]                  # brief.tags 命中即强制送审
+
diff --git a/router_system/workspace.py b/router_system/workspace.py
new file mode 100644
index 0000000..fd08d39
--- /dev/null
+++ b/router_system/workspace.py
@@ -0,0 +1,516 @@
+"""交流文本（Workspace）—— 端云协同 LLM 协作系统的核心协议（零依赖）。
+
+大模型（Architect）与小模型（Worker）互不共享内部状态，只通过这份
+schema 约束的结构化 JSON 共享工作区交接（类比前后端通过 API 契约协作）。
+
+本模块实现（对齐《实现方案_v2》第 4 节）：
+- WORKSPACE_SCHEMA：draft-07 风格 schema 常量（文档/校验依据）
+- validate()：结构 + 字段长度校验（写入前必过，D2/D9）
+- 锚点寻址：a://<file>#L<start>-<end>（引用工件片段，替代全文复制）
+- 双渲染函数：render_for_architect（≤1200 token）、render_for_worker（≤8K token）
+- rollup()：已完成步骤折叠为 archive 摘要行；超限压缩（只减不删，4.5）
+- 状态机：draft -> in_progress -> reviewing -> done / escalated / failed
+"""
+from __future__ import annotations
+
+import copy
+import json
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+VERSION = "1.0"
+
+# ---------------------------------------------------------------------------
+# 字段长度上限（同时是 rollup 依据，4.2）
+# ---------------------------------------------------------------------------
+LIMITS = {
+    "goal": 500,          # brief.goal 字数
+    "task": 300,          # brief.plan[].task 字数
+    "summary": 200,       # progress[].summary 字数
+    "issue_text": 300,    # issues[].observed/expected/tried/ask 字数
+    "reply": 600,         # decisions[].reply 字数
+    "archive": 160,       # archive[] 每条字数
+    "constraints": 8,     # brief.constraints 上限条数
+    "plan_steps": 5,      # brief.plan 上限步数
+    "acceptance": 20,     # brief.acceptance 上限条数
+    "query_truncate": 200,  # render_for_architect 中 query 截断
+}
+
+# 允许的领域标签（4.2 brief.tags；仅用于安全标记与验证接地，不做路由 D3）
+ALLOWED_TAGS = {"code", "math", "legal", "medical", "finance",
+                "life", "education", "general", "safety"}
+
+STATUS_FLOW = {
+    "draft": {"in_progress"},
+    "in_progress": {"reviewing", "escalated", "failed", "in_progress"},
+    "reviewing": {"done", "in_progress", "failed"},
+    "escalated": {"reviewing", "done", "failed"},
+    "done": set(),
+    "failed": set(),
+}
+
+# ---------------------------------------------------------------------------
+# JSON Schema（draft-07 风格，draft-07 依赖内嵌；供校验与文档参考）
+# ---------------------------------------------------------------------------
+WORKSPACE_SCHEMA: Dict[str, Any] = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "Communication Workspace",
+    "type": "object",
+    "required": ["version", "request_id", "query", "meta"],
+    "additionalProperties": False,
+    "properties": {
+        "version": {"const": VERSION},
+        "request_id": {"type": "string", "minLength": 1},
+        "query": {"type": "string", "minLength": 1},
+        "meta": {
+            "type": "object",
+            "required": ["status", "round", "budget"],
+            "properties": {
+                "status": {"enum": ["draft", "in_progress", "reviewing", "escalated", "done", "failed"]},
+                "round": {"type": "integer", "minimum": 0},
+                "budget": {
+                    "type": "object",
+                    "required": ["api_input_tokens", "api_output_tokens", "api_token_cap", "rounds_cap"],
+                    "properties": {
+                        "api_input_tokens": {"type": "integer", "minimum": 0},
+                        "api_output_tokens": {"type": "integer", "minimum": 0},
+                        "api_token_cap": {"type": "integer", "minimum": 1},
+                        "rounds_cap": {"type": "integer", "minimum": 1},
+                    },
+                },
+            },
+        },
+        "brief": {
+            "type": "object",
+            "required": ["goal", "constraints", "tags", "acceptance", "plan"],
+            "properties": {
+                "goal": {"type": "string"},
+                "constraints": {"type": "array", "items": {"type": "string"}},
+                "tags": {"type": "array", "items": {"type": "string"}},
+                "acceptance": {"type": "array", "items": {"type": "object"}},
+                "plan": {"type": "array", "items": {"type": "object"}},
+            },
+        },
+        "progress": {"type": "array", "items": {"type": "object"}},
+        "issues": {"type": "array", "items": {"type": "object"}},
+        "decisions": {"type": "array", "items": {"type": "object"}},
+        "archive": {"type": "array", "items": {"type": "string"}},
+    },
+}
+
+# ---------------------------------------------------------------------------
+# 工具
+# ---------------------------------------------------------------------------
+_TOKEN_PER_CHAR_ZH = 1 / 1.6    # 中文约 1.6 字/token
+_TOKEN_PER_CHAR_EN = 1 / 4.0    # 英文约 4 字/token
+
+
+def estimate_tokens(text: str) -> int:
+    """粗略 token 估算（中英混合，用于渲染预算校验）。"""
+    if not text:
+        return 0
+    zh = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff")
+    en = len(text) - zh
+    return max(1, int(zh * _TOKEN_PER_CHAR_ZH + en * _TOKEN_PER_CHAR_EN))
+
+
+def build_anchor(filename: str, start: int = 1, end: Optional[int] = None) -> str:
+    """构造锚点：a://<file>#L<start>-<end>；end 缺省仅 L<start>。"""
+    if end is None:
+        return f"a://{filename}#L{start}"
+    return f"a://{filename}#L{start}-{end}"
+
+
+_ANCHOR_RE = re.compile(r"^a://(?P<file>[^#]+?)(?:#L(?P<start>\d+)(?:-(?P<end>\d+))?)?$")
+
+
+def parse_anchor(anchor: str) -> Optional[Dict[str, Any]]:
+    """解析锚点为 {file, start, end}；非法返回 None。"""
+    m = _ANCHOR_RE.match(anchor)
+    if not m:
+        return None
+    start = int(m.group("start")) if m.group("start") else 1
+    end = int(m.group("end")) if m.group("end") else start
+    return {"file": m.group("file"), "start": start, "end": end}
+
+
+def _clip(text: str, limit: int) -> str:
+    """按字数截断（中文按字符）。"""
+    if len(text) <= limit:
+        return text
+    return text[:limit] + "…"
+
+
+# ---------------------------------------------------------------------------
+# 校验
+# ---------------------------------------------------------------------------
+def validate(ws: Dict[str, Any]) -> List[str]:
+    """校验 workspace 结构 + 字段长度。返回错误列表（空 = 合法）。"""
+    errors: List[str] = []
+    if not isinstance(ws, dict):
+        return ["workspace 必须是 object"]
+    if ws.get("version") != VERSION:
+        errors.append(f"version 必须是 {VERSION}")
+    if not isinstance(ws.get("request_id"), str) or not ws["request_id"]:
+        errors.append("request_id 必须是非空字符串")
+    if not isinstance(ws.get("query"), str) or not ws["query"]:
+        errors.append("query 必须是非空字符串")
+
+    meta = ws.get("meta")
+    if not isinstance(meta, dict):
+        errors.append("meta 必须是 object")
+    else:
+        if meta.get("status") not in STATUS_FLOW:
+            errors.append(f"meta.status 非法: {meta.get('status')}")
+        budget = meta.get("budget")
+        if not isinstance(budget, dict):
+            errors.append("meta.budget 必须是 object")
+        else:
+            for k in ("api_input_tokens", "api_output_tokens", "api_token_cap", "rounds_cap"):
+                if not isinstance(budget.get(k), int) or budget.get(k) < 0:
+                    errors.append(f"meta.budget.{k} 必须是非负整数")
+
+    brief = ws.get("brief")
+    if brief is not None:
+        if not isinstance(brief, dict):
+            errors.append("brief 必须是 object")
+        else:
+            if not isinstance(brief.get("goal"), str):
+                errors.append("brief.goal 必须是字符串")
+            elif len(brief["goal"]) > LIMITS["goal"]:
+                errors.append(f"brief.goal 超长（>{LIMITS['goal']}字）")
+            if not isinstance(brief.get("constraints"), list):
+                errors.append("brief.constraints 必须是数组")
+            elif len(brief["constraints"]) > LIMITS["constraints"]:
+                errors.append(f"brief.constraints 超过 {LIMITS['constraints']} 条")
+            if not isinstance(brief.get("tags"), list):
+                errors.append("brief.tags 必须是数组")
+            for t in brief.get("tags", []) or []:
+                if t not in ALLOWED_TAGS:
+                    errors.append(f"brief.tags 含非法标签: {t}")
+            acc = brief.get("acceptance")
+            if not isinstance(acc, list) or len(acc) > LIMITS["acceptance"]:
+                errors.append(f"brief.acceptance 需为 ≤{LIMITS['acceptance']} 的数组")
+            plan = brief.get("plan")
+            if not isinstance(plan, list) or len(plan) > LIMITS["plan_steps"]:
+                errors.append(f"brief.plan 需为 ≤{LIMITS['plan_steps']} 步的数组")
+            else:
+                ids = [p.get("id") for p in plan if isinstance(p, dict)]
+                if len(set(ids)) != len(ids):
+                    errors.append("brief.plan 存在重复 step id")
+                for p in plan:
+                    if not isinstance(p, dict):
+                        errors.append("brief.plan 元素必须是 object")
+                        continue
+                    if not isinstance(p.get("task"), str):
+                        errors.append(f"brief.plan[{p.get('id')}].task 必须是字符串")
+                    elif len(p["task"]) > LIMITS["task"]:
+                        errors.append(f"brief.plan[{p.get('id')}].task 超长（>{LIMITS['task']}字）")
+
+    for i, entry in enumerate(ws.get("progress", []) or []):
+        if not isinstance(entry, dict):
+            errors.append(f"progress[{i}] 必须是 object"); continue
+        if entry.get("status") not in ("done", "failed", "blocked"):
+            errors.append(f"progress[{i}].status 非法")
+        if not isinstance(entry.get("summary"), str) or len(entry["summary"]) > LIMITS["summary"]:
+            errors.append(f"progress[{i}].summary 非法或超长")
+
+    for i, entry in enumerate(ws.get("issues", []) or []):
+        if not isinstance(entry, dict):
+            errors.append(f"issues[{i}] 必须是 object"); continue
+        for k in ("observed", "expected", "tried", "ask"):
+            if isinstance(entry.get(k), str) and len(entry[k]) > LIMITS["issue_text"]:
+                errors.append(f"issues[{i}].{k} 超长（>{LIMITS['issue_text']}字）")
+
+    for i, entry in enumerate(ws.get("decisions", []) or []):
+        if not isinstance(entry, dict):
+            errors.append(f"decisions[{i}] 必须是 object"); continue
+        if isinstance(entry.get("reply"), str) and len(entry["reply"]) > LIMITS["reply"]:
+            errors.append(f"decisions[{i}].reply 超长（>{LIMITS['reply']}字）")
+
+    for i, line in enumerate(ws.get("archive", []) or []):
+        if not isinstance(line, str) or len(line) > LIMITS["archive"]:
+            errors.append(f"archive[{i}] 非法或超长")
+    return errors
+
+
+# ---------------------------------------------------------------------------
+# Workspace
+# ---------------------------------------------------------------------------
+class Workspace:
+    """交流文本对象：持有状态、执行写入前校验、渲染、rollup、持久化。"""
+
+    def __init__(self, data: Dict[str, Any]):
+        errors = validate(data)
+        if errors:
+            raise ValueError("workspace 校验失败: " + "; ".join(errors[:5]))
+        self._data = data
+        self._brief_locked = False
+
+    # ---------- 构造 ----------
+    @classmethod
+    def new(cls, request_id: str, query: str,
+            api_token_cap: int = 8000, rounds_cap: int = 6) -> "Workspace":
+        data = {
+            "version": VERSION,
+            "request_id": request_id,
+            "query": query,
+            "meta": {
+                "status": "draft",
+                "round": 0,
+                "budget": {
+                    "api_input_tokens": 0,
+                    "api_output_tokens": 0,
+                    "api_token_cap": api_token_cap,
+                    "rounds_cap": rounds_cap,
+                },
+            },
+            "brief": None,
+            "progress": [],
+            "issues": [],
+            "decisions": [],
+            "archive": [],
+        }
+        return cls(data)
+
+    # ---------- 访问 ----------
+    @property
+    def request_id(self) -> str:
+        return self._data["request_id"]
+
+    @property
+    def status(self) -> str:
+        return self._data["meta"]["status"]
+
+    @property
+    def data(self) -> Dict[str, Any]:
+        return copy.deepcopy(self._data)
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._data.get(key, default)
+
+    def __getitem__(self, key: str) -> Any:
+        return self._data[key]
+
+    def meta(self) -> Dict[str, Any]:
+        return self._data["meta"]
+
+    def budget(self) -> Dict[str, int]:
+        return self._data["meta"]["budget"]
+
+    # ---------- 写入（均先校验） ----------
+    def _commit(self, data: Dict[str, Any]) -> None:
+        errors = validate(data)
+        if errors:
+            raise ValueError("写入校验失败: " + "; ".join(errors[:5]))
+        self._data = data
+
+    def transition(self, new_status: str) -> None:
+        cur = self.status
+        if new_status == cur:
+            return
+        if new_status not in STATUS_FLOW.get(cur, set()):
+            raise ValueError(f"非法状态迁移: {cur} -> {new_status}")
+        self._data["meta"]["status"] = new_status
+
+    def apply_brief(self, brief: Dict[str, Any]) -> None:
+        """写入 brief（写一次后锁定，D2：brief 恒定位于文档前部，prefix cache 友好）。"""
+        if self._data.get("brief") is not None or self._brief_locked:
+            raise ValueError("brief 已写入，不可重复")
+        new = copy.deepcopy(self._data)
+        new["brief"] = brief
+        new["meta"]["status"] = "in_progress"
+        self._commit(new)
+        self._brief_locked = True
+
+    def add_progress(self, step: str, status: str, summary: str,
+                     artifact: Optional[str] = None) -> None:
+        new = copy.deepcopy(self._data)
+        entry: Dict[str, Any] = {"step": step, "status": status, "summary": summary}
+        if artifact:
+            entry["artifact"] = artifact
+        new["progress"].append(entry)
+        self._commit(new)
+
+    def add_issue(self, step: str, anchor: str, observed: str, expected: str,
+                  tried: str, ask: str) -> str:
+        new = copy.deepcopy(self._data)
+        iid = f"i{len(new['issues']) + 1}"
+        entry = {
+            "id": iid, "step": step, "anchor": anchor,
+            "observed": observed, "expected": expected,
+            "tried": tried, "ask": ask,
+        }
+        new["issues"].append(entry)
+        self._commit(new)
+        return iid
+
+    def add_decision(self, ref: str, reply: str,
+                     patch_plan: Optional[List[Dict[str, str]]] = None) -> None:
+        new = copy.deepcopy(self._data)
+        new["decisions"].append({
+            "ref": ref, "reply": reply,
+            "patch_plan": patch_plan or [],
+        })
+        self._commit(new)
+
+    def mark_round(self) -> None:
+        self._data["meta"]["round"] += 1
+
+    def add_budget(self, input_tokens: int = 0, output_tokens: int = 0) -> None:
+        b = self._data["meta"]["budget"]
+        b["api_input_tokens"] += int(input_tokens)
+        b["api_output_tokens"] += int(output_tokens)
+        self._commit(self._data)
+
+    def exhausted(self) -> bool:
+        """预算熔断判定：API token 或回合任一触顶（D6）。"""
+        b = self._data["meta"]["budget"]
+        used = b["api_input_tokens"] + b["api_output_tokens"]
+        if b["api_token_cap"] and used >= b["api_token_cap"]:
+            return True
+        if b["rounds_cap"] and self._data["meta"]["round"] >= b["rounds_cap"]:
+            return True
+        return False
+
+    # ---------- rollup（4.5） ----------
+    def rollup(self) -> int:
+        """把 done 的 progress 折叠为 archive 摘要行，并清理解析完成的问题。
+
+        只减不删 archive 历史；progress 中 done 的条目折叠后移除（保留 failed/blocked）。
+        返回本次折叠的条目数。
+        """
+        folded = 0
+        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"])
+                if line not in self._data["archive"]:
+                    self._data["archive"].append(line)
+                folded += 1
+            else:
+                new_progress.append(entry)
+        resolved = {d.get("ref") for d in self._data.get("decisions", [])}
+        kept_issues: List[Dict[str, Any]] = []
+        for iss in self._data.get("issues", []):
+            if iss.get("id") in resolved:
+                line = _clip(f"{iss['id']}: {iss.get('expected', '')[:60]}", LIMITS["archive"])
+                if line not in self._data["archive"]:
+                    self._data["archive"].append(line)
+            else:
+                kept_issues.append(iss)
+        self._data["issues"] = kept_issues
+        self._data["progress"] = new_progress
+        return folded
+
+    # ---------- 渲染 ----------
+    def render_for_architect(self) -> str:
+        """渲染 Architect 输入（D7）：meta+query(截断)+全部 issues+最近3条 decisions
+        +最近回合 progress 摘要。目标 ≤1200 token。"""
+        d = self._data
+        parts: List[str] = []
+        m = d["meta"]
+        parts.append("== meta ==")
+        parts.append(f"status={m['status']} round={m['round']} "
+                     f"budget={json.dumps(m['budget'], ensure_ascii=False)}")
+        parts.append("== query ==")
+        parts.append(_clip(d["query"], LIMITS["query_truncate"]))
+        if d.get("brief"):
+            b = d["brief"]
+            parts.append("== brief(锁定) ==")
+            parts.append(f"goal: {_clip(b['goal'], 120)}")
+            parts.append(f"plan: {[p['id'] for p in b.get('plan', [])]}")
+            parts.append(f"acceptance: {[a.get('id') for a in b.get('acceptance', [])]}")
+        parts.append("== issues ==")
+        for iss in d.get("issues", []):
+            parts.append(f"{iss['id']} step={iss.get('step')} anchor={iss.get('anchor')} "
+                         f"ask={_clip(iss.get('ask', ''), 80)}")
+        parts.append("== 最近 3 条 decisions ==")
+        for dec in d.get("decisions", [])[-3:]:
+            parts.append(f"ref={dec.get('ref')} reply={_clip(dec.get('reply', ''), 80)}")
+        parts.append("== progress 摘要 ==")
+        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:]:
+            parts.append(line)
+        # token 预算：超限先截断最旧 archive（已只保留 3 条 decisions）
+        out = "\n".join(parts)
+        while estimate_tokens(out) > 1200 and len(d.get("archive", [])) > 4:
+            d = copy.deepcopy(d)
+            d["archive"] = d["archive"][4:]
+            out = "\n".join(_rerender(self, d))
+        return out
+
+    def render_for_worker(self, step_id: str,
+                          artifact_text: Optional[str] = None) -> str:
+        """渲染 Worker 输入（4.4）：brief 全文+该 step 定义+依赖 step 的 archive 摘要行
+        +该 step 现有工件全文+验收标准。目标 ≤8K token。"""
+        d = self._data
+        b = d.get("brief")
+        parts: List[str] = []
+        if b:
+            parts.append("== 任务目标 (goal) ==")
+            parts.append(b["goal"])
+            parts.append("== 约束 (constraints) ==")
+            parts.extend(f"- {c}" for c in b.get("constraints", []))
+            parts.append("== 全部步骤 (plan) ==")
+            for p in b.get("plan", []):
+                mark = " <-- 当前步" if p.get("id") == step_id else ""
+                parts.append(f"{p['id']}: {p.get('task', '')}{mark}")
+                parts.append(f"   done_criteria: {p.get('done_criteria', '')}")
+        parts.append("== 依赖步摘要 (archive) ==")
+        for line in d.get("archive", [])[-6:]:
+            parts.append(line)
+        if artifact_text:
+            parts.append(f"== 当前步已有工件（{step_id}） ==")
+            parts.append(artifact_text)
+        parts.append("== 验收标准 ==")
+        if b:
+            for a in b.get("acceptance", []):
+                parts.append(f"- {a.get('id')}: {a.get('check', '')} "
+                             f"(machine_checkable={a.get('machine_checkable', False)})")
+        parts.append("== 要求 ==")
+        parts.append("请实现当前步，并用可执行验证/事实对照/结构检查自验证；"
+                     "通过则写 progress(done)，失败自修 ≤2 次，仍失败则写 issue。")
+        return "\n".join(parts)
+
+    # ---------- 持久化 ----------
+    def save(self, path: Path) -> None:
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(self._data, f, ensure_ascii=False, indent=2)
+
+    @classmethod
+    def load(cls, path: Path) -> "Workspace":
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        return cls(data)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return self.data
+
+
+def _rerender(ws: "Workspace", d: Dict[str, Any]) -> List[str]:
+    """用裁剪后的数据重建 Architect 渲染（供超限压缩内部用）。"""
+    parts: List[str] = []
+    m = d["meta"]
+    parts.append("== meta ==")
+    parts.append(f"status={m['status']} round={m['round']}")
+    parts.append("== query ==")
+    parts.append(_clip(d["query"], LIMITS["query_truncate"]))
+    parts.append("== issues ==")
+    for iss in d.get("issues", []):
+        parts.append(f"{iss['id']} step={iss.get('step')} ask={_clip(iss.get('ask', ''), 80)}")
+    parts.append("== decisions(最近3) ==")
+    for dec in d.get("decisions", [])[-3:]:
+        parts.append(f"ref={dec.get('ref')} reply={_clip(dec.get('reply', ''), 80)}")
+    parts.append("== progress ==")
+    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", [])[-6:]:
+        parts.append(line)
+    return parts
diff --git a/tests/test_workspace.py b/tests/test_workspace.py
new file mode 100644
index 0000000..487af2f
--- /dev/null
+++ b/tests/test_workspace.py
@@ -0,0 +1,231 @@
+"""T4 交流文本 Workspace 单测（封闭，纯逻辑）。"""
+import json
+
+import pytest
+
+from router_system.workspace import (
+    LIMITS,
+    STATUS_FLOW,
+    Workspace,
+    build_anchor,
+    estimate_tokens,
+    parse_anchor,
+    validate,
+)
+
+ALLOWED_BRIEF = {
+    "goal": "用 Python 实现快速排序并解释复杂度",
+    "constraints": ["必须用标准库", "时间复杂度 O(n log n)"],
+    "tags": ["code"],
+    "acceptance": [{"id": "a1", "check": "排序结果正确", "machine_checkable": True}],
+    "plan": [
+        {"id": "s1", "task": "实现快排主函数", "deps": [], "done_criteria": "可运行"},
+        {"id": "s2", "task": "补充单元测试", "deps": ["s1"], "done_criteria": "3/3 通过"},
+    ],
+}
+
+
+def _ws(**kw):
+    return Workspace.new(request_id=kw.get("request_id", "abc123def456"),
+                         query=kw.get("query", "写个快排"),
+                         api_token_cap=kw.get("api_token_cap", 8000),
+                         rounds_cap=kw.get("rounds_cap", 6))
+
+
+# ---------- 结构/校验 ----------
+def test_new_has_valid_defaults():
+    ws = _ws()
+    assert ws.status == "draft"
+    assert validate(ws.data) == []
+    b = ws.budget()
+    assert b["api_token_cap"] == 8000 and b["rounds_cap"] == 6
+
+
+def test_brief_write_once():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    assert ws.status == "in_progress"
+    with pytest.raises(ValueError):
+        ws.apply_brief(ALLOWED_BRIEF)  # 二次写入被拒
+
+
+def test_apply_brief_missing_field_rejected():
+    ws = _ws()
+    bad = dict(ALLOWED_BRIEF)
+    del bad["plan"]
+    with pytest.raises(ValueError):
+        ws.apply_brief(bad)
+
+
+def test_apply_brief_invalid_tag_rejected():
+    ws = _ws()
+    bad = dict(ALLOWED_BRIEF, tags=["hacker"])
+    with pytest.raises(ValueError):
+        ws.apply_brief(bad)
+
+
+def test_apply_brief_overlong_goal_rejected():
+    ws = _ws()
+    bad = dict(ALLOWED_BRIEF, goal="长" * (LIMITS["goal"] + 1))
+    with pytest.raises(ValueError):
+        ws.apply_brief(bad)
+
+
+def test_apply_brief_too_many_plan_steps_rejected():
+    ws = _ws()
+    plan = [{"id": f"s{i}", "task": "t", "deps": [], "done_criteria": "c"}
+            for i in range(LIMITS["plan_steps"] + 1)]
+    bad = dict(ALLOWED_BRIEF, plan=plan)
+    with pytest.raises(ValueError):
+        ws.apply_brief(bad)
+
+
+# ---------- 状态机 ----------
+def test_status_transitions():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.transition("reviewing")
+    ws.transition("done")
+    assert ws.status == "done"
+    with pytest.raises(ValueError):
+        ws.transition("in_progress")  # done 之后不允许再回
+
+
+def test_illegal_transition():
+    ws = _ws()
+    with pytest.raises(ValueError):
+        ws.transition("done")  # draft 不能直接到 done
+
+
+# ---------- 写入 + 预算 ----------
+def test_progress_and_issues_and_decisions():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_progress("s1", "done", "实现完成", artifact="a://s1_main.py")
+    iid = ws.add_issue("s2", "a://s2_tests.py#L12-18",
+                       "测试失败", "3/3 通过", "修了两次", "请裁决")
+    assert iid == "i1"
+    ws.add_decision(iid, "改用排序后断言", [{"id": "s2", "task": "修测试"}])
+    assert len(ws["progress"]) == 1
+    assert len(ws["issues"]) == 1
+    assert len(ws["decisions"]) == 1
+
+
+def test_budget_accumulation_and_exhaustion():
+    ws = _ws(api_token_cap=10)
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_budget(input_tokens=6, output_tokens=2)
+    assert ws.exhausted() is False
+    ws.add_budget(input_tokens=3)  # 累计 11 > 10
+    assert ws.exhausted() is True
+
+
+def test_round_exhaustion():
+    ws = _ws(rounds_cap=2)
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.mark_round()
+    ws.mark_round()
+    assert ws.exhausted() is True
+
+
+def test_issue_text_overlong_rejected():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    with pytest.raises(ValueError):
+        ws.add_issue("s1", "a://f.py", "观" * (LIMITS["issue_text"] + 1), "e", "t", "q")
+
+
+# ---------- rollup ----------
+def test_rollup_folds_done_progress_and_resolved_issues():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_progress("s1", "done", "快排实现完成", "a://s1_main.py")
+    ws.add_progress("s2", "failed", "测试未过")
+    iid = ws.add_issue("s2", "a://s2_tests.py#L1", "失败", "通过", "试过", "问")
+    ws.add_decision(iid, "已修复")
+    n = ws.rollup()
+    assert n == 1  # 折叠 1 条 done progress
+    assert "s1" in "".join(ws["archive"])
+    # s2 failed 保留；已 resolve 的 issue 从 issues 移除
+    steps = [p["step"] for p in ws["progress"]]
+    assert steps == ["s2"]
+    assert len(ws["issues"]) == 0
+
+
+def test_rollup_keeps_unresolved_issue():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_progress("s1", "done", "ok", "a://s1.py")
+    ws.add_issue("s2", "a://s2.py#L1", "obs", "exp", "try", "ask")  # 无 decision
+    ws.rollup()
+    assert len(ws["issues"]) == 1
+
+
+# ---------- 渲染 ----------
+def test_render_for_architect_within_budget():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    for i in range(6):
+        ws.add_progress(f"s{i}", "done", f"步骤{i}完成" * 3)
+        ws.add_issue(f"s{i}", f"a://f{i}.py#L1", "obs", "exp", "try", "ask" * 10)
+        ws.add_decision(f"i{i+1}", "裁决" * 20)
+    out = ws.render_for_architect()
+    assert estimate_tokens(out) <= 1200
+    assert "== meta ==" in out
+
+
+def test_render_for_worker_has_step_and_criteria():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    out = ws.render_for_worker("s2", artifact_text="def test(): pass")
+    assert "当前步" in out
+    assert "s2" in out
+    assert "def test(): pass" in out
+    assert "验收标准" in out and "a1" in out
+
+
+def test_render_for_worker_marks_current_step():
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    out = ws.render_for_worker("s1")
+    assert "<-- 当前步" in out
+
+
+# ---------- 锚点 ----------
+def test_anchor_build_and_parse():
+    a = build_anchor("s1_main.py", 12, 18)
+    assert a == "a://s1_main.py#L12-18"
+    p = parse_anchor(a)
+    assert p == {"file": "s1_main.py", "start": 12, "end": 18}
+    # 无行号
+    assert parse_anchor("a://f.py") == {"file": "f.py", "start": 1, "end": 1}
+    assert parse_anchor("not-an-anchor") is None
+
+
+# ---------- 持久化 ----------
+def test_save_load_roundtrip(tmp_path):
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_progress("s1", "done", "ok")
+    path = tmp_path / "ws.json"
+    ws.save(path)
+    loaded = Workspace.load(path)
+    assert loaded.to_dict() == ws.to_dict()
+    assert loaded.request_id == ws.request_id
+
+
+def test_json_serializable(tmp_path):
+    ws = _ws()
+    ws.apply_brief(ALLOWED_BRIEF)
+    ws.add_issue("s1", "a://f.py#L1", "obs", "exp", "try", "ask")
+    # 应能被 json 直接序列化（中文 ensure_ascii=False）
+    s = json.dumps(ws.to_dict(), ensure_ascii=False)
+    assert json.loads(s)["request_id"] == ws.request_id
+
+
+def test_validate_rejects_non_object():
+    assert validate("nope") != []
+
+
+def test_status_flow_enumeration():
+    assert set(STATUS_FLOW.keys()) == {"draft", "in_progress", "reviewing", "escalated", "done", "failed"}
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index 81dd85c..db48aac 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -70,7 +70,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T1 | 环境与基线确认（126 测试全绿；README 环境备忘） | ✅ 完成 | （并入 T2 commit） |
 | T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
 | T3 | ArchitectClient（DeepSeek API，JSON 约束） | ⬜ | |
-| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ⬜ | |
+| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ✅ 完成 | T4 |
 | T5 | WorkerLoop + 接地验证 | ⬜ | |
 | T6 | CollaborativePipeline 编排 | ⬜ | |
 | T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | |
@@ -81,28 +81,3 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T12 | 实验脚本 bench_tokens.py + 数据集 | ⬜ | |
 | T13 | E1–E5 跑数到 research/v2_experiments/ | ⬜ | |
 | T14 | 文档收口（README v2 改写） | ⬜ | |
-
-
----
-
-## 五、v2 任务登记（端云协同 LLM 协作系统，见《实现方案_v2_端云协同LLM协作系统.md》）
-
-> 每个任务一个 commit（`feat(v2): Tn 描述`），交付含封闭单测；v1 的 126 项测试保持全绿。
-
-| T | 内容 | 状态 | commit |
-|---|------|------|--------|
-| T1 | 环境与基线确认（126 测试全绿；README 环境备忘） | ✅ 完成 | （并入 T2 commit） |
-| T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
-| T3 | ArchitectClient（DeepSeek API，JSON 约束） | ⬜ | |
-| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ⬜ | |
-| T5 | WorkerLoop + 接地验证 | ⬜ | |
-| T6 | CollaborativePipeline 编排 | ⬜ | |
-| T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | |
-| T8 | 人工检验队列 ReviewQueue | ⬜ | |
-| T9 | token 计量与账单 | ⬜ | |
-| T10 | rollup + prefix cache 调优 | ⬜ | |
-| T11 | 打包分发 setup_runtime.py | ⬜ | |
-| T12 | 实验脚本 bench_tokens.py + 数据集 | ⬜ | |
-| T13 | E1–E5 跑数到 research/v2_experiments/ | ⬜ | |
-| T14 | 文档收口（README v2 改写） | ⬜ | |
-