core/core/epistemic_state.py

"""First-class epistemic-state and normative-clearance enums.

Phase 3 of the epistemic-state program makes the ratified taxonomy
queryable from runtime artifacts without entangling callers with the
scope document.  The values are intentionally lower_snake_case so they
serialize stably into JSONL, metadata dictionaries, and test fixtures.
"""

from __future__ import annotations

from enum import Enum, unique
from typing import Any, Literal, get_args


GroundingSource = Literal[
    "pack",
    "teaching",
    "vault",
    "partial",
    "oov",
    "none",
    "deduction",
    "curriculum",
]
"""Ratified grounding-source labels.

Single source of truth for the values ``epistemic_state_for_grounding_source``
maps, the cold-start-grounding lane validates, the Workbench API coerces
against, and the Workbench UI badges bind to (ADR-0162 §3d).  Adding a value
here without adding a corresponding badge fails the build-time enum coverage
test under ``workbench-ui/``.

``deduction`` (ADR-0256) and ``curriculum`` (ADR-0262) joined the closed set
once deduction serving was ratified ON: ``workbench/api.py``'s live chat route
builds a bare ``ChatRuntime()``, so those composers decide Workbench turns and
stamp their label on the ``TurnEvent``.  While the labels were unregistered the
API's coercion floored them to ``"none"`` — recording a proved answer as
ungrounded in a durable audit artifact.  Registration is what makes the
recorded provenance true; ``GROUNDING_SOURCES`` below is what keeps any second
copy of this set from drifting away from it again.
"""

GROUNDING_SOURCES: frozenset[str] = frozenset(get_args(GroundingSource))
"""The runtime-iterable form of :data:`GroundingSource`.

A ``Literal`` is invisible at runtime, which is why consumers historically
restated its members by hand — and why one of those restatements silently fell
behind.  Anything that needs to *check* a grounding source reads this set;
nothing re-types the members.
"""


@unique
class EpistemicState(str, Enum):
    PERCEIVED = "perceived"
    EVIDENCED = "evidenced"
    EVIDENCED_INCOMPLETE = "evidenced_incomplete"
    VERIFIED = "verified"
    DECODED = "decoded"
    DECODED_UNARTICULATED = "decoded_unarticulated"
    INFERRED = "inferred"
    UNVERIFIED_POSSIBLE = "unverified_possible"
    UNVERIFIED_NOVEL = "unverified_novel"
    CONTRADICTED = "contradicted"
    AMBIGUOUS = "ambiguous"
    UNDETERMINED = "undetermined"
    SCOPE_BOUNDARY = "scope_boundary"
    COMPUTATIONALLY_BOUNDED = "computationally_bounded"
    EPISTEMIC_STATE_NEEDED = "epistemic_state_needed"


@unique
class NormativeClearance(str, Enum):
    CLEARED = "cleared"
    VIOLATED = "violated"
    UNASSESSABLE = "unassessable"
    SUPPRESSED = "suppressed"


def coerce_epistemic_state(value: object | None, *, default: EpistemicState = EpistemicState.UNDETERMINED) -> EpistemicState:
    if isinstance(value, EpistemicState):
        return value
    if isinstance(value, str):
        normalized = value.strip().lower().replace("-", "_")
        for state in EpistemicState:
            if normalized in {state.value, state.name.lower()}:
                return state
    return default


def coerce_normative_clearance(value: object | None, *, default: NormativeClearance = NormativeClearance.UNASSESSABLE) -> NormativeClearance:
    if isinstance(value, NormativeClearance):
        return value
    if isinstance(value, str):
        normalized = value.strip().lower().replace("-", "_")
        for clearance in NormativeClearance:
            if normalized in {clearance.value, clearance.name.lower()}:
                return clearance
    return default


