37d508ffb0
step2_write_chapter.py 在 sys.path 加入 prevent-ai-flavor / diagnose-ai-flavor，新增「1.5) 人感前置预防」环节——生成前调用 build_prevention_contract（负约束+声音账）冻结进 WriterContext，并把 diagnosis 列为 shadow 候选的伴随检测。step2a_retrofit_ch1_context.py 相应补回退补写。 脚本只做拼装与持久化触发，五技能合同与 hard gate 由各自 skill 把守。 人感体系资产层仍由 humanization/ 维护（example 自治）；本脚本只消费。
547 lines
30 KiB
Python
547 lines
30 KiB
Python
#!/usr/bin/env python3
|
||
"""真写一章 · 阶段二：走完整生产链写下一章（默认写目标作品的下一章）。
|
||
|
||
生产链（meta/chains continuation 登记的保护节点序列落地）：
|
||
读已确认细纲 + 前章正文基线
|
||
→ build_retrieval_plan + retrieve_writer_sources（生产仓储；新书无卡诚实返空）
|
||
→ assemble_context 冻结 WriterContext v1
|
||
→ run_writer_pipeline（持久 CAS 状态链 + 机械门 + 语义 detector，有限补证/重写合同）
|
||
writer 真调经 run_writer_with_receipt：runtime 自动把模型输入/输出原文与调用明细原子落库；
|
||
篇幅越界在 writer 适配层自动重抽（最多三遍）；语义 detector 走冻结 detector profile 真调；
|
||
生产链暂未建补证重组装，证据缺口失败关闭（拒绝候选，留痕待人工处理）
|
||
→ persist_writer_execution（冻结 + 运行注册 + 候选[含语义状态] + 回执 + 机械/语义质量证据一次落库）
|
||
→ accept_preflight（check_writer_acceptance 纯函数 + acceptance_state 实时重读）
|
||
→ confirm.accept 接受为正式正文（DB 级兜底：state=passed 且 semantic_status=passed 才可写）
|
||
|
||
用法：.venv/bin/python docs/write-chapter/step2_write_chapter.py [目标章号] [--dry-run]
|
||
缺省写下一章（库内最大章序 +1）。前置：该章已建且有 confirmed 细纲（镜像
|
||
step2_setup_chapter2.py 建章 + 落细纲），且门锚合同 GATE_ANCHORS 已登记该章。
|
||
--dry-run：全链试跑——writer 与语义 detector 真调、候选与证据真实落 Shadow、接受事务试跑后
|
||
整体回滚，正典正文不动；候选留库待用户三决策。
|
||
"""
|
||
import hashlib
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import Any, Mapping
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
AGENT_ROOT = SCRIPT_DIR.parents[1]
|
||
SKILLS = AGENT_ROOT / ".claude" / "skills"
|
||
for sub in (
|
||
"assemble-context/scripts",
|
||
"write-next-chapter/scripts",
|
||
"execute-claude-task/scripts",
|
||
"record-run-evidence/scripts",
|
||
"access-database/scripts",
|
||
"check-content-consistency/scripts",
|
||
"decide-candidate/scripts",
|
||
"prevent-ai-flavor/scripts",
|
||
"diagnose-ai-flavor/scripts",
|
||
):
|
||
p = str(SKILLS / sub)
|
||
if p not in sys.path:
|
||
sys.path.insert(0, p)
|
||
|
||
from db import connect, DSN # noqa: E402
|
||
from assemble_writer_context import assemble_context # noqa: E402
|
||
from prevent_ai_flavor import ( # noqa: E402
|
||
PreventionContractError, build_prevention_contract, persist_prevention, render_writer_constraints,
|
||
)
|
||
from diagnose_ai_flavor import run_diagnosis, persist_diagnosis # noqa: E402
|
||
from retrieve_writer_sources import ( # noqa: E402
|
||
ProductionCardIndexRepository, FrozenProseRepository, RetrievalError,
|
||
build_retrieval_plan, retrieve_writer_sources, load_confirmed_fine_outline,
|
||
load_confirmed_pattern_bindings, load_confirmed_style,
|
||
)
|
||
from run_writer import ( # noqa: E402
|
||
WriterAdapterError, build_writer_execution_profile,
|
||
calculate_dynamic_output_contract, run_writer_with_receipt,
|
||
)
|
||
from run_writer_pipeline import PipelineError, run_writer_pipeline # noqa: E402
|
||
from candidate_cas import PostgresCasStateStore # noqa: E402
|
||
from run_writer_semantic_detector import ( # noqa: E402
|
||
SEMANTIC_DETECTOR_REPORT_JSON_SCHEMA, build_safe_semantic_diagnostic,
|
||
build_semantic_input_v3, run_writer_semantic_detector,
|
||
)
|
||
from claude_runtime import ExecutionProfile, run_claude, sha256_json # noqa: E402
|
||
from persist_llm_call import persist_call as persist_llm_event # noqa: E402
|
||
from persist_writer_run import persist_writer_execution # noqa: E402
|
||
from run_registry import finish_run, start_run # noqa: E402
|
||
from check_writer_acceptance import AcceptanceError, check_writer_acceptance # noqa: E402
|
||
from acceptance_state import LiveStateError, build_live_acceptance_state # noqa: E402
|
||
from write_canonical import accept # noqa: E402
|
||
|
||
WORK_ID = 12
|
||
# 2026-08-14 运行绑定再登记：本机 Claude CLI 自动升级 2.1.211→2.1.231，
|
||
# 仅同步可执行文件版本与哈希（模型/预算/提示词/schema 一律未动）。
|
||
CLAUDE_BIN = "/Users/qingse/.nvm/versions/node/v24.15.0/bin/claude"
|
||
CLAUDE_SHA256 = "ba790279cab6ef77b713864d4bf5f764fcea87d3a3eb7591a41f741e45212b5c"
|
||
CLAUDE_VERSION = "2.1.231"
|
||
RESOLVED_MODEL_ID = "claude-opus-4-8[1m]"
|
||
MAX_ATTEMPTS = 3 # 篇幅越界重抽上限（编排层响应，对齐 replay 自纠环语义）
|
||
|
||
GATE_A_CONFIG = json.loads((SKILLS / "evaluate-frozen-replay" / "configs" /
|
||
"writer-gate-a-deep-space-v1.json").read_text(encoding="utf-8"))
|
||
GATE_A_WRITER = GATE_A_CONFIG["executionProfiles"]["writer"]
|
||
GATE_A_DETECTOR = GATE_A_CONFIG["executionProfiles"]["semantic_detector"]
|
||
SYSTEM_PROMPT = GATE_A_WRITER["systemPrompt"]
|
||
SYSTEM_PROMPT_ID = GATE_A_WRITER["systemPromptId"]
|
||
SYSTEM_PROMPT_SHA256 = "sha256:" + hashlib.sha256(SYSTEM_PROMPT.encode("utf-8")).hexdigest()
|
||
assert SYSTEM_PROMPT_SHA256 == GATE_A_WRITER["systemPromptSha256"], "冻结提示词漂移，停止"
|
||
|
||
ARTIFACTS = SCRIPT_DIR / "artifacts"
|
||
|
||
# 门锚合同按章登记：锚点是章级创作判断，any-hit 子串匹配。新章必须先登记再跑。
|
||
GATE_ANCHORS = {
|
||
2: {
|
||
"requiredEvents": [
|
||
{"requirementId": "event-1-isolation",
|
||
"anchors": ["隔离", "收押", "禁闭", "关押", "封锁"]},
|
||
{"requirementId": "event-2-interrogation",
|
||
"anchors": ["审讯", "审问", "询问", "盘问", "讯问"]},
|
||
{"requirementId": "event-3-conceal",
|
||
"anchors": ["隐瞒", "没有告诉", "没说", "没有说", "咽了回去", "沉默", "闭上嘴"]},
|
||
{"requirementId": "event-4-hunger",
|
||
"anchors": ["饥饿", "渴望", "吞噬", "进食", "吃", "贪"]},
|
||
],
|
||
"requiredCharacters": ["林深", "何岚"],
|
||
"foreshadowingActions": [
|
||
{"requirementId": "foreshadow-upgrade",
|
||
"anchors": ["异种核心", "融合", "升级"]},
|
||
],
|
||
"chapterEndHook": {
|
||
"requirementId": "hook-ch2",
|
||
"anchors": ["调令", "实战", "出击", "部署", "任务", "出征", "离不开", "不愿离开"],
|
||
"maxDistanceFromEnd": 900,
|
||
},
|
||
},
|
||
3: {
|
||
# 接第2章结尾硬钩子（茧撕开舱门出击、要吃掉更强核心）。锚点 any-hit 子串匹配；
|
||
# requiredCharacters 只硬约束主角（避免过度约束触发 costly 重抽），其余靠事件锚点。
|
||
"requiredEvents": [
|
||
{"requirementId": "event-1-sortie",
|
||
"anchors": ["出击", "实战", "战斗", "交火", "搏杀", "拦截", "扑向", "战场"]},
|
||
{"requirementId": "event-2-devour",
|
||
"anchors": ["吞噬", "吞食", "吃掉", "进食", "撕碎", "吸收", "吞下", "吞"]},
|
||
{"requirementId": "event-3-upgrade",
|
||
"anchors": ["升级", "蜕变", "进化", "变强", "增强", "新的力量", "蜕变"]},
|
||
{"requirementId": "event-4-pollution",
|
||
"anchors": ["黑纹", "污染", "扩散", "蔓延", "加深", "恶化"]},
|
||
],
|
||
"requiredCharacters": ["林深"],
|
||
"foreshadowingActions": [
|
||
{"requirementId": "foreshadow-voice-merge",
|
||
"anchors": ["分不清", "像他自己", "脑内的声音", "低语", "渴望", "哪个念头", "另一个"]},
|
||
],
|
||
"chapterEndHook": {
|
||
"requirementId": "hook-ch3",
|
||
"anchors": ["深渊", "更深", "回应", "召唤", "更大", "下一", "不止", "饥饿", "注视", "凝视"],
|
||
"maxDistanceFromEnd": 900,
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
def make_logging_runner(log_path: Path):
|
||
"""透传 subprocess.run，只把 sandbox 调用现场原样留档（审计用，不影响落库）。"""
|
||
|
||
def runner(command, *, input, text, capture_output, timeout, check, cwd, env,
|
||
start_new_session):
|
||
record = {"command": list(command), "envKeys": sorted(env.keys()), "cwd": cwd,
|
||
"inputChars": len(input), "startedAt": datetime.now().isoformat()}
|
||
completed = subprocess.run(list(command), input=input, text=text,
|
||
capture_output=capture_output, timeout=timeout,
|
||
check=check, cwd=cwd, env=dict(env),
|
||
start_new_session=start_new_session)
|
||
record.update({"returncode": completed.returncode,
|
||
"stdout": completed.stdout or "", "stderr": completed.stderr or ""})
|
||
log_path.write_text(json.dumps(record, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
return completed
|
||
|
||
return runner
|
||
|
||
|
||
def make_sequential_logging_runner(directory: Path, prefix: str):
|
||
"""多次模型调用（重抽 / detector 纠错环）逐次留档，不互相覆盖。"""
|
||
|
||
counter = {"n": 0}
|
||
|
||
def runner(command, *, input, text, capture_output, timeout, check, cwd, env,
|
||
start_new_session):
|
||
counter["n"] += 1
|
||
return make_logging_runner(directory / f"{prefix}-{counter['n']}.json")(
|
||
command, input=input, text=text, capture_output=capture_output,
|
||
timeout=timeout, check=check, cwd=cwd, env=env,
|
||
start_new_session=start_new_session)
|
||
|
||
return runner
|
||
|
||
|
||
def build_semantic_detector_profile() -> ExecutionProfile:
|
||
"""从冻结配置构造语义 detector 的 ExecutionProfile（哈希在构造时自检，漂移即停）。"""
|
||
|
||
cfg = GATE_A_DETECTOR
|
||
assert sha256_json(SEMANTIC_DETECTOR_REPORT_JSON_SCHEMA) == cfg["jsonSchemaSha256"], \
|
||
"语义 detector schema 与冻结配置漂移，停止"
|
||
return ExecutionProfile(
|
||
profile_version=cfg["profileVersion"],
|
||
adapter_role="semantic_detector",
|
||
claude_executable_path=cfg["claudeExecutablePath"],
|
||
claude_executable_sha256=cfg["claudeExecutableSha256"],
|
||
claude_cli_version=cfg["claudeCliVersion"],
|
||
model_alias=cfg["modelAlias"],
|
||
resolved_model_id=cfg["resolvedModelId"],
|
||
effort=cfg["effort"],
|
||
max_budget_usd_per_call=Decimal(cfg["maxBudgetUsdPerCall"]),
|
||
timeout_seconds=cfg["timeoutSeconds"],
|
||
max_context_chars=cfg["maxContextChars"],
|
||
json_schema_id=cfg["jsonSchemaId"],
|
||
json_schema=SEMANTIC_DETECTOR_REPORT_JSON_SCHEMA,
|
||
json_schema_sha256=cfg["jsonSchemaSha256"],
|
||
system_prompt_id=cfg["systemPromptId"],
|
||
system_prompt=cfg["systemPrompt"],
|
||
system_prompt_sha256=cfg["systemPromptSha256"],
|
||
normal_terminal_reasons=tuple(cfg["normalTerminalReasons"]),
|
||
)
|
||
|
||
|
||
class ProductionSemanticRunner:
|
||
"""语义 detector 生产 runner：与 ClaudeRuntimeModelRunner 同合同，但调用随 run_id 落库。"""
|
||
|
||
def __init__(self, profile: ExecutionProfile, *, run_id: str, log_runner) -> None:
|
||
self.profile = profile
|
||
self.run_id = run_id
|
||
self.log_runner = log_runner
|
||
|
||
def run(self, *, adapter_role: str, model_input: Mapping[str, Any],
|
||
output_schema: Mapping[str, Any]) -> Mapping[str, Any]:
|
||
if self.profile.adapter_role != adapter_role or self.profile.json_schema != output_schema:
|
||
raise PipelineError("SEMANTIC_DETECTOR_FAILED", "ExecutionProfile 与语义 detector 适配不一致")
|
||
|
||
def persist_detector_event(event):
|
||
# runtime 默认把非 writer 调用标为 evaluation；生产链语义审查改标 production_detection。
|
||
event = dict(event)
|
||
event["purpose"] = "production_detection"
|
||
return persist_llm_event(event)
|
||
|
||
result = run_claude(
|
||
self.profile, model_input, runner=self.log_runner,
|
||
run_id=self.run_id, caller="semantic_detector",
|
||
persist_call=persist_detector_event)
|
||
receipt = result.receipt
|
||
receipt_dict = receipt.as_dict() if hasattr(receipt, "as_dict") else receipt
|
||
return {"structuredOutput": dict(result.structured_output),
|
||
"modelReceiptSha256": sha256_json(receipt_dict)}
|
||
|
||
|
||
def resolve_target_chapter(requested: int | None) -> int:
|
||
with connect(readonly=True) as conn:
|
||
if requested is not None:
|
||
return requested
|
||
row = conn.execute(
|
||
"SELECT COALESCE(MAX(order_no),0)+1 FROM muse_content_chapter "
|
||
"WHERE work_id=%s AND deleted=false", (WORK_ID,)).fetchone()
|
||
return int(row[0])
|
||
|
||
|
||
def _dump(path: Path, value: Any) -> None:
|
||
path.write_text(json.dumps(value, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
|
||
|
||
def main():
|
||
args = [arg for arg in sys.argv[1:] if not arg.startswith("--")]
|
||
dry_run = "--dry-run" in sys.argv[1:]
|
||
target = resolve_target_chapter(int(args[0]) if args else None)
|
||
as_of = target - 1
|
||
run_id = f"run-prod-work12-ch{target}-{uuid.uuid4().hex[:8]}"
|
||
generated_at = datetime.now(timezone.utc).isoformat()
|
||
ARTIFACTS.mkdir(exist_ok=True)
|
||
if target not in GATE_ANCHORS:
|
||
raise SystemExit(f"第{target}章门锚合同未登记（GATE_ANCHORS），先登记锚点再跑。")
|
||
|
||
# 1) 已确认细纲（read-context 统一消费点）+ 前章全文基线（asOf 起连续四章；不足四章从第1章起）
|
||
with connect(readonly=True) as conn:
|
||
try:
|
||
fine_outline = load_confirmed_fine_outline(conn, work_id=WORK_ID, target_chapter=target)
|
||
except RetrievalError as exc:
|
||
raise SystemExit(f"{exc}（镜像 step2_setup_chapter2.py 建章并落细纲）")
|
||
first = max(1, as_of - 3)
|
||
recent_rows = conn.execute(
|
||
"SELECT c.order_no, b.id, b.revision, b.content_text FROM muse_content_chapter c "
|
||
"JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=false "
|
||
"WHERE c.work_id=%s AND c.deleted=false AND c.order_no BETWEEN %s AND %s "
|
||
"ORDER BY c.order_no", (WORK_ID, first, as_of)).fetchall()
|
||
style_constraints = load_confirmed_style(conn, work_id=WORK_ID)
|
||
pattern_references = load_confirmed_pattern_bindings(conn, work_id=WORK_ID)
|
||
recent_chapters = [{
|
||
"chapter": order_no,
|
||
"sourceRef": {"sourceId": f"content-block:{block_id}",
|
||
"sourceVersion": f"rev{revision}",
|
||
"blockId": int(block_id), "chapter": order_no,
|
||
"startCodePoint": 0, "endCodePoint": len(body)},
|
||
"text": body,
|
||
} for order_no, block_id, revision, body in recent_rows]
|
||
expected = list(range(first, as_of + 1))
|
||
got = [item["chapter"] for item in recent_chapters]
|
||
if got != expected:
|
||
raise SystemExit(f"连续前章基线缺章：期望 {expected}，实际 {got}。")
|
||
print(f"目标第{target}章（asOf={as_of}）；细纲已读，基线 {got}，"
|
||
f"基线总字数 {sum(len(item['text']) for item in recent_chapters)}")
|
||
|
||
# 1.5) 人感前置预防：规则/声音账先形成合同，再冻结进 WriterContext。
|
||
try:
|
||
humanization_contract = build_prevention_contract(
|
||
f"work:{WORK_ID}", load_database=True
|
||
)
|
||
humanization_contract["writer_constraints"] = render_writer_constraints(humanization_contract)
|
||
prevention_receipt = persist_prevention(humanization_contract)
|
||
except PreventionContractError as exc:
|
||
raise SystemExit(f"人感前置预防合同失败：{exc}") from exc
|
||
print(f"人感前置预防：约束 {len(humanization_contract['writer_constraints'])} 条，"
|
||
f"规则库 {humanization_contract['built_from']['rule_library_version']}，"
|
||
f"run_id={prevention_receipt['run_id']}")
|
||
|
||
# 2) 检索计划 + 执行（生产仓储；新书无卡诚实返空）
|
||
token_budget = {"maxContextChars": 200000}
|
||
plan = build_retrieval_plan(
|
||
run_id=run_id, work_id=WORK_ID, target_chapter=target, as_of=as_of,
|
||
fine_outline=fine_outline, card_index_version="knowledge-index-v1",
|
||
prose_index_version="content-block-v1", token_budget=token_budget)
|
||
retrieval = retrieve_writer_sources(
|
||
plan=plan, card_repository=ProductionCardIndexRepository(),
|
||
prose_repository=FrozenProseRepository(dsn=DSN, tenant_id=0))
|
||
_dump(ARTIFACTS / f"{run_id}-retrieval.json",
|
||
{"plan": plan, "resultCounts": {k: len(v) for k, v in retrieval.items()
|
||
if isinstance(v, list)}})
|
||
print(f"检索: 卡={len(retrieval['cards'])} 事实={len(retrieval['factEvidence'])} "
|
||
f"原文={len(retrieval['proseEvidence'])}")
|
||
|
||
# 3) 组装冻结 WriterContext v1
|
||
output_contract = calculate_dynamic_output_contract(
|
||
fine_outline=fine_outline,
|
||
recent_chapter_bodies=[item["text"] for item in recent_chapters])
|
||
narrative_state = {
|
||
"time": f"第{as_of}章结束后",
|
||
"location": "承接上一章结尾的场景",
|
||
"characterPositions": {},
|
||
"immediateSituation": "按细纲 chapterGoal 展开（上一章结尾状态见基线正文）。",
|
||
}
|
||
authorization_snapshot = {"snapshotId": "auth-work12-production-v1",
|
||
"allowedPurpose": "production_generation",
|
||
"verifiedAt": generated_at, "sourceVersion": "v1"}
|
||
assembled = assemble_context(
|
||
run_id=run_id, attempt=1, mode="production", purpose="production",
|
||
quality_policy_version="writer-production-v1", work_id=WORK_ID,
|
||
target_chapter=target, as_of=as_of, source_version="outline@v1",
|
||
authorization_snapshot=authorization_snapshot, source_status="active",
|
||
retrieval_plan=plan, retrieval_result=retrieval, fine_outline=fine_outline,
|
||
narrative_state=narrative_state, recent_chapters=recent_chapters,
|
||
output_contract=output_contract, token_budget=token_budget,
|
||
pattern_references=pattern_references, style_constraints=style_constraints,
|
||
humanization_contract=humanization_contract, generated_at=generated_at,
|
||
evidence_strategy="production_dual_evidence")
|
||
writer_context = assembled["context"]
|
||
(ARTIFACTS / f"{run_id}-writer-context.json").write_text(
|
||
assembled["contextJson"], encoding="utf-8")
|
||
print(f"上下文冻结: contextSha256={writer_context['contextSnapshot']['contextSha256'][:24]}... "
|
||
f"篇幅合同 {output_contract['minChars']}-{output_contract['maxChars']}"
|
||
f"（目标 {output_contract['targetChars']}），范式绑定 {len(pattern_references)} 张，"
|
||
f"文风约束 {len(style_constraints)} 条")
|
||
|
||
# 4) 生产 pipeline：持久 CAS + writer 真调 + 机械门 + 语义 detector（先审后入）
|
||
writer_profile = build_writer_execution_profile(
|
||
claude_executable_path=CLAUDE_BIN, claude_executable_sha256=CLAUDE_SHA256,
|
||
claude_cli_version=CLAUDE_VERSION, resolved_model_id=RESOLVED_MODEL_ID,
|
||
effort="high", max_budget_usd_per_call=Decimal("5.0"), timeout_seconds=1200,
|
||
max_context_chars=token_budget["maxContextChars"], system_prompt=SYSTEM_PROMPT)
|
||
detector_profile = build_semantic_detector_profile()
|
||
semantic_runner = ProductionSemanticRunner(
|
||
detector_profile, run_id=run_id,
|
||
log_runner=make_sequential_logging_runner(ARTIFACTS, f"{run_id}-semantic-call"))
|
||
state_store = PostgresCasStateStore(
|
||
work_id=WORK_ID, target_chapter=target, creator="continuation")
|
||
receipts_by_version: dict[int, Any] = {}
|
||
candidates_by_version: dict[int, dict] = {}
|
||
contexts_by_attempt: dict[int, dict] = {}
|
||
|
||
def production_writer(current_context: Mapping[str, Any], candidate_version: int):
|
||
"""writer 适配：篇幅越界自动重抽（≤3 遍），成功回执按候选版本留证。"""
|
||
|
||
contexts_by_attempt[current_context["attempt"]] = dict(current_context)
|
||
for retry in range(1, MAX_ATTEMPTS + 1):
|
||
try:
|
||
candidate, receipt = run_writer_with_receipt(
|
||
current_context, profile=writer_profile,
|
||
candidate_version=candidate_version,
|
||
runner=make_sequential_logging_runner(
|
||
ARTIFACTS, f"{run_id}-sandbox-call-v{candidate_version}"),
|
||
persist_call=persist_llm_event)
|
||
receipts_by_version[candidate_version] = receipt
|
||
candidates_by_version[candidate_version] = candidate
|
||
# 人感技能 3：每个候选先做只读诊断并自动落质量账；不在这里改正文。
|
||
try:
|
||
deai_artifact = run_diagnosis(
|
||
candidate["candidateBody"], work_ref=f"work:{WORK_ID}",
|
||
chapter_ref=f"chapter:{target}", mode="Audit",
|
||
)
|
||
_dump(ARTIFACTS / f"{run_id}-ai-flavor-diagnosis-v{candidate_version}.json", deai_artifact)
|
||
persist_diagnosis(deai_artifact, text=candidate["candidateBody"])
|
||
except Exception as exc:
|
||
raise PipelineError(
|
||
"AI_FLAVOR_DIAGNOSIS_FAILED", "候选 AI 味诊断或落库失败",
|
||
details={"errorType": type(exc).__name__, "message": str(exc)},
|
||
) from exc
|
||
print(f"AI 味诊断(v{candidate_version})：发现 {len(deai_artifact['findings'])} 条")
|
||
return candidate
|
||
except WriterAdapterError as exc:
|
||
if exc.code != "candidate_length_out_of_range" or retry == MAX_ATTEMPTS:
|
||
raise PipelineError(exc.code, f"writer 真调失败: {exc}",
|
||
details=exc.details) from exc
|
||
print(f" 第{retry}遍越界（{exc.details.get('actualHanChars')} 字），重抽……")
|
||
|
||
def production_semantic_detector(current_context, candidate, mechanical_report):
|
||
"""语义 detector 适配：构造冻结输入、真调模型、留档输入输出。"""
|
||
|
||
version = candidate["candidateVersion"]
|
||
detector_input = build_semantic_input_v3(
|
||
run_id=run_id, sample_id=f"writer-ch{target}", opaque_arm_id="production",
|
||
writer_context=current_context, candidate=candidate)
|
||
_dump(ARTIFACTS / f"{run_id}-semantic-input-v{version}.json", detector_input)
|
||
outcome = run_writer_semantic_detector(detector_input, model_runner=semantic_runner)
|
||
_dump(ARTIFACTS / f"{run_id}-semantic-output-v{version}.json", outcome)
|
||
if outcome.get("ok") is not True or not isinstance(outcome.get("report"), Mapping):
|
||
diagnostic = build_safe_semantic_diagnostic(outcome)
|
||
raise PipelineError(
|
||
"SEMANTIC_DETECTOR_FAILED",
|
||
f"语义 detector 未产生有效报告: {diagnostic['primaryCode']}",
|
||
details={"safeDiagnostic": diagnostic})
|
||
print(f"语义 detector(v{version}): status={outcome['status']} "
|
||
f"调用={outcome['attemptCount']}次 纠错={outcome['correctionCount']}次")
|
||
return outcome["report"]
|
||
|
||
def production_evidence_provider(_context, gaps, _attempt):
|
||
"""生产链补证重组装未建：证据缺口失败关闭，候选拒绝留痕，交人工处理。"""
|
||
|
||
raise PipelineError(
|
||
"PRODUCTION_EVIDENCE_SUPPLEMENT_UNBUILT",
|
||
f"语义审查发现 {len(gaps)} 个证据缺口，生产链暂不支持补证重组装，候选拒绝待人工处理")
|
||
|
||
# 候选版本接续：候选表对 (作品,章,candidate_version) 唯一，重跑同章必须从已有最大版本+1 起，
|
||
# 否则与上一轮留库的被拒候选撞版本。
|
||
with connect(readonly=True) as conn:
|
||
max_version_row = conn.execute(
|
||
"SELECT COALESCE(MAX(CASE WHEN candidate_version ~ '^[0-9]+$' "
|
||
"THEN candidate_version::integer END),0) FROM example_candidate "
|
||
"WHERE tenant_id=0 AND work_id=%s AND target_chapter=%s AND deleted=false",
|
||
(WORK_ID, target)).fetchone()
|
||
initial_candidate_version = int(max_version_row[0]) + 1
|
||
|
||
start_run(run_id=run_id, work_id=WORK_ID, target_chapter=target,
|
||
trigger_detail={"stage": "writer-production-pipeline",
|
||
"contextSha256": writer_context["contextSnapshot"]["contextSha256"]},
|
||
creator="continuation")
|
||
try:
|
||
pipeline_result = run_writer_pipeline(
|
||
context=writer_context,
|
||
requirements=GATE_ANCHORS[target],
|
||
writer=production_writer,
|
||
evidence_provider=production_evidence_provider,
|
||
semantic_detector=production_semantic_detector,
|
||
state_store=state_store,
|
||
result_path=ARTIFACTS / f"{run_id}-pipeline-result.json",
|
||
initial_candidate_version=initial_candidate_version,
|
||
)
|
||
except PipelineError as exc:
|
||
# 被拒候选留痕：凡跑过机械门的版本都落 Shadow（state=rejected + 机械/语义证据）
|
||
trace = (exc.result or {}).get("trace") or []
|
||
audit_entry = next((entry for entry in reversed(trace)
|
||
if isinstance(entry.get("mechanicalReport"), Mapping)), None)
|
||
if audit_entry is not None:
|
||
version = audit_entry.get("candidateVersion")
|
||
failed_candidate = candidates_by_version.get(version)
|
||
failed_receipt = receipts_by_version.get(version)
|
||
if failed_candidate is not None and failed_receipt is not None:
|
||
try:
|
||
persisted = persist_writer_execution(
|
||
contexts_by_attempt.get(failed_candidate.get("attempt"), writer_context),
|
||
failed_candidate, failed_receipt, audit_entry["mechanicalReport"],
|
||
semantic_report=audit_entry.get("semanticReport"),
|
||
assemble_result=assembled)
|
||
print(f"[被拒候选留库] candidate_id={persisted['candidate_id']} "
|
||
f"state={persisted['state']} semantic={persisted.get('semantic_status')}")
|
||
except Exception as persist_exc: # 留痕失败不掩盖原始失败码
|
||
print(f"[警告] 被拒候选留库失败: {persist_exc}", file=sys.stderr)
|
||
finish_run(run_id, "failed", creator="continuation",
|
||
trigger_detail={"stage": "writer-production-pipeline", "failureCode": exc.code})
|
||
print(f"[停止] 生产 pipeline 未通过: code={exc.code}；{exc}")
|
||
print(f"复核 artifacts/{run_id}-pipeline-result.json 后决定下一步。")
|
||
print(f"\nRUN_ID={run_id}")
|
||
raise SystemExit(1)
|
||
except Exception as exc:
|
||
# 非 PipelineError（库连接断、适配层异常等）也要收口运行态，不留 running 悬挂
|
||
try:
|
||
finish_run(run_id, "failed", creator="continuation",
|
||
trigger_detail={"stage": "writer-production-pipeline",
|
||
"error": type(exc).__name__})
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
# 5) pipeline 通过：机械门 + 语义 detector 双证据落库（候选 semantic_status=passed）
|
||
candidate = pipeline_result["candidateArtifact"]
|
||
final_context = contexts_by_attempt.get(pipeline_result["attempt"], writer_context)
|
||
final_trace = pipeline_result["trace"][-1]
|
||
receipt = receipts_by_version[pipeline_result["candidateVersion"]]
|
||
persisted = persist_writer_execution(
|
||
final_context, candidate, receipt, final_trace["mechanicalReport"],
|
||
semantic_report=final_trace.get("semanticReport"), assemble_result=assembled)
|
||
cand_id = persisted["candidate_id"]
|
||
print(f"writer 产出: sha256={candidate['candidateSha256'][:24]}..., "
|
||
f"实际模型={receipt.actual_model_id}, 成本=${receipt.total_cost_usd}")
|
||
print(f"落库: candidate_id={cand_id}, receipt_id={persisted['receipt_id']}, "
|
||
f"raw_content_id={persisted['raw_content_id']}, state={persisted['state']}, "
|
||
f"semantic={persisted['semantic_status']}")
|
||
|
||
# 6) 接受前置检查：实时状态重读 + 纯函数全检（上下文/授权/来源/有效期/detector 终态）
|
||
try:
|
||
live_state = build_live_acceptance_state(final_context)
|
||
preflight = check_writer_acceptance(
|
||
decision="accept", confirmed=True, context=final_context, candidate=candidate,
|
||
detector_result=pipeline_result, live_state=live_state,
|
||
expected_revision=live_state["canonicalRevision"])
|
||
except (LiveStateError, AcceptanceError) as exc:
|
||
finish_run(run_id, "failed", creator="continuation",
|
||
trigger_detail={"stage": "accept-preflight",
|
||
"error": getattr(exc, "code", type(exc).__name__)})
|
||
print(f"[停止] 接受前置检查未通过: {getattr(exc, 'code', '')} {exc}")
|
||
print(f"候选 {cand_id} 已留库（state=passed, semantic=passed），人工复核后决定。")
|
||
print(f"\nCANDIDATE_ID={cand_id}\nRUN_ID={run_id}")
|
||
raise SystemExit(1)
|
||
print(f"接受前置检查: {preflight['status']} canonicalRevision={live_state['canonicalRevision']}")
|
||
|
||
# 7) 接受为正式正文（confirm 唯一通道；DB 兜底复检 state + semantic_status）
|
||
semantic_report_sha = str(final_trace.get("semanticReport", {}).get("reportSha256") or "")[:24]
|
||
accept_result = accept(
|
||
cand_id, decided_by="1",
|
||
rationale=(f"第{target}章生产路径：冻结上下文(asOf={as_of}) → writer sandbox 真调 → "
|
||
f"机械门通过 → 语义 detector 通过(报告sha256={semantic_report_sha}...) "
|
||
f"→ accept_preflight {preflight['status']}。"),
|
||
basis_ref=f"细纲@v1 (第{target}章 confirmed) + 前章正文基线",
|
||
expected_revision=live_state["canonicalRevision"],
|
||
command_id=f"accept-work12-ch{target}-{run_id}",
|
||
projection_kinds=("extraction",), # 提交后副作用：章后抽取投影随提交登记，异步执行
|
||
dry_run=dry_run)
|
||
print(f"接受结果: {json.dumps(accept_result, ensure_ascii=False)}")
|
||
if dry_run:
|
||
print("[试跑] 全链验证完成：正文未动，候选已留 Shadow，等待用户三决策。")
|
||
print(f"\nCANDIDATE_ID={cand_id}\nRUN_ID={run_id}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|