symfonic.agent.middleware

middleware

symfonic.agent.middleware -- post-draft reflection and metacognition.

Public API

MetacognitiveMiddleware -- the middleware class ReflectionEvent -- frozen event emitted after reflection ReflectionVerdict -- Literal type alias compute_confidence -- deterministic heuristic detect_sensitive_hit -- sensitive-tag detection

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,
    )

ReflectionEvent dataclass

ReflectionEvent(
    run_id: str,
    verdict: ReflectionVerdict,
    confidence_before: float,
    sensitive_tag_hit: str | None,
    original_draft: str,
    revised_draft: str | None,
    refusal_message: str | None,
    matched_preferences: list[str] = list(),
    reflection_latency_ms: float = 0.0,
    reflection_tokens_in: int = 0,
    reflection_tokens_out: int = 0,
)

Emitted by MetacognitiveMiddleware after a reflection pass.

compute_confidence

compute_confidence(activation_log: dict[str, Any]) -> float

Deterministic heuristic in [0.0, 1.0]. Returns 0.5 on empty input.

Formula

confidence = clamp01( 0.4 * seed_strength + 0.3 * path_coverage + 0.3 * edge_weight_factor )

Where

seed_strength = avg score of activated_nodes (>= 1.0 if seeded) path_coverage = min(1.0, len(inference_paths) / 3) edge_weight_factor = avg edge weight of traversed_edges / 5.0, capped at 1.0

Public so tests can exercise edge cases without spinning up a full agent.

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

def compute_confidence(activation_log: dict[str, Any]) -> float:
    """Deterministic heuristic in [0.0, 1.0]. Returns 0.5 on empty input.

    Formula:
        confidence = clamp01(
            0.4 * seed_strength
          + 0.3 * path_coverage
          + 0.3 * edge_weight_factor
        )

    Where:
        seed_strength  = avg score of activated_nodes (>= 1.0 if seeded)
        path_coverage  = min(1.0, len(inference_paths) / 3)
        edge_weight_factor = avg edge weight of traversed_edges / 5.0, capped at 1.0

    Public so tests can exercise edge cases without spinning up a full agent.
    """
    if not activation_log:
        return 0.5

    activated_nodes = activation_log.get("activated_nodes") or []
    inference_paths = activation_log.get("inference_paths") or []
    traversed_edges = activation_log.get("traversed_edges") or []

    # seed_strength: average score of activated_nodes
    if activated_nodes:
        scores = [n.get("score", 0.0) for n in activated_nodes]
        seed_strength = sum(scores) / len(scores)
    else:
        return 0.5  # cold start — neutral

    # path_coverage
    path_coverage = min(1.0, len(inference_paths) / 3.0)

    # edge_weight_factor
    if traversed_edges:
        weights = [e.get("weight", 0.0) for e in traversed_edges]
        edge_weight_factor = min(1.0, (sum(weights) / len(weights)) / 5.0)
    else:
        edge_weight_factor = 0.0

    confidence = (
        0.4 * seed_strength
        + 0.3 * path_coverage
        + 0.3 * edge_weight_factor
    )
    return _clamp01(confidence)

detect_sensitive_hit

detect_sensitive_hit(
    activation_log: dict[str, Any],
    sensitive_tags: frozenset[str],
) -> str | None

Return the first matching tag when any activated_node carries one of sensitive_tags in properties['tags'], else None.

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

def detect_sensitive_hit(
    activation_log: dict[str, Any],
    sensitive_tags: frozenset[str],
) -> str | None:
    """Return the first matching tag when any activated_node carries one of
    *sensitive_tags* in properties['tags'], else None."""
    if not sensitive_tags:
        return None

    activated_nodes = (activation_log or {}).get("activated_nodes") or []
    for node in activated_nodes:
        node_tags = node.get("tags") or node.get("properties", {}).get("tags") or []
        for tag in node_tags:
            if tag in sensitive_tags:
                return str(tag)
    return None