决策菜单
"
"
"
diff --git a/docs/plans/2026-08-22-阶段E-生产链切换-第一部分-授权终态.md b/docs/plans/2026-08-22-阶段E-生产链切换-第一部分-授权终态.md
new file mode 100644
index 0000000..694e542
--- /dev/null
+++ b/docs/plans/2026-08-22-阶段E-生产链切换-第一部分-授权终态.md
@@ -0,0 +1,50 @@
+# 阶段 E：生产链切换（第一部分：授权终态合同）
+
+日期：2026-08-22
+总 plan：[2026-08-22-agent-example整体收敛总plan.md](2026-08-22-agent-example整体收敛总plan.md)
+状态：第一部分完成并提交；第二部分（写作/检测智能体接入框架派发、授权后继续的接线、真实生产烟测）另行启动，真实模型调用需人工授权。
+
+## 1. 意图
+
+删除补证/重写的业务硬上限（补证≤3 / 重写≤2），改为授权终态：检查发现缺口后系统停在结构化报告，由主代理转述给人，人授权后以新运行继续。技术保护（超时、预算、取消）保留在执行策略层。
+
+## 2. 合同设计（本部分核心决策）
+
+授权门只挡昂贵的模型重调用，不挡廉价的确定性检索：
+
+```text
+语义检查发现证据缺口
+ → 确定性检索（evidence_provider，只读、无模型开销）
+ ├─ 零命中 = 新设定提案 → 升格 passed 进人闸（原行为保留）
+ └─ 命中 = 需要模型重写 → AUTHORIZATION_REQUIRED 授权终态
+ 携带：缺口清单、机械/语义检查快照、下一 attempt、重组上下文哈希
+ 人授权后：编排方以新运行继续（新候选版本，会话可复用）
+```
+
+依据：CAS 链按运行唯一（`ON CONFLICT (run_id) DO NOTHING`），同运行不能开新轮；授权后的继续天然是新运行，与"一章复用同一写作智能体会话"通过框架会话续接（阶段 D 能力）组合。
+
+## 3. 文件台账
+
+| 处置 | 文件 | 原因 |
+|---|---|---|
+| 修改 | `run_writer_pipeline.py` | 删除 `MAX_EVIDENCE_REQUESTS/MAX_REWRITES` 常量、自动补证/重写循环与两个上限错误码；缺口分支改为"检索→零命中升格/命中授权终态"；单次收敛文档化。CAS 基元（含 `start_next`）保留，状态机闭集不动 |
+| 修改 | `produce_next_chapter.py` | AUTHORIZATION_REQUIRED 显式分支：输出结构化缺口报告与授权指引；模块头口径同步 |
+| 修改 | `test_run_writer_pipeline.py` | 两个自动补证测试改写为授权终态合同（缺口停终态、授权报告落结果文件）；provider 改为真实检索推进 |
+| 修改 | `test_production_evidence_reassemble.py` | 命中重写测试改为授权终态（写手单次调用、details 带 nextAttempt 与重组上下文哈希）；零命中升格测试不变 |
+| 修改 | `test_candidate_cas.py` | CAS 上的补证环测试改为授权终态链断言（create→CHECKING→REJECTED，revision=3） |
+| 修改 | `test_writer_acceptance.py` | 失败码 fixture 对齐（REWRITE_LIMIT_REACHED → AUTHORIZATION_REQUIRED） |
+| 修改 | `05-创作流程领域`、`write-next-chapter/SKILL.md`、`dashboard/server.py` 注释 | 补证口径全仓同步（检索语义不变、命中停授权终态） |
+| 保留 | `production_evidence_reassemble.py` | 确定性检索是新合同的组成部分（授权前置检索），职责不变 |
+| 保留 | `start_next` 等 CAS 基元 | 总 plan 不重造状态机；续跑走新运行的 `create` 链 |
+| 遗留 | 授权后继续的编排接线（新运行 + 重组上下文 + 会话续接） | 阶段 E 第二部分 |
+| 遗留 | 写作/检测智能体框架派发接入 | 阶段 E 第二部分；真实烟测需人工授权 |
+
+## 4. 验证
+
+- 管线测试：10 项通过（含授权终态 2 项新合同测试）。
+- 补证重组装测试：5 项通过（命中授权终态 + 零命中升格）。
+- CAS 测试：离线全通过；真实库集成测试（PostgresCasStateStore）全部通过。
+- 候选接受测试：通过（失败码对齐后）。
+- 全量离线清单：102 通过、8 项外部依赖阻断、0 失败。
+- 技能严格审计：58 个 Skill，阻断 0；索引一致性、架构门禁、`git diff --check` 通过。
+- 旧口径零残留：`补证≤3 / 重写≤2 / EVIDENCE_REQUEST_LIMIT / REWRITE_LIMIT` 全仓检索无生产代码引用（写手篇幅适配器的"≤3 遍重抽"是独立技术重试合同，不属于补证/重写，保留）。
diff --git a/tests/skills/decide-candidate/test_writer_acceptance.py b/tests/skills/decide-candidate/test_writer_acceptance.py
index a11b70a..d8383b1 100644
--- a/tests/skills/decide-candidate/test_writer_acceptance.py
+++ b/tests/skills/decide-candidate/test_writer_acceptance.py
@@ -327,7 +327,7 @@ class WriterAcceptanceTest(unittest.TestCase):
rejected_detector = copy.deepcopy(detector)
rejected_detector["status"] = "REJECTED"
- rejected_detector["failureCode"] = "REWRITE_LIMIT_REACHED"
+ rejected_detector["failureCode"] = "AUTHORIZATION_REQUIRED"
cases.append(("DETECTOR_NOT_PASSED", candidate, rejected_detector, live))
wrong_detector_schema = copy.deepcopy(detector)
diff --git a/tests/skills/write-next-chapter/test_candidate_cas.py b/tests/skills/write-next-chapter/test_candidate_cas.py
index b5c99c7..98051e7 100644
--- a/tests/skills/write-next-chapter/test_candidate_cas.py
+++ b/tests/skills/write-next-chapter/test_candidate_cas.py
@@ -207,45 +207,46 @@ class PipelineOnPostgresStoreTest(unittest.TestCase):
self.assertEqual(chain["state"], "PASSED")
self.assertEqual(chain["revision"], 3) # create→CHECKING→PASSED
- def test_pipeline_evidence_loop_uses_start_next(self) -> None:
+ def test_pipeline_evidence_hit_stops_at_authorization(self) -> None:
+ """新合同：补证检索命中后停在授权终态，CAS 收敛 REJECTED，不开新轮。"""
context, _ = _valid_pair()
rows: dict = {}
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
- if candidate["candidateVersion"] == 1:
- report = {
- "schemaVersion": "semantic-detection-v3",
- "runId": current["runId"], "sampleId": "cas-sample",
- "opaqueArmId": "cas-candidate", "inputSha256": "sha256:" + "1" * 64,
- "candidateVersion": candidate["candidateVersion"],
+ report = {
+ "schemaVersion": "semantic-detection-v3",
+ "runId": current["runId"], "sampleId": "cas-sample",
+ "opaqueArmId": "cas-candidate", "inputSha256": "sha256:" + "1" * 64,
+ "candidateVersion": candidate["candidateVersion"],
+ "candidateSha256": candidate["candidateSha256"],
+ "contextSnapshotSha256": current["contextSnapshot"]["contextSha256"],
+ "modelReceiptSha256": "sha256:" + "2" * 64, "status": "needs_evidence",
+ "claims": [], "findings": [], "assertionVerdicts": [],
+ "hardConstraintVerdicts": [], "newSettingCandidates": [],
+ "evidenceGaps": [{
+ "gapId": "gap-1", "query": "补证", "reason": "缺口", "priority": "high",
"candidateSha256": candidate["candidateSha256"],
- "contextSnapshotSha256": current["contextSnapshot"]["contextSha256"],
- "modelReceiptSha256": "sha256:" + "2" * 64, "status": "needs_evidence",
- "claims": [], "findings": [], "assertionVerdicts": [],
- "hardConstraintVerdicts": [], "newSettingCandidates": [],
- "evidenceGaps": [{
- "gapId": "gap-1", "query": "补证", "reason": "缺口", "priority": "high",
- "candidateSha256": candidate["candidateSha256"],
- "candidateQuote": "旧徽章", "startCodePoint": 11, "endCodePoint": 14,
- }],
- }
- report["reportSha256"] = canonical_sha256(report)
- return report
- return _semantic_pass(current, candidate, _mechanical)
+ "candidateQuote": "旧徽章", "startCodePoint": 11, "endCodePoint": 14,
+ }],
+ }
+ report["reportSha256"] = canonical_sha256(report)
+ return report
- result = run_writer_pipeline(
- context=context,
- requirements=_requirements(),
- writer=lambda current, version: _writer_output(current, version),
- evidence_provider=lambda current, _gaps, attempt: _advance_context(current, attempt),
- semantic_detector=detector,
- state_store=_store(rows),
- )
- self.assertEqual(result["status"], "PASSED")
+ with self.assertRaises(PipelineError) as caught:
+ run_writer_pipeline(
+ context=context,
+ requirements=_requirements(),
+ writer=lambda current, version: _writer_output(current, version),
+ evidence_provider=lambda current, _gaps, attempt: _advance_context(current, attempt),
+ semantic_detector=detector,
+ state_store=_store(rows),
+ )
+ self.assertEqual(caught.exception.code, "AUTHORIZATION_REQUIRED")
+ self.assertEqual(caught.exception.details["nextAttempt"], 2)
chain = rows[context["runId"]]
- # create→CHECKING→REJECTED→新DRAFT→CHECKING→PASSED
+ # create→CHECKING→REJECTED：授权终态不开同运行新轮，续跑是新运行的事。
self.assertEqual((chain["state"], chain["revision"], chain["attempt"],
- chain["candidate_version"]), ("PASSED", 6, 2, 2))
+ chain["candidate_version"]), ("REJECTED", 3, 1, 1))
if __name__ == "__main__":
diff --git a/tests/skills/write-next-chapter/test_production_evidence_reassemble.py b/tests/skills/write-next-chapter/test_production_evidence_reassemble.py
index c455eae..2f11e1b 100644
--- a/tests/skills/write-next-chapter/test_production_evidence_reassemble.py
+++ b/tests/skills/write-next-chapter/test_production_evidence_reassemble.py
@@ -21,7 +21,7 @@ from production_evidence_reassemble import ( # noqa: E402
EvidenceReassembleError,
reassemble_writer_context_for_gaps,
)
-from run_writer_pipeline import InMemoryCasStateStore, run_writer_pipeline # noqa: E402
+from run_writer_pipeline import InMemoryCasStateStore, PipelineError, run_writer_pipeline # noqa: E402
from writer_contract import build_writer_creative_input, retrieval_identity # noqa: E402
from test_check_writer_candidate import _requirements, _valid_pair # noqa: E402
@@ -98,8 +98,8 @@ class ProductionEvidenceReassembleTests(unittest.TestCase):
work_id=12,
)
- def test_pipeline_needs_evidence_then_pass_with_reassemble(self) -> None:
- """needs_evidence + 检索命中 → 重写 → 再检 passed。"""
+ def test_pipeline_evidence_hit_stops_at_authorization(self) -> None:
+ """needs_evidence + 检索命中 -> 授权终态（重写需人授权，以新运行继续）。"""
context, _ = _valid_pair()
calls: list[int] = []
@@ -121,25 +121,23 @@ class ProductionEvidenceReassembleTests(unittest.TestCase):
return _writer_output(current, version)
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
- if current["attempt"] == 1:
- return _semantic_report(
- current,
- candidate,
- status="needs_evidence",
- gaps=[
- {
- "gapId": "gap-1",
- "query": "旧徽章归属",
- "reason": "缺口",
- "priority": "high",
- "candidateSha256": candidate["candidateSha256"],
- "candidateQuote": "旧徽章",
- "startCodePoint": 11,
- "endCodePoint": 14,
- }
- ],
- )
- return _semantic_report(current, candidate, status="passed")
+ return _semantic_report(
+ current,
+ candidate,
+ status="needs_evidence",
+ gaps=[
+ {
+ "gapId": "gap-1",
+ "query": "旧徽章归属",
+ "reason": "缺口",
+ "priority": "high",
+ "candidateSha256": candidate["candidateSha256"],
+ "candidateQuote": "旧徽章",
+ "startCodePoint": 11,
+ "endCodePoint": 14,
+ }
+ ],
+ )
def provider(current, gaps, attempt):
with patch(
@@ -150,17 +148,20 @@ class ProductionEvidenceReassembleTests(unittest.TestCase):
current, gaps, attempt, work_id=int(current["workId"])
)
- result = run_writer_pipeline(
- context=context,
- requirements=_requirements(),
- writer=writer,
- evidence_provider=provider,
- semantic_detector=detector,
- state_store=InMemoryCasStateStore(),
- )
- self.assertEqual(result["status"], "PASSED")
- self.assertEqual(calls, [1, 2])
- self.assertGreaterEqual(result.get("rewriteCount", 0), 1)
+ with self.assertRaises(PipelineError) as caught:
+ run_writer_pipeline(
+ context=context,
+ requirements=_requirements(),
+ writer=writer,
+ evidence_provider=provider,
+ semantic_detector=detector,
+ state_store=InMemoryCasStateStore(),
+ )
+ self.assertEqual(caught.exception.code, "AUTHORIZATION_REQUIRED")
+ # 只跑单次收敛：写手只被调用一次，重写等人授权。
+ self.assertEqual(calls, [1])
+ self.assertEqual(caught.exception.details["nextAttempt"], 2)
+ self.assertTrue(caught.exception.details["reassembledContextSha256"].startswith("sha256:"))
def test_unfillable_gaps_pass_without_rewrite(self) -> None:
"""正典零命中的缺口是新设定提案：不重写，报告升格 passed，进人闸。"""
diff --git a/tests/skills/write-next-chapter/test_run_writer_pipeline.py b/tests/skills/write-next-chapter/test_run_writer_pipeline.py
index c3976f1..7950bd4 100644
--- a/tests/skills/write-next-chapter/test_run_writer_pipeline.py
+++ b/tests/skills/write-next-chapter/test_run_writer_pipeline.py
@@ -109,45 +109,45 @@ class RunWriterPipelineV3Test(unittest.TestCase):
for forbidden in ("claimLedger", "evidenceRequests", "newSettingDeclarations"):
self.assertNotIn(forbidden, result["candidateArtifact"])
- def test_only_detector_evidence_gaps_trigger_fresh_attempt(self) -> None:
+ def test_evidence_gaps_stop_at_authorization_terminal(self) -> None:
+ """新合同：补证检索命中后收敛为 AUTHORIZATION_REQUIRED，重写等人授权。"""
context, _ = _valid_pair()
- writer_calls: list[tuple[int, int, str]] = []
- detector_calls: list[int] = []
- provider_calls: list[list[str]] = []
+ writer_calls: list[tuple[int, int]] = []
def writer(current: dict, version: int) -> dict:
- writer_calls.append((current["attempt"], version, current["contextSnapshot"]["contextSha256"]))
+ writer_calls.append((current["attempt"], version))
return _writer_output(current, version)
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
- detector_calls.append(candidate["candidateVersion"])
- if candidate["candidateVersion"] == 1:
- return _semantic_report(current, candidate, status="needs_evidence", gaps=[{
- "gapId": "gap-1", "query": "旧徽章来源", "reason": "证据不足", "priority": "high",
- "candidateSha256": candidate["candidateSha256"], "candidateQuote": "旧徽章",
- "startCodePoint": 11, "endCodePoint": 14,
- }])
- return _semantic_report(current, candidate)
+ return _semantic_report(current, candidate, status="needs_evidence", gaps=[{
+ "gapId": "gap-1", "query": "旧徽章来源", "reason": "证据不足", "priority": "high",
+ "candidateSha256": candidate["candidateSha256"], "candidateQuote": "旧徽章",
+ "startCodePoint": 11, "endCodePoint": 14,
+ }])
- def provider(current: dict, gaps: list[dict], attempt: int) -> dict:
- provider_calls.append([item["gapId"] for item in gaps])
- return _advance_context(current, attempt, add_evidence=True)
-
- result = run_writer_pipeline(
- context=context,
- requirements=_requirements(),
- writer=writer,
- evidence_provider=provider,
- semantic_detector=detector,
- state_store=InMemoryCasStateStore(),
- )
- self.assertEqual(result["status"], "PASSED")
- self.assertEqual(writer_calls[0][:2], (1, 1))
- self.assertEqual(writer_calls[1][:2], (2, 2))
- self.assertNotEqual(writer_calls[0][2], writer_calls[1][2])
- self.assertEqual(detector_calls, [1, 2])
- self.assertEqual(provider_calls, [["gap-1"]])
- self.assertEqual(result["evidenceRequestCount"], 1)
+ store = InMemoryCasStateStore()
+ with tempfile.TemporaryDirectory() as directory:
+ result_path = pathlib.Path(directory) / "result.json"
+ with self.assertRaises(PipelineError) as caught:
+ run_writer_pipeline(
+ context=context,
+ requirements=_requirements(),
+ writer=writer,
+ evidence_provider=lambda current, _gaps, attempt: _advance_context(current, attempt, add_evidence=True),
+ semantic_detector=detector,
+ state_store=store,
+ result_path=result_path,
+ )
+ result = json.loads(result_path.read_text(encoding="utf-8"))
+ self.assertEqual(caught.exception.code, "AUTHORIZATION_REQUIRED")
+ # 只跑单次收敛：写手只被调用一次，无自动第二轮。
+ self.assertEqual(writer_calls, [(1, 1)])
+ self.assertEqual(store.latest(context["runId"]).state, "REJECTED")
+ # 授权终态携带结构化缺口报告，供主代理转述给人。
+ gaps = caught.exception.details["evidenceGaps"]
+ self.assertEqual([item["gapId"] for item in gaps], ["gap-1"])
+ self.assertEqual(result["failureCode"], "AUTHORIZATION_REQUIRED")
+ self.assertEqual(result["status"], "REJECTED")
def test_writer_cannot_trigger_evidence_provider(self) -> None:
context, _ = _valid_pair()
@@ -165,7 +165,8 @@ class RunWriterPipelineV3Test(unittest.TestCase):
code="MECHANICAL_DETECTION_REJECTED",
)
- def test_reassembled_context_must_have_new_snapshot_and_attempt(self) -> None:
+ def test_authorization_terminal_publishes_gap_report(self) -> None:
+ """授权终态结果文件必须带缺口与检查快照，编排方据此向人请求授权。"""
context, _ = _valid_pair()
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
@@ -175,15 +176,24 @@ class RunWriterPipelineV3Test(unittest.TestCase):
"startCodePoint": 11, "endCodePoint": 14,
}])
- same_snapshot = copy.deepcopy(context)
- same_snapshot["attempt"] = 2
- self._assert_failure(
- context=context,
- writer=lambda current, version: _writer_output(current, version),
- detector=detector,
- provider=lambda *_args: same_snapshot,
- code="REASSEMBLED_CONTEXT_INVALID",
- )
+ store = InMemoryCasStateStore()
+ with tempfile.TemporaryDirectory() as directory:
+ result_path = pathlib.Path(directory) / "result.json"
+ with self.assertRaises(PipelineError) as caught:
+ run_writer_pipeline(
+ context=context,
+ requirements=_requirements(),
+ writer=lambda current, version: _writer_output(current, version),
+ evidence_provider=lambda current, _gaps, attempt: _advance_context(current, attempt, add_evidence=True),
+ semantic_detector=detector,
+ state_store=store,
+ result_path=result_path,
+ )
+ result = json.loads(result_path.read_text(encoding="utf-8"))
+ self.assertEqual(caught.exception.code, "AUTHORIZATION_REQUIRED")
+ self.assertEqual(result["candidateSha256"], caught.exception.result["candidateSha256"])
+ self.assertEqual(caught.exception.details["mechanicalPassed"], True)
+ self.assertEqual(caught.exception.details["semanticStatus"], "needs_evidence")
def test_old_semantic_v2_report_fails_closed(self) -> None:
context, _ = _valid_pair()