symfonic.capabilities.governance.stages.metacognition

metacognition

Metacognition: the final, expensive stage of the governance pipeline.

The default policy carries the legacy selective gate's deterministic evidence forward. Confidence remains one optional signal; it never replaces claims, actions, findings, or recall-time sensitivity. None withholds confidence without inventing a low score.

MetacognitionStage

MetacognitionStage(reflector: Reflector | None, *, confidence_floor: float = 0.6, sensitive_terms: Sequence[str] = (), sensitive_tags: Sequence[str] = (), read_only_tools: Sequence[str] = (), trivial_ack_patterns: Sequence[str] = (), confidence: ConfidenceReporter | None = None, trigger_policy: MetacognitionTriggerPolicy | None = None)

Reflect on the draft when, and only when, the gate fires.

Source code in src/symfonic/capabilities/governance/stages/metacognition.py

def __init__(
    self,
    reflector: Reflector | None,
    *,
    confidence_floor: float = 0.6,
    sensitive_terms: Sequence[str] = (),
    sensitive_tags: Sequence[str] = (),
    read_only_tools: Sequence[str] = (),
    trivial_ack_patterns: Sequence[str] = (),
    confidence: ConfidenceReporter | None = None,
    trigger_policy: MetacognitionTriggerPolicy | None = None,
) -> None:
    self._reflector = reflector
    self._floor = validate_floor(confidence_floor)
    self._confidence = check_reporter(confidence)
    self._policy = _usable_policy(trigger_policy) or SelectiveMetacognitionPolicy(
        confidence_floor=confidence_floor,
        sensitive_terms=sensitive_terms,
        sensitive_tags=sensitive_tags,
        read_only_tools=read_only_tools,
        trivial_ack_patterns=trivial_ack_patterns,
    )

check_reporter

check_reporter(confidence: ConfidenceReporter | None) -> ConfidenceReporter | None

The reporter, or a refusal raised where nothing will contain it.

This stage is FAIL_OPEN: the pipeline catches whatever apply raises and allows the turn. So a per-turn refusal cannot protect a composed caller -- it produces the disabled gate :func:_validated exists to prevent, recorded in degraded_stages and nowhere a caller is obliged to look. A port that can never work is knowable before any turn runs, and construction happens outside the containment boundary, so that is where it is refused.

An async reporter is the mistake worth naming: :class:Reflector sits beside this port and is async, and runtime_checkable cannot tell them apart because it only checks that the method exists. Left to run, it returns an un-awaited coroutine every turn -- a warning, a leak, and a gate that quietly stopped guarding.

Source code in src/symfonic/capabilities/governance/stages/metacognition_validation.py

def check_reporter(confidence: ConfidenceReporter | None) -> ConfidenceReporter | None:
    """The reporter, or a refusal raised where nothing will contain it.

    This stage is ``FAIL_OPEN``: the pipeline catches whatever ``apply``
    raises and allows the turn. So a per-turn refusal cannot protect a
    composed caller -- it *produces* the disabled gate :func:`_validated`
    exists to prevent, recorded in ``degraded_stages`` and nowhere a caller
    is obliged to look. A port that can never work is knowable before any
    turn runs, and construction happens outside the containment boundary, so
    that is where it is refused.

    An async reporter is the mistake worth naming: :class:`Reflector` sits
    beside this port and *is* async, and ``runtime_checkable`` cannot tell
    them apart because it only checks that the method exists. Left to run, it
    returns an un-awaited coroutine every turn -- a warning, a leak, and a
    gate that quietly stopped guarding.
    """
    if confidence is None:
        return None
    method = getattr(confidence, "confidence", None)
    if not callable(method):
        raise ValueError(
            "a confidence reporter must provide a callable confidence(subject, "
            f"context); {type(confidence).__name__} does not"
        )
    if is_async_callable(method):
        raise ValueError(
            "a confidence reporter must be synchronous: the gate decides "
            "whether the one port that costs a model call is worth calling, "
            f"so the signal it decides on cannot cost one. {type(confidence).__name__}"
            ".confidence is async"
        )
    try:
        inspect.signature(method).bind(object(), object())
    except TypeError as exc:
        raise ValueError(
            "a confidence reporter must accept confidence(subject, context); "
            f"{type(confidence).__name__}.confidence does not: {exc}"
        ) from None
    except ValueError:
        # Some C-backed callables have no inspectable signature. They remain
        # valid; their per-turn result still has the conservative value policy.
        pass
    return confidence