091b66a9bb
将角色与 Skill 从 .claude 迁入 .agent，移除 Claude CLI 运行时并接入固定 Opus 角色 profile、完整 schema、预算 deadline、raw 与回执证据链。 同步拆分 Skill 职责、复利 lesson、Gate 回放、Dashboard 人审入口、数据库登记和机械门禁；候选设计正文不包含在本提交中。
279 lines
11 KiB
Python
279 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""CandidateEnvelope v2 的机械硬门；不调用模型或外部服务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import pathlib
|
||
import sys
|
||
from typing import Any, Mapping, Sequence
|
||
|
||
SCRIPT_DIR = pathlib.Path(__file__).resolve().parent
|
||
READ_CONTEXT_DIR = SCRIPT_DIR.parents[1] / "assemble-context" / "scripts"
|
||
if str(READ_CONTEXT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(READ_CONTEXT_DIR))
|
||
|
||
from writer_contract import ( # noqa: E402
|
||
ContractError,
|
||
han_count,
|
||
normalize_text,
|
||
validate_writer_context,
|
||
validate_writer_output,
|
||
)
|
||
|
||
|
||
SEMANTIC_CHECKS = (
|
||
"hard_event_semantics",
|
||
"claim_discovery_and_truth_alignment",
|
||
"new_setting_candidates",
|
||
"evidence_gaps",
|
||
"knowledge_scope",
|
||
"ability_cost_consistency",
|
||
)
|
||
|
||
|
||
def _failure(code: str, message: str, **details: Any) -> dict[str, Any]:
|
||
"""生成稳定且可追踪的阻塞项。"""
|
||
|
||
return {"code": code, "message": message, **details}
|
||
|
||
|
||
def _anchors_present(body: str, anchors: Any) -> bool:
|
||
"""只执行确定性的文本锚点匹配，语义等价判断留给模型 detector。"""
|
||
|
||
return isinstance(anchors, list) and bool(anchors) and any(
|
||
isinstance(anchor, str) and anchor and normalize_text(anchor) in body for anchor in anchors
|
||
)
|
||
|
||
|
||
def _safe_sequence(value: Any) -> Sequence[Any]:
|
||
"""把非法数组降为空序列，随后由输入合同阻塞项统一报告。"""
|
||
|
||
return value if isinstance(value, list) else ()
|
||
|
||
|
||
def _precheck_candidate_integrity(candidate: Mapping[str, Any]) -> list[dict[str, Any]]:
|
||
"""在严格 schema 前检查候选正文与哈希，保留精确失败码。"""
|
||
|
||
failures: list[dict[str, Any]] = []
|
||
body = candidate.get("candidateBody")
|
||
candidate_hash = candidate.get("candidateSha256")
|
||
if not isinstance(body, str):
|
||
return [_failure("OUTPUT_CONTRACT_INVALID", "candidateBody 必须是字符串")]
|
||
try:
|
||
normalized_body = normalize_text(body)
|
||
except ContractError as exc:
|
||
return [_failure("OUTPUT_CONTRACT_INVALID", str(exc))]
|
||
expected_hash = "sha256:" + hashlib.sha256(normalized_body.encode("utf-8")).hexdigest()
|
||
if candidate_hash != expected_hash:
|
||
failures.append(
|
||
_failure(
|
||
"CANDIDATE_HASH_MISMATCH",
|
||
"candidateSha256 与规范化候选正文不一致",
|
||
expectedSha256=expected_hash,
|
||
)
|
||
)
|
||
return failures
|
||
|
||
|
||
def _check_context_binding(
|
||
context: Mapping[str, Any], candidate: Mapping[str, Any]
|
||
) -> list[dict[str, Any]]:
|
||
"""检查候选元数据来自本次冻结上下文。"""
|
||
|
||
failures: list[dict[str, Any]] = []
|
||
expected = {
|
||
"runId": context["runId"],
|
||
"attempt": context["attempt"],
|
||
"mode": context["mode"],
|
||
"qualityPolicyVersion": context["qualityPolicyVersion"],
|
||
"contextSnapshotId": context["contextSnapshot"]["manifestId"],
|
||
"contextSnapshotSha256": context["contextSnapshot"]["contextSha256"],
|
||
"acceptanceEligible": context["acceptanceEligible"],
|
||
}
|
||
mismatches = [field for field, value in expected.items() if candidate.get(field) != value]
|
||
if mismatches:
|
||
failures.append(
|
||
_failure("CONTEXT_BINDING_MISMATCH", "候选未绑定当前冻结上下文", fields=mismatches)
|
||
)
|
||
return failures
|
||
|
||
|
||
def _check_outline_anchors(body: str, requirements: Mapping[str, Any]) -> list[dict[str, Any]]:
|
||
"""机械检查硬事件、必须出场角色、伏笔动作和章末钩子。"""
|
||
|
||
failures: list[dict[str, Any]] = []
|
||
for item in _safe_sequence(requirements.get("requiredEvents")):
|
||
if not isinstance(item, Mapping) or not _anchors_present(body, item.get("anchors")):
|
||
failures.append(
|
||
_failure(
|
||
"HARD_EVENT_MISSING",
|
||
"候选未命中细纲硬事件锚点",
|
||
requirementId=item.get("requirementId") if isinstance(item, Mapping) else None,
|
||
)
|
||
)
|
||
for character in _safe_sequence(requirements.get("requiredCharacters")):
|
||
if not isinstance(character, str) or not character or normalize_text(character) not in body:
|
||
failures.append(
|
||
_failure("REQUIRED_CHARACTER_MISSING", "细纲要求的角色未出场", character=character)
|
||
)
|
||
for item in _safe_sequence(requirements.get("foreshadowingActions")):
|
||
if not isinstance(item, Mapping) or not _anchors_present(body, item.get("anchors")):
|
||
failures.append(
|
||
_failure(
|
||
"FORESHADOWING_ACTION_MISSING",
|
||
"候选未命中细纲伏笔动作锚点",
|
||
requirementId=item.get("requirementId") if isinstance(item, Mapping) else None,
|
||
)
|
||
)
|
||
hook = requirements.get("chapterEndHook")
|
||
if isinstance(hook, Mapping):
|
||
max_distance = hook.get("maxDistanceFromEnd")
|
||
if isinstance(max_distance, bool) or not isinstance(max_distance, int) or max_distance <= 0:
|
||
failures.append(_failure("DETECTOR_INPUT_INVALID", "章末钩子距离必须是正整数"))
|
||
else:
|
||
tail = body[-max_distance:]
|
||
if not _anchors_present(tail, hook.get("anchors")):
|
||
failures.append(
|
||
_failure(
|
||
"CHAPTER_END_HOOK_MISSING",
|
||
"候选结尾未命中细纲章末钩子",
|
||
requirementId=hook.get("requirementId"),
|
||
)
|
||
)
|
||
else:
|
||
failures.append(_failure("DETECTOR_INPUT_INVALID", "chapterEndHook 必须是对象"))
|
||
return failures
|
||
|
||
|
||
def _check_outline_paraphrase_leak(
|
||
body: str, context: Mapping[str, Any] | None
|
||
) -> list[dict[str, Any]]:
|
||
"""机械拦截：细纲硬约束里的描写短句不得原样/近似复读进正文。
|
||
|
||
硬约束只应写事件与结果；若约束文本里的连续片段（≥4 汉字，去掉空白）
|
||
直接出现在候选正文中，判 OUTLINE_PHRASE_LEAKED。语义等价泄漏仍靠 diagnose。
|
||
"""
|
||
|
||
if context is None:
|
||
return []
|
||
outline = context.get("fineOutline")
|
||
if not isinstance(outline, Mapping):
|
||
return []
|
||
raw_constraints = outline.get("hardConstraints") or []
|
||
if not isinstance(raw_constraints, list):
|
||
return []
|
||
# 去空白后的正文，便于匹配「干净，直接」类短句
|
||
compact_body = "".join(ch for ch in body if not ch.isspace())
|
||
failures: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for item in raw_constraints:
|
||
text = item.get("text") if isinstance(item, Mapping) else item
|
||
if not isinstance(text, str) or not text.strip():
|
||
continue
|
||
# 按中文标点切成片段；只检查足够长的描写性短句
|
||
chunks: list[str] = []
|
||
buf: list[str] = []
|
||
for ch in text:
|
||
if ch in "，。；、：；!！?？;,:":
|
||
if buf:
|
||
chunks.append("".join(buf))
|
||
buf = []
|
||
elif not ch.isspace():
|
||
buf.append(ch)
|
||
if buf:
|
||
chunks.append("".join(buf))
|
||
for chunk in chunks:
|
||
if len(chunk) < 4:
|
||
continue
|
||
# 事件骨架常含专名+动词，过短专名已过滤；命中即泄漏
|
||
if chunk in compact_body and chunk not in seen:
|
||
seen.add(chunk)
|
||
failures.append(
|
||
_failure(
|
||
"OUTLINE_PHRASE_LEAKED",
|
||
"候选正文复读了细纲硬约束措辞；硬约束只写事件/结果，正文须戏剧化展开",
|
||
leakedPhrase=chunk,
|
||
)
|
||
)
|
||
return failures
|
||
|
||
|
||
def check_writer_candidate(
|
||
context: Mapping[str, Any], candidate: Mapping[str, Any], requirements: Mapping[str, Any]
|
||
) -> dict[str, Any]:
|
||
"""运行不含模型调用的机械硬门，并返回结构化 detector 报告。"""
|
||
|
||
failures: list[dict[str, Any]] = []
|
||
if not isinstance(context, Mapping) or not isinstance(candidate, Mapping) or not isinstance(requirements, Mapping):
|
||
failures.append(_failure("DETECTOR_INPUT_INVALID", "context、candidate、requirements 必须是对象"))
|
||
return {
|
||
"schemaVersion": "writer-detector-report-v1",
|
||
"passed": False,
|
||
"blockingFailures": failures,
|
||
"semanticReview": {"status": "pending", "checks": list(SEMANTIC_CHECKS)},
|
||
}
|
||
try:
|
||
normalized_context = validate_writer_context(context)
|
||
except ContractError as exc:
|
||
failures.append(_failure("CONTEXT_CONTRACT_INVALID", str(exc)))
|
||
normalized_context = None
|
||
failures.extend(_precheck_candidate_integrity(candidate))
|
||
try:
|
||
normalized_candidate = validate_writer_output(candidate)
|
||
except ContractError as exc:
|
||
failures.append(_failure("OUTPUT_CONTRACT_INVALID", str(exc)))
|
||
normalized_candidate = None
|
||
if normalized_context is not None:
|
||
failures.extend(_check_context_binding(normalized_context, candidate))
|
||
body = candidate.get("candidateBody")
|
||
if isinstance(body, str):
|
||
normalized_body = normalize_text(body)
|
||
if normalized_context is not None:
|
||
contract = normalized_context["outputContract"]
|
||
actual_han_chars = han_count(normalized_body)
|
||
if not contract["minChars"] <= actual_han_chars <= contract["maxChars"]:
|
||
failures.append(
|
||
_failure(
|
||
"CANDIDATE_LENGTH_OUT_OF_RANGE",
|
||
"候选正文汉字数超出动态篇幅合同",
|
||
actualHanChars=actual_han_chars,
|
||
minChars=contract["minChars"],
|
||
maxChars=contract["maxChars"],
|
||
targetChars=contract["targetChars"],
|
||
)
|
||
)
|
||
failures.extend(_check_outline_anchors(normalized_body, requirements))
|
||
# 复读门用规范化前的上下文细纲 + 规范化正文
|
||
failures.extend(
|
||
_check_outline_paraphrase_leak(
|
||
normalized_body, normalized_context if normalized_context is not None else context
|
||
)
|
||
)
|
||
# 按 code 与定位去重，避免严格合同与预检重复报告同一处问题。
|
||
unique: list[dict[str, Any]] = []
|
||
seen: set[tuple[Any, ...]] = set()
|
||
for item in failures:
|
||
key = (item.get("code"), item.get("requirementId"), item.get("character"), item.get("leakedPhrase"))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(item)
|
||
source = normalized_candidate or candidate
|
||
return {
|
||
"schemaVersion": "writer-detector-report-v1",
|
||
"runId": source.get("runId"),
|
||
"attempt": source.get("attempt"),
|
||
"candidateVersion": source.get("candidateVersion"),
|
||
"candidateSha256": source.get("candidateSha256"),
|
||
"passed": not unique,
|
||
"blockingFailures": unique,
|
||
"semanticReview": {
|
||
"status": "pending",
|
||
"checks": list(SEMANTIC_CHECKS),
|
||
"inputContract": "WriterContext v1 + CandidateEnvelope v2 -> SemanticDetection v3",
|
||
},
|
||
}
|
||
|
||
|
||
__all__ = ["SEMANTIC_CHECKS", "check_writer_candidate"]
|