symfonic.core.nodes.precondition_gate

precondition_gate

PreconditionGateNode -- routing-time enforcement of authored-tier procedural preconditions.

v7.7.4 (adopter PRE-FLIGHT enforcement ask, authored tier only). Inserted into the LangGraph topology as the third sibling of elicitation and interrupt in the react fan-out when FrameworkConfig.procedural_enforce_preconditions=True.

Architectural rationale (from the adopter's v7.7.2 push-back eval):

• The elicitation node at presets.py:106-110 IS the routing layer -- we don't need new-graph territory. precondition_gate is a parallel sibling that fires when a different state predicate hits.
• The synthetic ToolMessage(content=..., tool_call_id=...) pattern is already production-ready (see ask_user resume at engine.py:3173-3369). The gate reuses it.
• Authored skills already store structured metadata['precondition'] via seed_authored_skill(... precondition=...) (v7.5.0 / v7.6.2). Zero schema surgery for the authored tier. LLM-extractor structured extraction lands in v7.8 with the corpus.

Behavioural contract:

• When the LLM emits an AIMessage whose tool_calls reference a tool name matching ANY procedural skill's "action tool" (metadata['action_tool'] if set, else metadata['steps'][0]), the gate reads that skill's metadata['precondition'].
• The precondition is treated as enforceable ONLY when it is a bare identifier string (e.g. "get_integration_status") or a list of bare identifiers. Free-text rules (e.g. "before delivery, call get_integration_status") fall through with NO enforcement -- the bare-identifier regex partitions enforced-vs-advisory adopters cleanly without forcing schema migration.
• For each bare identifier in the precondition, the gate scans the conversation message history for a prior ToolMessage with name (or tool_call_id-resolved tool name) matching the identifier.
• Any precondition tool NOT yet called this conversation produces a synthetic ToolMessage(content="Missing precondition: ...", tool_call_id=<the action's tool_call_id>, name=<action>, status= "error") injected into state. React sees the error and the model can recover by calling the precondition tool first.

create_precondition_gate_node

create_precondition_gate_node(get_active_skills: ActiveSkillsGetter) -> Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]

Build a LangGraph node that enforces structured preconditions.

Parameters:

Name	Type	Description	Default
get_active_skills	ActiveSkillsGetter	async callable that takes the current state and returns the list of procedural skill entries the gate should match against. Typical implementation queries ProceduralLayer.query_skills(scope, "", top_k=N, include_drafts=False). Returning [] makes the gate a no-op for that turn.	required

Returns:

Type	Description
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	The node function suitable for ``workflow.add_node(
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	"precondition_gate", node)``. The node reads the latest
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	AIMessage.tool_calls, runs the pure-function check, and
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	returns a state update injecting synthetic ToolMessage
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	errors when enforcement fires. Returns an empty dict when
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	every tool call passes (so LangGraph treats the node as a
Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]	no-op).

Source code in src/symfonic/core/nodes/precondition_gate.py

def create_precondition_gate_node(
    get_active_skills: ActiveSkillsGetter,
) -> Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]:
    """Build a LangGraph node that enforces structured preconditions.

    Args:
        get_active_skills: async callable that takes the current state
            and returns the list of procedural skill entries the gate
            should match against.  Typical implementation queries
            ``ProceduralLayer.query_skills(scope, "", top_k=N,
            include_drafts=False)``.  Returning ``[]`` makes the gate
            a no-op for that turn.

    Returns:
        The node function suitable for ``workflow.add_node(
        "precondition_gate", node)``.  The node reads the latest
        ``AIMessage.tool_calls``, runs the pure-function check, and
        returns a state update injecting synthetic ``ToolMessage``
        errors when enforcement fires.  Returns an empty dict when
        every tool call passes (so LangGraph treats the node as a
        no-op).
    """

    async def precondition_gate(state: dict[str, Any]) -> dict[str, Any]:
        from langchain_core.messages import ToolMessage

        messages = state.get("messages") or []
        tool_calls = _latest_ai_tool_calls(messages)
        if not tool_calls:
            return {}

        try:
            skills = await get_active_skills(state)
        except Exception:
            logger.debug(
                "precondition_gate: active-skill lookup failed; "
                "degrading to passthrough",
                exc_info=True,
            )
            return {}
        if not skills:
            return {}

        # v7.18.0 state-predicate filter (Option A).  Skills whose
        # ``metadata['precondition']`` carries any ``{"state": {...}}``
        # entries that do NOT match the current state are dropped
        # BEFORE the tool-call scan.  Off-platform skills never reach
        # the synthetic-error path -- the adopter per-user-memory
        # tenant-data-leak class is closed at the gate boundary.
        # Skills with no state predicates pass through unchanged
        # (preserves v7.7.4 behaviour byte-for-byte).
        skills = filter_skills_by_state_predicates(skills, state)
        if not skills:
            return {}

        called_tools = _called_tool_names_from_history(messages)
        errors = _check_tool_calls_against_skills(
            tool_calls, skills, called_tools,
        )
        if not errors:
            return {}

        new_messages = [
            ToolMessage(
                content=err["content"],
                tool_call_id=err["tool_call_id"],
                name=err["action_tool"],
                status="error",
            )
            for err in errors
            if isinstance(err.get("tool_call_id"), str)
        ]
        logger.info(
            "precondition_gate: enforced %d precondition(s); "
            "action_tools=%s",
            len(new_messages),
            sorted({err["action_tool"] for err in errors}),
        )
        return {"messages": new_messages}

    return precondition_gate

