290 lines
13 KiB
Python
290 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""两阶段写手离线测试（阶段 F 第一部分）：探索与生成分离。
|
|
|
|
固定合同：探索清单严格校验（失败关闭）、资料确定性回放（无模型参与）、
|
|
生成输入只含探索资料与合同部分（不含预组装）、编排两阶段各记各的派发运行、
|
|
探索失败/清单非法一律失败关闭、派发桥的覆盖位与工具开关。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[3]
|
|
SCRIPT_DIR = PROJECT_ROOT / ".agent" / "skills" / "write-next-chapter" / "scripts"
|
|
for path in (SCRIPT_DIR,):
|
|
if str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
import two_phase_writer # noqa: E402
|
|
import dispatch_writer_bridge as bridge # noqa: E402
|
|
from two_phase_writer import ( # noqa: E402
|
|
EXPLORATION_MAX_MATERIALS,
|
|
EXPLORATION_SCHEMA_VERSION,
|
|
ExplorationError,
|
|
build_exploration_spec,
|
|
build_generation_creative_input,
|
|
parse_exploration_manifest,
|
|
replay_manifest_materials,
|
|
run_two_phase_writer,
|
|
)
|
|
|
|
|
|
def _valid_manifest(materials=None, **extra):
|
|
if materials is None:
|
|
materials = [
|
|
{"tool": "read_fine_outline", "args": {"work_id": 12, "target_chapter": 3}, "reason": "硬事件与约束"},
|
|
{"tool": "read_chapter_text", "args": {"work_id": 12, "chapter_order": 2}, "reason": "上一章衔接"},
|
|
]
|
|
data = {
|
|
"schemaVersion": EXPLORATION_SCHEMA_VERSION,
|
|
"requiredMaterials": materials,
|
|
}
|
|
data.update(extra)
|
|
return data
|
|
|
|
|
|
def _fake_projected():
|
|
return {
|
|
"lengthContract": {"targetChars": 7000, "minChars": 4000, "maxChars": 10000, "frontmatterRequired": False},
|
|
"styleConstraints": ["保留第一人称"],
|
|
}
|
|
|
|
|
|
class ParseManifestTest(unittest.TestCase):
|
|
|
|
def test_parse_ok(self):
|
|
data = parse_exploration_manifest(json.dumps(_valid_manifest(), ensure_ascii=False))
|
|
self.assertEqual(len(data["requiredMaterials"]), 2)
|
|
|
|
def test_rejects_bad_schema_version(self):
|
|
bad = _valid_manifest()
|
|
bad["schemaVersion"] = "other-v1"
|
|
with self.assertRaises(ExplorationError) as ctx:
|
|
parse_exploration_manifest(json.dumps(bad, ensure_ascii=False))
|
|
self.assertEqual(ctx.exception.code, "EXPLORATION_MANIFEST_INVALID")
|
|
|
|
def test_rejects_unknown_tool(self):
|
|
bad = _valid_manifest(materials=[{"tool": "drop_table", "args": {}, "reason": "x"}])
|
|
with self.assertRaises(ExplorationError) as ctx:
|
|
parse_exploration_manifest(json.dumps(bad, ensure_ascii=False))
|
|
self.assertIn("不在只读工具登记表", str(ctx.exception))
|
|
|
|
def test_rejects_empty_materials(self):
|
|
bad = _valid_manifest(materials=[])
|
|
with self.assertRaises(ExplorationError):
|
|
parse_exploration_manifest(json.dumps(bad, ensure_ascii=False))
|
|
|
|
def test_rejects_too_many_materials(self):
|
|
many = [{"tool": "read_fine_outline", "args": {"work_id": 12, "target_chapter": 3}, "reason": "x"}
|
|
for _ in range(EXPLORATION_MAX_MATERIALS + 1)]
|
|
with self.assertRaises(ExplorationError):
|
|
parse_exploration_manifest(json.dumps(_valid_manifest(materials=many), ensure_ascii=False))
|
|
|
|
def test_rejects_blank_reason(self):
|
|
bad = _valid_manifest(materials=[{"tool": "read_fine_outline", "args": {"work_id": 12, "target_chapter": 3}, "reason": " "}])
|
|
with self.assertRaises(ExplorationError):
|
|
parse_exploration_manifest(json.dumps(bad, ensure_ascii=False))
|
|
|
|
def test_rejects_non_json(self):
|
|
with self.assertRaises(ExplorationError):
|
|
parse_exploration_manifest("not json")
|
|
|
|
|
|
class ReplayTest(unittest.TestCase):
|
|
|
|
def test_calls_registry_and_preserves_order(self):
|
|
calls = []
|
|
|
|
def fake_execute(tool, args, connect=None):
|
|
calls.append((tool, dict(args)))
|
|
return {"ok": True, "tool": tool}
|
|
|
|
with mock.patch.object(two_phase_writer, "execute_tool", fake_execute):
|
|
materials = replay_manifest_materials(_valid_manifest())
|
|
self.assertEqual(calls, [
|
|
("read_fine_outline", {"work_id": 12, "target_chapter": 3}),
|
|
("read_chapter_text", {"work_id": 12, "chapter_order": 2}),
|
|
])
|
|
self.assertEqual([m["tool"] for m in materials], ["read_fine_outline", "read_chapter_text"])
|
|
self.assertTrue(all(m["result"]["ok"] is True for m in materials))
|
|
self.assertEqual(materials[0]["reason"], "硬事件与约束")
|
|
|
|
def test_fails_closed_on_tool_error(self):
|
|
def boom(tool, args, connect=None):
|
|
raise RuntimeError("db down")
|
|
|
|
with mock.patch.object(two_phase_writer, "execute_tool", boom):
|
|
with self.assertRaises(ExplorationError) as ctx:
|
|
replay_manifest_materials(_valid_manifest())
|
|
self.assertEqual(ctx.exception.code, "EXPLORATION_REPLAY_FAILED")
|
|
|
|
|
|
class GenerationInputTest(unittest.TestCase):
|
|
|
|
def test_uses_explored_materials_not_preassembly(self):
|
|
materials = [{"tool": "read_fine_outline", "args": {"work_id": 12, "target_chapter": 3},
|
|
"reason": "硬事件", "result": {"outline": "x"}}]
|
|
manifest = _valid_manifest(styleNotes=["节奏偏快"], continuityNotes="上一章结尾钩子在尾句")
|
|
with mock.patch.object(two_phase_writer, "build_writer_creative_input", lambda context: _fake_projected()):
|
|
creative = build_generation_creative_input(
|
|
{"runId": "r", "workId": 12, "targetChapter": 3},
|
|
manifest, materials, human_instruction="往污染线推进",
|
|
)
|
|
self.assertEqual(creative["inputMode"], "two-phase-generation-v1")
|
|
self.assertEqual(creative["humanInstruction"], "往污染线推进")
|
|
self.assertEqual(creative["explorationMaterials"], materials)
|
|
self.assertEqual(creative["lengthContract"]["targetChars"], 7000)
|
|
self.assertEqual(creative["styleConstraints"], ["保留第一人称"])
|
|
# 预组装字段不得进入生成输入（写手事实只来自探索资料）
|
|
self.assertNotIn("fineOutline", creative)
|
|
self.assertEqual(creative["explorationNotes"]["styleNotes"], ["节奏偏快"])
|
|
self.assertIn("仅供参考", creative["explorationNotes"]["note"])
|
|
|
|
def test_default_instruction(self):
|
|
with mock.patch.object(two_phase_writer, "build_writer_creative_input", lambda context: _fake_projected()):
|
|
creative = build_generation_creative_input({}, _valid_manifest(), [], human_instruction="")
|
|
self.assertEqual(creative["humanInstruction"], "基于探索资料续写本章完整正文。")
|
|
self.assertNotIn("explorationNotes", creative)
|
|
|
|
|
|
class OrchestrationTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp_path = pathlib.Path(self._tmp.name)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _run(self, fake_dispatch, fake_generation=None, fake_execute=None):
|
|
context = {"runId": "run-test", "workId": 12, "targetChapter": 3}
|
|
with mock.patch.object(two_phase_writer, "run_dispatch", fake_dispatch), \
|
|
mock.patch.object(two_phase_writer, "execute_tool",
|
|
fake_execute or (lambda tool, args, connect=None: {"ok": True})), \
|
|
mock.patch.object(two_phase_writer, "build_writer_creative_input",
|
|
lambda context: _fake_projected()):
|
|
if fake_generation is not None:
|
|
with mock.patch.object(two_phase_writer, "run_writer_via_dispatch", fake_generation):
|
|
return run_two_phase_writer(
|
|
context, candidate_version=9, repo_root=self.tmp_path,
|
|
provider="catproxy-anthropic", model="claude-opus-5", thinking="high",
|
|
human_instruction="", spec_dir=self.tmp_path,
|
|
)
|
|
return run_two_phase_writer(
|
|
context, candidate_version=9, repo_root=self.tmp_path,
|
|
provider="catproxy-anthropic", model="claude-opus-5", thinking="high",
|
|
human_instruction="", spec_dir=self.tmp_path,
|
|
)
|
|
|
|
def test_two_phase_orchestration(self):
|
|
manifest = _valid_manifest(styleNotes=["短句"])
|
|
dispatched = []
|
|
|
|
def fake_dispatch(spec_file, **kwargs):
|
|
dispatched.append({"spec": json.loads(pathlib.Path(spec_file).read_text(encoding="utf-8")), "kwargs": kwargs})
|
|
run_dir = self.tmp_path / kwargs["run_id"]
|
|
run_dir.mkdir(parents=True)
|
|
(run_dir / "output.json").write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
|
return {"status": "completed", "runDir": str(run_dir)}, 0
|
|
|
|
captured = {}
|
|
|
|
class _GenReceipt:
|
|
dispatch_run_id = "run-test-writer-v9"
|
|
|
|
def fake_generation(context, **kwargs):
|
|
captured.update(kwargs)
|
|
return {"candidateBody": "正文"}, _GenReceipt(), (7, 8)
|
|
|
|
envelope, receipt, raw_ref, exploration = self._run(fake_dispatch, fake_generation)
|
|
|
|
# 探索阶段：独立运行与会话、只读工具开启
|
|
self.assertEqual(dispatched[0]["kwargs"]["run_id"], "run-test-explore-v9")
|
|
self.assertEqual(dispatched[0]["kwargs"]["session_id"], "writer-explore-work12-ch3")
|
|
self.assertTrue(dispatched[0]["kwargs"]["enable_read_tools"])
|
|
self.assertEqual(dispatched[0]["spec"]["outputSchemaId"], "writer-exploration-manifest-v1")
|
|
|
|
# 生成阶段：无工具、输入为探索整理
|
|
self.assertFalse(captured["enable_read_tools"])
|
|
self.assertEqual(captured["session_label"], "writer")
|
|
self.assertEqual(captured["creative_input"]["inputMode"], "two-phase-generation-v1")
|
|
self.assertEqual(captured["creative_input"]["explorationMaterials"][0]["tool"], "read_fine_outline")
|
|
self.assertIn("不要再调用工具", captured["task_prompt"])
|
|
|
|
self.assertEqual(envelope["candidateBody"], "正文")
|
|
self.assertEqual(raw_ref, (7, 8))
|
|
self.assertEqual(exploration["explorationRunId"], "run-test-explore-v9")
|
|
self.assertEqual(exploration["materialCount"], 2)
|
|
self.assertEqual(exploration["generationRunId"], "run-test-writer-v9")
|
|
self.assertEqual(exploration["tools"], {"read_fine_outline": 1, "read_chapter_text": 1})
|
|
|
|
def test_fails_closed_when_exploration_dispatch_fails(self):
|
|
def fake_dispatch(spec_file, **kwargs):
|
|
return {"status": "failed", "errorCode": "TIMEOUT"}, 1
|
|
|
|
with self.assertRaises(ExplorationError) as ctx:
|
|
self._run(fake_dispatch)
|
|
self.assertEqual(ctx.exception.code, "TIMEOUT")
|
|
|
|
def test_fails_closed_when_manifest_invalid(self):
|
|
def fake_dispatch(spec_file, **kwargs):
|
|
run_dir = self.tmp_path / kwargs["run_id"]
|
|
run_dir.mkdir(parents=True)
|
|
(run_dir / "output.json").write_text('{"schemaVersion": "wrong"}', encoding="utf-8")
|
|
return {"status": "completed", "runDir": str(run_dir)}, 0
|
|
|
|
with self.assertRaises(ExplorationError) as ctx:
|
|
self._run(fake_dispatch)
|
|
self.assertEqual(ctx.exception.code, "EXPLORATION_MANIFEST_INVALID")
|
|
|
|
|
|
class ExplorationSpecTest(unittest.TestCase):
|
|
|
|
def test_spec_shape(self):
|
|
spec = build_exploration_spec(
|
|
{"workId": 12}, target_chapter=3, human_instruction="推进污染线", candidate_version=2,
|
|
)
|
|
self.assertEqual(spec["role"], "writer")
|
|
self.assertEqual(spec["input"], {"workId": 12, "targetChapter": 3, "candidateVersion": 2, "humanInstruction": "推进污染线"})
|
|
self.assertIn("不要写正文", spec["taskPrompt"])
|
|
self.assertTrue(spec["toolAllowlist"])
|
|
self.assertEqual(spec["maxDurationSeconds"], two_phase_writer.EXPLORATION_MAX_DURATION_SECONDS)
|
|
|
|
|
|
class BridgeExtensionTest(unittest.TestCase):
|
|
|
|
def test_spec_overrides_and_tool_switch(self):
|
|
context = {"workId": 12, "runId": "r", "targetChapter": 3}
|
|
with mock.patch.object(bridge, "build_writer_creative_input", lambda context: {"fineOutline": {}}):
|
|
spec = bridge.build_writer_dispatch_spec(
|
|
context, target_chapter=3, human_instruction="", candidate_version=1,
|
|
task_prompt="自定义任务", creative_input={"inputMode": "two-phase-generation-v1"},
|
|
enable_read_tools=False,
|
|
)
|
|
self.assertEqual(spec["taskPrompt"], "自定义任务")
|
|
self.assertEqual(spec["input"]["creativeInput"], {"inputMode": "two-phase-generation-v1"})
|
|
self.assertEqual(spec["toolAllowlist"], [])
|
|
|
|
with mock.patch.object(bridge, "build_writer_creative_input", lambda context: {"fineOutline": {}}):
|
|
default_spec = bridge.build_writer_dispatch_spec(
|
|
context, target_chapter=3, human_instruction="", candidate_version=1,
|
|
)
|
|
self.assertTrue(default_spec["toolAllowlist"]) # 缺省保持单阶段行为：带工具
|
|
self.assertIn("先用授权只读工具", default_spec["taskPrompt"])
|
|
|
|
def test_session_labels(self):
|
|
sid, _ = bridge.writer_session_paths(12, 3)
|
|
explore_sid, _ = bridge.writer_session_paths(12, 3, label="writer-explore")
|
|
self.assertEqual(sid, "writer-work12-ch3")
|
|
self.assertEqual(explore_sid, "writer-explore-work12-ch3")
|
|
self.assertNotEqual(explore_sid, sid)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|