symfonic.capabilities.critic

critic

Post-draft reflection as an opt-in capability (CORE-403).

A port of MetacognitiveMiddleware's core contract -- the confidence/sensitive gate plus one hidden reflection call returning approve/revise/refuse -- as CriticCapability. Composed, it records its verdict on the turn snapshot; absent, the stage does not exist and turns behave exactly as before. Recorded, not applied: post-model contributions apply nothing yet, so wiring the verdict into the response is later work.

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

CriticPreference dataclass

CriticPreference(priority: float, statement: str, sensitive_tags: frozenset[str] = frozenset())

One rule the draft must respect, as values (no domain import).

ProceduralRulesSource

Bases: Protocol

Per-turn procedural rules, supplied by the composition root.

retrieve async

retrieve(query: str, top_k: int) -> Sequence[str]

Return up to top_k rule contents for query.

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

async def retrieve(self, query: str, top_k: int) -> Sequence[str]:
    """Return up to ``top_k`` rule contents for ``query``."""
    ...

activation_from_resolved

activation_from_resolved(resolved: Any) -> dict[str, Any]

The turn's activation record, or {} when no producer supplied one.

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

def activation_from_resolved(resolved: Any) -> dict[str, Any]:
    """The turn's activation record, or ``{}`` when no producer supplied one."""
    if resolved is None:
        return {}
    if isinstance(resolved, Mapping):
        entry = resolved.get(SIGNALS_ENTRY)
    else:
        get = getattr(resolved, "get", None)
        entry = get(SIGNALS_ENTRY) if callable(get) else None
    if entry is None:
        return {}
    value = getattr(entry, "value", entry)
    return dict(value) if isinstance(value, Mapping) else {}

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

Source code in src/symfonic/capabilities/critic/signals.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
    """
    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)

critic_gate_fires

critic_gate_fires(confidence: float, sensitive_hit: str | None, threshold: float) -> bool

Fire on low confidence OR a sensitive hit (legacy gate, verbatim).

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

def critic_gate_fires(
    confidence: float, sensitive_hit: str | None, threshold: float
) -> bool:
    """Fire on low confidence OR a sensitive hit (legacy gate, verbatim)."""
    if confidence < threshold:
        return True
    return sensitive_hit is not None

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/capabilities/critic/signals.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