From fabcd4ff914753a90a5d8a6d0021c02140279892 Mon Sep 17 00:00:00 2001 From: zizi Date: Sun, 23 Aug 2026 01:26:03 +0800 Subject: [PATCH] =?UTF-8?q?=E7=94=9F=E4=BA=A7=E9=93=BE:=20=E8=A1=A5?= =?UTF-8?q?=E8=AF=81/=E9=87=8D=E5=86=99=E5=88=A0=E9=99=A4=E4=B8=9A?= =?UTF-8?q?=E5=8A=A1=E7=A1=AC=E4=B8=8A=E9=99=90=EF=BC=8C=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E6=8E=88=E6=9D=83=E7=BB=88=E6=80=81=EF=BC=88=E9=98=B6=E6=AE=B5?= =?UTF-8?q?E=E7=AC=AC=E4=B8=80=E9=83=A8=E5=88=86=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../architecture/domains/05-创作流程领域.md | 4 +- .agent/skills/write-next-chapter/SKILL.md | 4 +- .../scripts/produce_next_chapter.py | 8 +- .../scripts/run_writer_pipeline.py | 84 +++++++---------- dashboard/server.py | 2 +- ...8-22-阶段E-生产链切换-第一部分-授权终态.md | 50 ++++++++++ .../test_writer_acceptance.py | 2 +- .../write-next-chapter/test_candidate_cas.py | 63 +++++++------ .../test_production_evidence_reassemble.py | 67 ++++++------- .../test_run_writer_pipeline.py | 94 ++++++++++--------- 10 files changed, 214 insertions(+), 164 deletions(-) create mode 100644 docs/plans/2026-08-22-阶段E-生产链切换-第一部分-授权终态.md diff --git a/.agent/docs/architecture/domains/05-创作流程领域.md b/.agent/docs/architecture/domains/05-创作流程领域.md index 9870c50..e12c8c6 100644 --- a/.agent/docs/architecture/domains/05-创作流程领域.md +++ b/.agent/docs/architecture/domains/05-创作流程领域.md @@ -262,7 +262,7 @@ DRAFT/CHECKING/PASSED -> DISCARDED（用户明确决策） 正文、规划、提取、检测和评审分别由单一职责角色执行，主代理不把多个角色合成一次模型调用。 -当前生产编排实现（`run_writer_pipeline`）的事实：机械门 → 语义检测 → 有限补证/重写环、CAS 乐观锁 + 失败收敛终态、生成篇幅与机械接受底线分层合同、评测候选四层强制；代码层的 `MAX_EVIDENCE_REQUESTS=3`、`MAX_REWRITES=2` 是技术保护，业务继续/停止的授权归人（见本节目标合同）。生产写手收到 4000–7000 汉字（目标 7000），可信适配器只拒绝不满 3001 汉字或超过 10000 汉字的异常输出；两层区间由 WriterContext 机械校验保持嵌套。 +当前生产编排实现（`run_writer_pipeline`）的事实：机械门 → 语义检测 → 单次收敛、证据缺口收敛为授权终态（AUTHORIZATION_REQUIRED，补证/重写经人授权后以新运行继续）、CAS 乐观锁 + 失败收敛终态、生成篇幅与机械接受底线分层合同、评测候选四层强制；业务继续/停止的授权归人，技术保护（超时、预算、取消）在执行策略层。生产写手收到 4000–7000 汉字（目标 7000），可信适配器只拒绝不满 3001 汉字或超过 10000 汉字的异常输出；两层区间由 WriterContext 机械校验保持嵌套。 ## 6. 运行与落库 @@ -276,7 +276,7 @@ DRAFT/CHECKING/PASSED -> DISCARDED（用户明确决策） > - 状态机持久化：**已建**——运行级 CAS 链 `example_candidate_cas`，业务态落 `example_candidate`；ARCHIVED 态未启用。 > - 结构化事实增量：**已建**——`example_fact_delta`（提案）+ `example_fact_ledger`（正典账本）；模型只提六型闭集增量且必须带正文证据引文，只有用户批准的增量随正文同事务入账本，抽取结果不自动升格。 > - 待建：生产链上的质量评分环节（盲评仍只接离线评测）；章后抽取的异步执行（接受时只登记 pending 投影）。 -> - 补证重组装：**已建**——语义 `needs_evidence` 时 `production_evidence_reassemble` 检索本作品 as_of 内正文/实体摘录并追加 factEvidence；命中则有限重写（技术保护上限见 §5；业务继续授权归人）。零命中不禁止写手发明新设定、不重写，缺口升格为 `newSettingCandidates`，当前候选进人闸。新设定是否进入正典由人决定；与既有正典冲突仍失败关闭。 +> - 补证重组装：**已建**——语义 `needs_evidence` 时 `production_evidence_reassemble` 检索本作品 as_of 内正文/实体摘录并追加 factEvidence；命中则停在授权终态（AUTHORIZATION_REQUIRED），重写经人授权后以新运行继续。零命中不禁止写手发明新设定、不重写，缺口升格为 `newSettingCandidates`，当前候选进人闸。新设定是否进入正典由人决定；与既有正典冲突仍失败关闭。 ## 7. 开发评测边界 diff --git a/.agent/skills/write-next-chapter/SKILL.md b/.agent/skills/write-next-chapter/SKILL.md index 929017d..990a8b6 100644 --- a/.agent/skills/write-next-chapter/SKILL.md +++ b/.agent/skills/write-next-chapter/SKILL.md @@ -34,7 +34,7 @@ Writer 不接收 `runId`、权限信息、manifest、hash、候选版本、验 2. 伏笔只按细纲动作执行:说埋就埋、说推就推、说收就收;不擅自提前回收,不新开大坑。 3. 前情衔接与上一章末场景无缝;章末钩子按文风画像的钩子风格。 4. 缺少细纲字段、`factConstraints` 字段或篇幅合同属于 adapter 输入错误,必须在模型调用前失败。`factConstraints=[]` 在冻结检索确实没有可确认事实时是合法输入，不等于“事实已验证”。写手可以在正文里设计新设定，但不得把新设定冒充已确认事实。 -5. detector 只把「对已有正典/前文章节的主张检索不够」标成 `evidenceGaps` 并触发补证重写。写手新写出的设定进 `newSettingCandidates`，不因此重写或禁写；与既有正典冲突才失败关闭。新设定是否进入正典由人决定。Writer 输出不承载补证请求或审查结论。 +5. detector 只把「对已有正典/前文章节的主张检索不够」标成 `evidenceGaps`，触发授权终态（补证/重写由人授权后以新运行继续）。写手新写出的设定进 `newSettingCandidates`，不因此重写或禁写；与既有正典冲突才失败关闭。新设定是否进入正典由人决定。Writer 输出不承载补证请求或审查结论。 ## 生产入口 @@ -42,7 +42,7 @@ Dashboard 与人工生产入口统一调用 `scripts/produce_next_chapter.py`。 ## 生产落库 -- 生产编排走 `run_writer_pipeline`（机械门→语义 detector→补证/重写有限环），状态链用 `scripts/candidate_cas.py` 的 `PostgresCasStateStore` 持久化到 `example_candidate_cas`（一次运行一条链，revision 单调，DB 触发器锁方向闭集）；内存 `InMemoryCasStateStore` 仅供离线测试。 +- 生产编排走 `run_writer_pipeline`（机械门→语义 detector→单次收敛；证据缺口收敛为 AUTHORIZATION_REQUIRED 授权终态，补证/重写经人授权后以新运行继续），状态链用 `scripts/candidate_cas.py` 的 `PostgresCasStateStore` 持久化到 `example_candidate_cas`（一次运行一条链，revision 单调，DB 触发器锁方向闭集）；内存 `InMemoryCasStateStore` 仅供离线测试。 - `run_writer_with_receipt()` 只负责可信 writer adapter；生产编排在组装后调用 `assemble-context/scripts/persist_context_freeze.py`，审查落库调用 `scripts/persist_writer_run.py`。 - `persist_writer_run.py` 要求本次 `run_id` 已有成功 writer 调用的 raw 指针，随后登记 `example_candidate`（传入 `semantic_report` 时校验绑定并把 `semantic_status`/`semantic_report_sha256` 固化到候选行）、追加 `example_run_receipt` 和机械/语义两行 `example_quality_result`；它不接受正文，Shadow→Canonical 仍只能由 `decide-candidate/scripts/write_canonical.py` 完成（接受通道 DB 级兜底复检 `state=passed` 且 `semantic_status=passed`）。 diff --git a/.agent/skills/write-next-chapter/scripts/produce_next_chapter.py b/.agent/skills/write-next-chapter/scripts/produce_next_chapter.py index 3b7b5df..ff9329c 100644 --- a/.agent/skills/write-next-chapter/scripts/produce_next_chapter.py +++ b/.agent/skills/write-next-chapter/scripts/produce_next_chapter.py @@ -5,7 +5,7 @@ 读已确认细纲 + 前章正文基线 → build_retrieval_plan + retrieve_writer_sources（生产仓储；新书无卡诚实返空） → assemble_context 冻结 WriterContext v1 - → run_writer_pipeline（持久 CAS 状态链 + 机械门 + 语义 detector，有限补证/重写合同） + → run_writer_pipeline（持久 CAS 状态链 + 机械门 + 语义 detector，单次收敛 + 授权终态合同） writer 真调经 run_writer_with_receipt：runtime 自动把模型输入/输出原文与调用明细原子落库； 篇幅越界在 writer 适配层自动重抽（最多三遍）；语义 detector 走冻结 detector profile 真调； 语义 needs_evidence 时走 production_evidence_reassemble（检索已有正典摘录）；零命中当新设定交人闸，不禁写不重写 @@ -524,6 +524,12 @@ def main(): print(f"[警告] 被拒候选留库失败: {persist_exc}", file=sys.stderr) finish_run(run_id, "failed", creator="continuation", trigger_detail={"stage": "writer-production-pipeline", "failureCode": exc.code}) + if exc.code == "AUTHORIZATION_REQUIRED": + gaps = (exc.details or {}).get("evidenceGaps") or [] + print("[需要授权] 语义检查发现证据缺口，补证或重写需要人授权：") + for item in gaps: + print(f" - {item.get('gapId')}: {item.get('reason')}（检索：{item.get('query')}）") + print("授权后由主代理发起新运行继续补证（本运行已收敛 REJECTED，新运行接续候选版本）。") print(f"[停止] 生产 pipeline 未通过: code={exc.code}；{exc}") print(f"复核 artifacts/{run_id}-pipeline-result.json 后决定下一步。") print(f"\nRUN_ID={run_id}") diff --git a/.agent/skills/write-next-chapter/scripts/run_writer_pipeline.py b/.agent/skills/write-next-chapter/scripts/run_writer_pipeline.py index fc333c9..99a9593 100644 --- a/.agent/skills/write-next-chapter/scripts/run_writer_pipeline.py +++ b/.agent/skills/write-next-chapter/scripts/run_writer_pipeline.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""正文候选的有限补证、重写、机械审查与 CAS 编排。""" +"""正文候选的单次收敛编排：机械审查、语义检查、授权终态与 CAS。""" from __future__ import annotations @@ -30,8 +30,6 @@ from writer_contract import ( # noqa: E402 ) -MAX_EVIDENCE_REQUESTS = 3 -MAX_REWRITES = 2 _HASH_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") @@ -440,7 +438,12 @@ def run_writer_pipeline( result_path: str | pathlib.Path | None = None, initial_candidate_version: int = 1, ) -> dict[str, Any]: - """执行最多 3 次补证、2 次重写的正文候选有限收敛循环。 + """正文候选单次收敛：写手 -> 机械门 -> 语义检查，全分支终态化。 + + 语义检查发现证据缺口时先做确定性检索（evidence_provider，只读）： + 零命中升格为新设定提案进人闸；命中则收敛为 AUTHORIZATION_REQUIRED 终态， + 携带缺口报告与重组上下文哈希，由编排方在人授权后发起新运行继续重写。 + 补证/重写业务次数不设硬上限，管控归人；授权门只挡模型重调用，不挡检索。 initial_candidate_version 给生产编排接续既有版本号：候选表对 (作品, 章, candidate_version) 唯一，同一章重跑必须从「已有最大版本+1」起， @@ -529,7 +532,7 @@ def run_writer_pipeline( try: candidate = validate_writer_output(candidate) except ContractError as exc: - # 合同错误作为可定位审查失败进入有限重写，而不是绕过状态机。 + # 合同错误作为可定位审查失败进入机械门定位，而不是绕过状态机。 try: mechanical_report = check_writer_candidate(current_context, candidate, requirements) candidate_failures = _validate_mechanical_report(mechanical_report, candidate) @@ -659,36 +662,11 @@ def run_writer_pipeline( result_path=result_path, ) from exc if evidence_gaps: - if evidence_request_count + len(evidence_gaps) > MAX_EVIDENCE_REQUESTS: - raise _terminal_failure( - code="EVIDENCE_REQUEST_LIMIT_REACHED", - message="detector 证据缺口累计超过 3 次", - token=checking, - state_store=state_store, - run_id=run_id, - evidence_request_count=evidence_request_count, - rewrite_count=rewrite_count, - trace=trace, - candidate=candidate, - result_path=result_path, - ) - if rewrite_count >= MAX_REWRITES: - raise _terminal_failure( - code="REWRITE_LIMIT_REACHED", - message="补证后重写已达到 2 次", - token=checking, - state_store=state_store, - run_id=run_id, - evidence_request_count=evidence_request_count, - rewrite_count=rewrite_count, - trace=trace, - candidate=candidate, - result_path=result_path, - ) - evidence_request_count += len(evidence_gaps) + # 缺口先做确定性检索（只读、无模型开销）；结果决定收敛方向： + # 零命中 = 新设定提案，升格 passed 进人闸；命中 = 需要模型重写， + # 停在授权终态等人授权，由编排方以新运行继续，不自动重写。 next_attempt = current_context["attempt"] + 1 previous_snapshot = current_context["contextSnapshot"]["contextSha256"] - previous_creative_input = build_writer_creative_input(current_context) previous_fact_ids = { str(item.get("evidenceId")) for item in (current_context.get("factEvidence") or []) @@ -719,9 +697,9 @@ def run_writer_pipeline( for item in (next_context.get("factEvidence") or []) if isinstance(item, Mapping) } - # 正典检索零命中 = 缺口是新设定提案，不是可补的检索债。 - # 无语义冲突时升格为 passed 交人闸；有冲突则不重写、走失败关闭。 if not (next_fact_ids - previous_fact_ids): + # 正典检索零命中 = 缺口是新设定提案，不是可补的检索债。 + # 无语义冲突时升格为 passed 交人闸；有冲突则不升格、走失败关闭。 if not candidate_failures and semantic_report is not None: semantic_report = _promote_unfillable_gaps_to_settings(semantic_report) evidence_gaps = [] @@ -730,7 +708,6 @@ def run_writer_pipeline( next_context["runId"] != run_id or next_context["attempt"] != next_attempt or next_context["contextSnapshot"]["contextSha256"] == previous_snapshot - or build_writer_creative_input(next_context) == previous_creative_input ): raise _terminal_failure( code="REASSEMBLED_CONTEXT_INVALID", @@ -744,28 +721,35 @@ def run_writer_pipeline( candidate=candidate, result_path=result_path, ) - rejected = _cas_or_fail( - state_store.transition(checking, "REJECTED"), "CHECKING -> REJECTED" - ) - rewrite_count += 1 - candidate_version += 1 - # 补证环的审计痕迹同样携带机械门报告，被拒候选留库不因走补证路径而丢失证据。 trace.append({ "attempt": checking.attempt, "candidateVersion": checking.candidate_version, "candidateSha256": candidate["candidateSha256"], - "status": "evidence_gap", + "status": "needs_authorization", "gapIds": [item["gapId"] for item in evidence_gaps], "mechanicalPassed": mechanical_report.get("passed", False), "mechanicalReport": dict(mechanical_report), - "semanticReport": dict(semantic_report), + "semanticReport": dict(semantic_report) if semantic_report else None, }) - current_context = next_context - draft = _cas_or_fail( - state_store.start_next(rejected, attempt=next_attempt, candidate_version=candidate_version), - "REJECTED -> next DRAFT", + raise _terminal_failure( + code="AUTHORIZATION_REQUIRED", + message="补证检索命中，重写需要人授权（以新运行继续）", + token=checking, + state_store=state_store, + run_id=run_id, + evidence_request_count=evidence_request_count, + rewrite_count=rewrite_count, + trace=trace, + candidate=candidate, + result_path=result_path, + details={ + "evidenceGaps": [dict(item) for item in evidence_gaps], + "mechanicalPassed": mechanical_report.get("passed", False), + "semanticStatus": semantic_report.get("status") if semantic_report else None, + "nextAttempt": next_attempt, + "reassembledContextSha256": next_context["contextSnapshot"]["contextSha256"], + }, ) - continue trace.append( { "attempt": checking.attempt, @@ -824,8 +808,6 @@ def run_writer_pipeline( __all__ = [ - "MAX_EVIDENCE_REQUESTS", - "MAX_REWRITES", "CasToken", "CasStateStore", "InMemoryCasStateStore", diff --git a/dashboard/server.py b/dashboard/server.py index 468d10f..351222c 100644 --- a/dashboard/server.py +++ b/dashboard/server.py @@ -2520,7 +2520,7 @@ def _run_decision_menu_panel(run_id: str, candidates: list, terminal_state, pipe f"写库决策请用 决策通道 :8767" f"（看板只读）

" ) - # 无候选行：本 run 失败关闭常见于 REASSEMBLED_CONTEXT_INVALID + # 无候选行：本 run 失败关闭常见于 AUTHORIZATION_REQUIRED / 语义或机械拒绝 return ( "
决策菜单
" "
" 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()