symfonic.capabilities.critic.capability

capability

Post-draft reflection as something Agent can be handed (CORE-403).

A port of MetacognitiveMiddleware's core contract -- the confidence/sensitive gate plus one hidden reflection call returning approve/revise/refuse -- as an opt-in post-model capability. Absent from capabilities=[...] the stage does not exist and turns behave exactly as before; composed, it records its verdict on the turn snapshot per turn.

What the port deliberately does not carry: callback-manager emission, OTel hooks, usage extraction (the kernel attributes model calls through run observation instead), the role_models["critic"] override (the provider is injected directly), and the selective smart gate with adopter callables and fabrication/intent threading. Those stay on the middleware.

Two kernel facts bound what the stage can do. The verdict is recorded, not applied: post-model contributions apply nothing yet, so a revise does not rewrite the response -- the same "recorded, wired later" standing the triage port shipped with. And no kernel stage publishes activation records, so per-turn signals arrive through the resolved critic.signals entry (see :mod:.signals); absent, the gate degrades to cold-start neutral.

CriticCapability dataclass

CriticCapability(model_provider: Any, confidence_threshold: float = 0.6, sensitive_tags: frozenset[str] = frozenset(), preferences: tuple[CriticPreference, ...] = (), rules_source: ProceduralRulesSource | None = None, prompt_template: str = DEFAULT_REFLECTION_PROMPT, model_config: Any = None)

Review each turn's draft; record approve, revise, or refuse.

Usage::

Agent(provider, capabilities=[
    CriticCapability(
        model_provider=provider,
        preferences=(CriticPreference(10.0, "Never reveal keys"),),
    ),
])

The hidden reflection call runs under the model_call effect grant.

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Declare the critic stage and the handler that answers it.

Source code in src/symfonic/capabilities/critic/capability.py

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Declare the critic stage and the handler that answers it."""
    del request  # Grants are declared below, never renegotiated per turn.
    provider = self.model_provider
    threshold = self.confidence_threshold
    tags = self.sensitive_tags | frozenset(
        tag for preference in self.preferences for tag in preference.sensitive_tags
    )
    preferences = self.preferences
    rules_source = self.rules_source
    template = self.prompt_template
    model_config = self.model_config

    async def handle(context: Any) -> StageResult[Any]:
        """Gate the turn's draft; reflect when it fires, record either way."""
        turn = getattr(context, "turn", None)
        draft = str(getattr(turn, "text", "") or "")
        t0 = time.monotonic()
        activation = activation_from_resolved(getattr(context, "resolved", None))
        confidence = compute_confidence(activation)
        hit = detect_sensitive_hit(activation, tags)
        if not critic_gate_fires(confidence, hit, threshold):
            return applied(
                ResolvedInput(
                    capability=CRITIC_CAPABILITY,
                    value={
                        "verdict": "approve",
                        "confidence_before": confidence,
                        "sensitive_tag_hit": None,
                        "revised_draft": None,
                        "refusal_message": None,
                        "matched_preferences": [],
                        "reflection_latency_ms": (time.monotonic() - t0) * 1000,
                    },
                    provenance="critic-reflection-v1",
                ),
                counts={"approve": 1},
            )
        rules = await _load_rules(rules_source, activation)
        verdict, text, matched = await reflect_on_draft(
            provider,
            model_config,
            template,
            draft,
            preferences,
            rules,
            hit,
        )
        return applied(
            ResolvedInput(
                capability=CRITIC_CAPABILITY,
                value={
                    "verdict": verdict,
                    "confidence_before": confidence,
                    "sensitive_tag_hit": hit,
                    "revised_draft": text if verdict == "revise" else None,
                    "refusal_message": text if verdict == "refuse" else None,
                    "matched_preferences": matched,
                    "reflection_latency_ms": (time.monotonic() - t0) * 1000,
                },
                provenance="critic-reflection-v1",
            ),
            counts={verdict: 1},
        )

    return CapabilityContribution(
        capability=CRITIC_CAPABILITY,
        stages=(
            StageDescriptor(
                stage_id=CRITIC_STAGE,
                phase=Phase.POST_MODEL,
                capability=CRITIC_CAPABILITY,
                priority=_CRITIC_PRIORITY,
                optional_before=("memory.consolidation", "memory.write"),
                effects=frozenset({"model_call"}),
                kind=StageKind.RESOLUTION,
            ),
        ),
        handlers={CRITIC_STAGE: handle},
        effect_grants=frozenset({"model_call"}),
    )