symfonic.agent.middleware.metacognitive

metacognitive

MetacognitiveMiddleware -- post-draft reflection pass.

Opt-in via FrameworkConfig.metacognition_enabled.

Gate (both disjuncts OR-ed): (a) compute_confidence(activation_log) < config.metacognition_confidence_threshold (b) detect_sensitive_hit(activation_log, sensitive_tags) is not None

On fire: ONE hidden LLM call comparing draft vs. core_preferences and procedural rules retrieved via the activation_log's inference_paths. Returns verdict in {approve, revise, refuse}.

MetacognitiveMiddleware

MetacognitiveMiddleware(
    *,
    model_provider: Any,
    config: FrameworkConfig,
    domain: DomainTemplate,
    procedural_layer: Any | None = None,
    reflection_prompt_template: str | None = None,
    callback_manager: CallbackManager | None = None,
    observability_hook: ObservabilityHook | None = None,
)

Post-draft reflection pass. Opt-in via FrameworkConfig.metacognition_enabled.

Gate (both disjuncts OR-ed): (a) compute_confidence(activation_log) < config.metacognition_confidence_threshold (b) detect_sensitive_hit(activation_log, sensitive_tags) is not None

On fire: ONE hidden LLM call comparing draft vs. core_preferences and procedural rules retrieved via the activation_log's inference_paths. Returns verdict in {approve, revise, refuse}.

Source code in src/symfonic/agent/middleware/metacognitive.py