def clearance_from_verdicts(verdicts: Any = None, *, safety_verdict: Any = None, ethics_verdict: Any = None) -> NormativeClearance:
    """Derive the orthogonal normative-clearance state for a turn.

    Refusal replacement is a SUPPRESSED surface.  Otherwise any failed
    safety/ethics verdict is VIOLATED.  If every available verdict is
    upheld but at least one result was not runtime-checkable, the turn is
    UNASSESSABLE rather than positively CLEARED.  Only fully upheld and
    fully runtime-checkable verdicts are CLEARED.
    """
    if verdicts is not None:
        if bool(getattr(verdicts, "refusal_emitted", False)):
            return NormativeClearance.SUPPRESSED
        safety_verdict = safety_verdict if safety_verdict is not None else getattr(verdicts, "safety_verdict", None)
        ethics_verdict = ethics_verdict if ethics_verdict is not None else getattr(verdicts, "ethics_verdict", None)

    saw_unassessable = False
    saw_verdict = False
    for verdict in (safety_verdict, ethics_verdict):
        if verdict is None:
            continue
        saw_verdict = True
        if not bool(getattr(verdict, "upheld", True)):
            return NormativeClearance.VIOLATED
        results = tuple(getattr(verdict, "results", ()) or ())
        if any(not bool(getattr(result, "runtime_checkable", True)) for result in results):
            saw_unassessable = True
        elif int(getattr(verdict, "runtime_checkable_count", 0) or 0) == 0:
            saw_unassessable = True
    if not saw_verdict or saw_unassessable:
        return NormativeClearance.UNASSESSABLE
    return NormativeClearance.CLEARED


def normative_detail_from_verdicts(verdicts: Any = None, *, safety_verdict: Any = None, ethics_verdict: Any = None) -> str:
    """Return a sorted, comma-separated string of violated boundary/commitment IDs.

    Non-empty only when normative clearance is VIOLATED or SUPPRESSED —
    i.e. when at least one runtime-checkable boundary or commitment failed.
    Returns "" when CLEARED or UNASSESSABLE so callers can test truthiness.
    Uses the same duck-typing pattern as ``clearance_from_verdicts``.
    """
    if verdicts is not None:
        safety_verdict = safety_verdict if safety_verdict is not None else getattr(verdicts, "safety_verdict", None)
        ethics_verdict = ethics_verdict if ethics_verdict is not None else getattr(verdicts, "ethics_verdict", None)

    ids: list[str] = []
    for r in getattr(safety_verdict, "results", []):
        if getattr(r, "runtime_checkable", False) and not getattr(r, "upheld", True):
            bid = getattr(r, "boundary_id", None)
            if bid:
                ids.append(str(bid))
    for r in getattr(ethics_verdict, "results", []):
        if getattr(r, "runtime_checkable", False) and not getattr(r, "upheld", True):
            cid = getattr(r, "commitment_id", None)
            if cid:
                ids.append(str(cid))
    return ",".join(sorted(ids))


def epistemic_state_for_grounding_source(source: str | None) -> EpistemicState:
    """Default runtime mapping for existing grounding-source labels."""
    normalized = (source or "none").strip().lower()
    # ``deduction`` / ``curriculum`` rank with the decoded sources rather than
    # the evidenced ones: both are *decided* by the ROBDD entailment engine
    # over ratified structure (an argument's own premises, or a subject's
    # ratified curriculum), not retrieved as supporting evidence.  Serving them
    # authoritatively additionally requires a SERVE license on the sealed
    # ledger — an unearned band is served DISCLOSED, which is a surface-level
    # hedge and does not change how the turn was grounded.
    if normalized in {"pack", "teaching", "vault", "deduction", "curriculum"}:
        return EpistemicState.DECODED
    if normalized == "partial":
        return EpistemicState.EVIDENCED_INCOMPLETE
    if normalized == "oov":
        return EpistemicState.UNVERIFIED_NOVEL
    if normalized == "none":
        return EpistemicState.UNDETERMINED
    return EpistemicState.EPISTEMIC_STATE_NEEDED


__all__ = [
    "GROUNDING_SOURCES",
    "EpistemicState",
    "GroundingSource",
    "NormativeClearance",
    "clearance_from_verdicts",
    "coerce_epistemic_state",
    "coerce_normative_clearance",
    "epistemic_state_for_grounding_source",
    "normative_detail_from_verdicts",
]