muse-agent-example/.claude/skills/check-content-consistency/scripts/check_writer_candidate.py

#!/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_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))
    # 按 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"))
        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"]