symfonic.agent.middleware.fabrication_wiring

fabrication_wiring

Engine-side wiring for the fabrication verdict (v7.0 T11).

Keeps the hot-path integration outside engine.py (which sits under the documented v7.1-T01 size waiver). Centralises the fabrication_check_mode dispatch table so the engine can call one function per turn.

v7.0.7: typed :class:~symfonic.core.contracts.callbacks.FabricationDetectedEvent dispatch. The previous ExtensionEvent(type='fabrication_detected'| 'fabrication_blocked') payload is preserved as a DEPRECATED back-compat path for one release -- it will be removed in v7.1. See docs/releases/v7.0.7.md.

FabricationOutcome dataclass

FabricationOutcome(
    findings: list[FabricationFinding],
    replacement_draft: str | None = None,
    forced_metacog_verdict: str | None = None,
    emitted_event: bool = False,
)

Result of the fabrication-mode dispatch.

Attributes:

Name	Type	Description
findings	list[FabricationFinding]	All flagged identifiers (empty when clean).
replacement_draft	str | None	Non-None when the caller should replace final_response with this value (refuse mode with findings).
forced_metacog_verdict	str | None	When the caller runs the MetacognitiveMiddleware, whether the fabrication finding should force a revise/refuse verdict. None means "do not override metacog".
emitted_event	bool	True when an ExtensionEvent was fanned out.

apply_fabrication_check async

apply_fabrication_check(
    *,
    config: FrameworkConfig,
    response_text: str | None,
    tool_calls_this_turn: list[dict[str, Any]] | None,
    hydrated_context: str,
    user_message: str,
    identifier_keys: list[str] | None,
    callbacks: Sequence[Any] | None,
    intent_verdict: IntentVerdict | None = None,
    run_id: str = "",
) -> FabricationOutcome

Run the fabrication detector per fabrication_check_mode.

Returns an outcome the caller can consult to decide whether to replace the draft (refuse), force a metacog revise verdict, or simply log telemetry (observe). Does nothing and returns a zero-cost empty outcome when the mode is off.

Parameters:

Name	Type	Description	Default
intent_verdict	IntentVerdict | None	v7.0.6 -- when supplied, enables the language-agnostic intent_action_no_tool finding kind. Pass the same IntentVerdict already computed by run_intent_filter for intent_filter_mode / tool_routing_mode so the classifier runs once per turn. Defaults to None so callers that have not opted into intent classification keep v7.0.5 behaviour.	None
run_id	str	v7.0.7 -- threaded into FabricationDetectedEvent. Empty string by default to preserve back-compat for the (rare) callers that invoke the wiring outside the engine run loop.	''

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

async def apply_fabrication_check(
    *,
    config: FrameworkConfig,
    response_text: str | None,
    tool_calls_this_turn: list[dict[str, Any]] | None,
    hydrated_context: str,
    user_message: str,
    identifier_keys: list[str] | None,
    callbacks: Sequence[Any] | None,
    intent_verdict: IntentVerdict | None = None,
    run_id: str = "",
) -> FabricationOutcome:
    """Run the fabrication detector per ``fabrication_check_mode``.

    Returns an ``outcome`` the caller can consult to decide whether to
    replace the draft (`refuse`), force a metacog ``revise`` verdict,
    or simply log telemetry (`observe`).  Does nothing and returns a
    zero-cost empty outcome when the mode is ``off``.

    Args:
        intent_verdict: v7.0.6 -- when supplied, enables the
            language-agnostic ``intent_action_no_tool`` finding kind.
            Pass the same ``IntentVerdict`` already computed by
            ``run_intent_filter`` for ``intent_filter_mode`` /
            ``tool_routing_mode`` so the classifier runs once per
            turn. Defaults to ``None`` so callers that have not
            opted into intent classification keep v7.0.5 behaviour.
        run_id: v7.0.7 -- threaded into ``FabricationDetectedEvent``.
            Empty string by default to preserve back-compat for the
            (rare) callers that invoke the wiring outside the engine
            run loop.
    """
    mode = config.fabrication_check_mode
    if mode == "off" or not response_text:
        return FabricationOutcome(findings=[])

    findings = detect_fabrication(
        response_text,
        tool_calls_this_turn=tool_calls_this_turn,
        hydrated_context=hydrated_context,
        user_message=user_message,
        identifier_keys=identifier_keys,
        intent_verdict=intent_verdict,
    )
    if not findings:
        return FabricationOutcome(findings=[])

    # v7.0.3: refuse mode only trips when at least one finding clears
    # ``fabrication_refuse_min_confidence``.  observe / revise ignore the
    # threshold -- they always see the full finding list.
    refuse_threshold = float(
        getattr(config, "fabrication_refuse_min_confidence", 0.8)
    )
    refuse_trips = any(
        float(getattr(f, "confidence", 0.8)) >= refuse_threshold
        for f in findings
    )

    replacement: str | None = None
    forced_verdict: str | None = None
    action_taken: str
    if mode == "refuse" and refuse_trips:
        replacement = _SAFE_REFUSAL
        action_taken = "refuse_replaced"
    elif mode == "revise":
        forced_verdict = "revise"
        action_taken = "revise_forced"
    else:
        # observe -> telemetry only
        # refuse + no tripping finding -> telemetry only (draft stays)
        action_taken = "observed"

    emitted = await _emit_event(
        mode=mode,
        action_taken=action_taken,
        findings=findings,
        callbacks=callbacks,
        blocked=(action_taken == "refuse_replaced"),
        response_text=response_text,
        replacement=replacement,
        refuse_threshold=refuse_threshold,
        run_id=run_id,
    )

    return FabricationOutcome(
        findings=findings,
        replacement_draft=replacement,
        forced_metacog_verdict=forced_verdict,
        emitted_event=emitted,
    )