feat(v2): T4 Workspace 交流文本协议（schema/校验/渲染/rollup）

2026-08-30 21:03:52 +08:00
parent 78a410b773
commit 3ed37a1d30
4 changed files with 861 additions and 75 deletions
@@ -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"}