Jyotisha/scripts/rectification_purpose_gate/__init__.py

"""出生时间「按用途够用」的用途决策合同（BUG-1833，#186 第一步）。

移植自占星师 2026-11-09 的参考决策器
（``docs/tasks/assets/purpose-gate-20261109/reference_gate.py``）：
没有区间宽度门槛，也没有分数差门槛——这是参考代码刻意为之。

本模块只回答「这个已经算好的评估，可以在什么范围内被当作够用」，
不做任何计算、不改出生记录、不证明调用方给的证据为真：

- ``can_use_interval`` 只有 ``scope_ready`` 才为真；
- ``can_modify_birth_record`` 与 ``representative_is_birth_truth`` 永远为假。

**尚未接入**：本单只交付合同与测试，没有任何生产入口读它，
``scope_ready`` 也不会影响任何现有流程。
"""

from __future__ import annotations

from typing import Any, Dict, Final, List, Mapping, Sequence, Tuple

PURPOSE_USE_SCHEMA: Final[str] = 'jyotish.purpose_use_decision.v1'

BINDING_KEYS: Final[Tuple[str, ...]] = (
    'input_revision', 'candidate_set_fingerprint', 'purpose_fingerprint',
    'profile_fingerprint', 'engine_version', 'precision_policy_version',
    'forecast_window_fingerprint',
)

LAYER_COVERAGE_STATES: Final[Tuple[str, ...]] = ('executed_complete', 'partial', 'blocked')
SCAN_STATES: Final[Tuple[str, ...]] = ('policy_covered', 'sampled_only', 'incomplete')
BRANCH_STATES: Final[Tuple[str, ...]] = (
    'resolved', 'bounded_equivalent', 'unresolved_material', 'not_evaluated',
)

STATUS_SCOPE_READY: Final[str] = 'scope_ready'
STATUS_CONDITIONAL: Final[str] = 'conditional'
STATUS_NEEDS_REFINEMENT: Final[str] = 'needs_refinement'
STATUS_INSUFFICIENT_EVIDENCE: Final[str] = 'insufficient_evidence'
STATUS_STALE: Final[str] = 'stale'


def _binding_is_complete(binding: Any) -> bool:
    return isinstance(binding, dict) and all(
        isinstance(binding.get(key), str) and binding[key] for key in BINDING_KEYS
    )


def _claim_ids(assessment: Mapping[str, Any], field: str) -> List[str]:
    values = assessment.get(field)
    if not isinstance(values, list) or any(not isinstance(value, str) or not value for value in values):
        raise ValueError(f'invalid {field}')
    if len(set(values)) != len(values):
        raise ValueError(f'duplicate {field}')
    return list(values)


def _validate_assessment(assessment: Mapping[str, Any]) -> None:
    """校验绑定一致的评估内容，字段不合法时抛出 ValueError。"""
    required = assessment.get('required_layers')
    coverage = assessment.get('layer_coverage')
    if (not isinstance(required, list) or not required
            or any(not isinstance(layer, str) or not layer for layer in required)
            or len(set(required)) != len(required)
            or not isinstance(coverage, dict)):
        raise ValueError('explicit nonempty required layer set required')
    if any(coverage.get(layer) not in LAYER_COVERAGE_STATES for layer in required):
        raise ValueError('required layers cannot be absent or not_applicable')
    if assessment.get('scan_state') not in SCAN_STATES:
        raise ValueError('invalid scan_state')
    if assessment.get('branch_state') not in BRANCH_STATES:
        raise ValueError('invalid branch_state')
    for flag in ('domain_complete', 'material_change'):
        if type(assessment.get(flag)) is not bool:  # noqa: E721 - 只认真正的 bool，不认 0/1
            raise ValueError(f'{flag} must be a boolean')
    _claim_ids(assessment, 'stable_claim_ids')
    _claim_ids(assessment, 'conditional_claim_ids')
    if set(assessment['stable_claim_ids']) & set(assessment['conditional_claim_ids']):
        raise ValueError('one fact cannot be both stable and conditional')


def _result(status: str, stable: Sequence[str], conditional: Sequence[str], reason: str) -> Dict[str, Any]:
    allowed = list(dict.fromkeys(list(stable) + list(conditional)))
    return {
        'schema': PURPOSE_USE_SCHEMA,
        'status': status,
        'can_use_interval': status == STATUS_SCOPE_READY,
        'can_continue_analysis': bool(allowed),
        'can_modify_birth_record': False,
        'representative_is_birth_truth': False,
        'stable_claim_ids': list(stable),
        'conditional_claim_ids': list(conditional),
        'allowed_claim_ids': allowed,
        'reason': reason,
    }


def assess_use(assessment: dict, expected_binding: dict) -> dict:
    """Return permissions for the evaluated scope, not birth-time confidence.

    Deliberately has no interval-width or score-margin threshold.
    stable_claim_ids must cover the entire remaining candidate domain;
    conditional_claim_ids must point to explicit branch/limitation evidence.
    """
    binding = assessment.get('binding')
    for candidate in (binding, expected_binding):
        if not _binding_is_complete(candidate):
            raise ValueError('complete server binding required')
    if any(binding[key] != expected_binding[key] for key in BINDING_KEYS):
        return _result(STATUS_STALE, [], [], 'binding_changed_recompute')

    _validate_assessment(assessment)
    required = assessment['required_layers']
    coverage = assessment['layer_coverage']
    stable = assessment['stable_claim_ids']
    conditional = assessment['conditional_claim_ids']

    if assessment['material_change']:
        return _result(STATUS_NEEDS_REFINEMENT, stable, conditional, 'material_output_change')
    complete = all(coverage[layer] == 'executed_complete' for layer in required)
    if (complete and assessment['domain_complete']
            and assessment['scan_state'] == 'policy_covered'
            and assessment['branch_state'] in ('resolved', 'bounded_equivalent')
            and stable and not conditional):
        return _result(STATUS_SCOPE_READY, stable, [], 'adequate_for_frozen_purpose')
    if stable or conditional:
        return _result(STATUS_CONDITIONAL, stable, conditional, 'only_listed_claims_are_available')
    return _result(STATUS_INSUFFICIENT_EVIDENCE, [], [], 'no_supported_outputs_for_purpose')