filter_skills_by_state_predicates

filter_skills_by_state_predicates(skills: list[Any], state: Any) -> list[Any]

Drop skills whose state predicates do not all match state.

Shared by both v7.18.0 evaluation sites (L1 PRE-FLIGHT synthesiser and reactive precondition_gate). Skills with no state predicates pass through untouched -- the v7.7.4 behaviour for tool-only preconditions is preserved byte-for-byte.

Conjunctive semantics: a skill must satisfy EVERY state predicate on its list to survive. Mirrors the existing tool-precondition semantic where every bare tool name must have been called.

Emits one DEBUG log line per dropped skill so adopters can debug "why isn't my Slack-only skill firing" without diffing the dispatched prompt.

Source code in src/symfonic/core/nodes/precondition_gate.py

def filter_skills_by_state_predicates(
    skills: list[Any],
    state: Any,
) -> list[Any]:
    """Drop skills whose state predicates do not all match ``state``.

    Shared by both v7.18.0 evaluation sites (L1 PRE-FLIGHT synthesiser
    and reactive ``precondition_gate``).  Skills with no state
    predicates pass through untouched -- the v7.7.4 behaviour for
    tool-only preconditions is preserved byte-for-byte.

    Conjunctive semantics: a skill must satisfy EVERY state predicate
    on its list to survive.  Mirrors the existing tool-precondition
    semantic where every bare tool name must have been called.

    Emits one DEBUG log line per dropped skill so adopters can debug
    "why isn't my Slack-only skill firing" without diffing the
    dispatched prompt.
    """
    if not skills:
        return []
    surviving: list[Any] = []
    for skill in skills:
        md = getattr(skill, "metadata", None) or {}
        precondition = md.get("precondition") if isinstance(md, dict) else None
        predicates = _resolve_precondition_state_predicates(precondition)
        if not predicates:
            surviving.append(skill)
            continue
        dropped = False
        for predicate in predicates:
            matched, reason = _evaluate_state_predicate(state, predicate)
            if not matched:
                label = _skill_label_for_log(skill)
                logger.debug(
                    "skill `%s` dropped this turn: state predicate %s",
                    label,
                    reason,
                )
                dropped = True
                break
        if not dropped:
            surviving.append(skill)
    return surviving

precondition_condition

precondition_condition(state: dict[str, Any]) -> bool

Cheap state-shape inspector used by the conditional edge router.

Returns True when the latest AIMessage has tool_calls AND state carries the _procedural_enforce_preconditions boolean flag set by the engine (which is itself set from FrameworkConfig.procedural_enforce_preconditions). The final enforcement decision happens inside the gate node when it actually queries the active skills -- this condition is a coarse pre-check.

The condition deliberately does NOT query the procedural layer itself (that would force an async call from a sync conditional-edge callback). When the precondition_gate node runs and finds no enforcement is needed, it returns an empty state update and the next conditional edge routes onward.

Source code in src/symfonic/core/nodes/precondition_gate.py

def precondition_condition(state: dict[str, Any]) -> bool:
    """Cheap state-shape inspector used by the conditional edge router.

    Returns ``True`` when the latest AIMessage has ``tool_calls`` AND
    state carries the ``_procedural_enforce_preconditions`` boolean
    flag set by the engine (which is itself set from
    ``FrameworkConfig.procedural_enforce_preconditions``).  The
    final enforcement decision happens inside the gate node when it
    actually queries the active skills -- this condition is a coarse
    pre-check.

    The condition deliberately does NOT query the procedural layer
    itself (that would force an async call from a sync conditional-edge
    callback).  When the precondition_gate node runs and finds no
    enforcement is needed, it returns an empty state update and the
    next conditional edge routes onward.
    """
    if not state.get("_procedural_enforce_preconditions"):
        return False
    messages = state.get("messages") or []
    return bool(_latest_ai_tool_calls(messages))