def __init__(
    self,
    *,
    model_provider: Any,
    config: FrameworkConfig,
    domain: DomainTemplate,
    procedural_layer: Any | None = None,
    reflection_prompt_template: str | None = None,
    callback_manager: CallbackManager | None = None,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    self._model_provider = model_provider
    self._config = config
    self._domain = domain
    self._procedural_layer = procedural_layer
    self._prompt_template = reflection_prompt_template or _DEFAULT_REFLECTION_PROMPT
    # v7.4.3 Jarvio Ask 5: opt-in callback manager so the critic LLM
    # call emits an ``on_llm_end`` with ``node_name="metacognition_critic"``
    # (mirrors ``react.py:303-311``).  ``None`` preserves byte-identical
    # pre-7.4.3 behaviour for callers that do not pass a manager.
    self._callback_manager = callback_manager
    # v8.3.1 observability-bug fix: the critic LLM call was INVISIBLE to
    # the always-on ``ObservabilityHook`` path (cortex/OTel spans) because
    # it emitted ONLY on the callback-manager path.  React emits on BOTH
    # (``react.py:1078``/``:1433``).  We now mirror react: emit
    # ``on_llm_start``/``on_llm_end`` around the critic call.  ``None``
    # resolves to a ``NoOpObservabilityHook`` so the default is
    # byte-identical (additive only).  ATTRIBUTION is MODEL-BASED: the
    # hook ``on_llm_end`` has no ``node_name`` (signature
    # ``(model, output, usage, run_id)``), so cortex distinguishes the
    # critic from react by the resolved SKU (sonnet/haiku vs opus).
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

enabled property

enabled: bool

Proxies FrameworkConfig.metacognition_enabled.

evaluate async

evaluate(
    *,
    run_id: str,
    draft: str,
    activation_log: dict[str, Any],
    scope: Any,
    callback_manager: CallbackManager | None = None,
    callback_node_name: str = "metacognition_critic",
    tool_calls_this_turn: list[dict[str, Any]]
    | None = None,
    fabrication_findings: list[FabricationFinding]
    | None = None,
    intent_verdict: IntentVerdict | None = None,
) -> ReflectionEvent

Run gate check and (if triggered) the hidden reflection call.

Returns a ReflectionEvent. Caller decides what to do with the verdict. If metacognition is disabled, returns an 'approve' event with confidence_before=1.0, reflection_latency_ms=0.0, and NO LLM call made.

callback_manager (v7.4.3) overrides the constructor-provided manager for this call only; when both are None no callback is emitted (back-compat). When the gate fires AND a manager is in scope, an on_llm_end event with node_name="metacognition_critic" is dispatched after the LLM call (mirrors react.py:303-311).

Source code in src/symfonic/agent/middleware/metacognitive.py

async def evaluate(
    self,
    *,
    run_id: str,
    draft: str,
    activation_log: dict[str, Any],
    scope: Any,
    callback_manager: CallbackManager | None = None,
    callback_node_name: str = "metacognition_critic",
    tool_calls_this_turn: list[dict[str, Any]] | None = None,
    fabrication_findings: list[FabricationFinding] | None = None,
    intent_verdict: IntentVerdict | None = None,
) -> ReflectionEvent:
    """Run gate check and (if triggered) the hidden reflection call.

    Returns a ReflectionEvent. Caller decides what to do with the verdict.
    If metacognition is disabled, returns an 'approve' event with
    confidence_before=1.0, reflection_latency_ms=0.0, and NO LLM call made.

    ``callback_manager`` (v7.4.3) overrides the constructor-provided
    manager for this call only; when both are ``None`` no callback is
    emitted (back-compat).  When the gate fires AND a manager is in
    scope, an ``on_llm_end`` event with
    ``node_name="metacognition_critic"`` is dispatched after the LLM
    call (mirrors ``react.py:303-311``).
    """
    if not self._config.metacognition_enabled:
        return ReflectionEvent(
            run_id=run_id,
            verdict="approve",
            confidence_before=1.0,
            sensitive_tag_hit=None,
            original_draft=draft,
            revised_draft=None,
            refusal_message=None,
            matched_preferences=[],
            reflection_latency_ms=0.0,
        )

    t0 = time.monotonic()
    confidence = compute_confidence(activation_log)

    # Union of per-turn (domain) and global (config) sensitive tags
    sensitive_tags = frozenset(
        t
        for p in self._domain.core_preferences
        for t in p.sensitive_tags
    ) | frozenset(self._config.metacognition_sensitive_tags)

    sensitive_hit = detect_sensitive_hit(activation_log, sensitive_tags)

    if not self._resolve_gate_fires(
        confidence=confidence,
        sensitive_hit=sensitive_hit,
        draft=draft,
        tool_calls_this_turn=tool_calls_this_turn,
        fabrication_findings=fabrication_findings,
        intent_verdict=intent_verdict,
        scope=scope,
        run_id=run_id,
    ):
        return ReflectionEvent(
            run_id=run_id,
            verdict="approve",
            confidence_before=confidence,
            sensitive_tag_hit=None,
            original_draft=draft,
            revised_draft=None,
            refusal_message=None,
            matched_preferences=[],
            reflection_latency_ms=(time.monotonic() - t0) * 1000,
        )

    procedural_rules = await self._load_procedural_rules(
        scope,
        activation_log.get("inference_paths", []),
    )

    # Resolve which manager to use for callback emission:
    # per-call override (engine path) wins over the constructor-provided
    # manager.  Both default to None; in that case no callback fires.
    active_cb_mgr = callback_manager or self._callback_manager

    verdict, revised_or_refusal, matched, tok_in, tok_out = (
        await self._run_reflection_llm(
            draft,
            list(self._domain.core_preferences),
            procedural_rules,
            sensitive_hit,
            run_id=run_id,
            callback_manager=active_cb_mgr,
            callback_node_name=callback_node_name,
        )
    )

    return ReflectionEvent(
        run_id=run_id,
        verdict=verdict,
        confidence_before=confidence,
        sensitive_tag_hit=sensitive_hit,
        original_draft=draft,
        revised_draft=revised_or_refusal if verdict == "revise" else None,
        refusal_message=revised_or_refusal if verdict == "refuse" else None,
        matched_preferences=matched,
        reflection_latency_ms=(time.monotonic() - t0) * 1000,
        reflection_tokens_in=tok_in,
        reflection_tokens_out=tok_out,
    )