Rules#

Reference for the rule-authoring surface: build_application_rule, the Entity-DSL atom forms, vars, Not, the application Rule value, RuleExpr composition, and fg.rules.inspect with its RuleExprInspect result. Most of these symbols are importable from factgraph.sdk. The supporting types RuleValidationError, RuleOccurrence, and RulePortRef are not exported from factgraph.sdk; import them from factgraph.application.protocol (their defining module is factgraph.application.protocol.rule). PortType is importable only from factgraph.application.protocol.rule.

A Rule is an AND-only conjunction of conditions over existing facts, with named output ports. Rules are values: they compose with & and | into a RuleExpr, and are inspected with fg.rules.inspect.

`build_application_rule(*, id, when, ports, version=None, repr=None) -> Rule`#

Builds an application-layer Rule from Entity-DSL when atoms and named ports. Lowers the when body to the core where-AST, validates it in "python" mode with rule references disallowed, and returns a frozen Rule.

Parameters:

Parameter	Type	Default	Description
`id`	`str`	required	Stable rule id. Must be a non-empty string.
`when`	`list[Any]`	required	Non-empty list of Entity-DSL atoms (the conjunction body). OR branches (a list of lists) and `Case(...)` are rejected.
`ports`	`Mapping[str, Any]`	required	Non-empty mapping from port name to a named `LogicVar` produced by `vars(...)`. Each var must appear in `when`.
`version`	`str \| None`	`None`	Optional version string. When not `None`, must be non-empty.
`repr`	`str \| None`	`None`	Optional human-readable template. `%name` tokens interpolate port names; each referenced name must be a declared port.

Returns: Rule (the application protocol Rule, also exported as ApplicationRule).

Raises:

DSLToApplicationRuleError when the input cannot be lowered into a Rule. This includes: when is not a non-empty list; when contains a Case(...); when is shaped as OR branches (all items are lists); a legacy atom is present (Pred(...), a RuleRef atom, a bare AttrRef, or a bare-var comparison such as u.status == "active"); ports is not a non-empty Mapping; a port key is not a non-empty string; a port value is not an SDK LogicVar; a port uses an anonymous LogicVar from Entity(...) with Ellipsis; a port var does not appear in when; or the lowered body is not a pure conjunction (the bridge accepts AND-only when bodies).
DSLToApplicationRuleError also wraps the underlying where-AST parse and validation errors raised during lowering.
RuleValidationError from the Rule constructor when the constructed DTO violates a shape invariant (for example, a port var that does not appear in the lowered when).

Example:

from factgraph.sdk import Entity, Field, Identity, build_application_rule, vars


class User(Entity):
    user_id: str = Identity()
    status: str = Field()


with vars("u") as (u,):
    rule = build_application_rule(
        id="active_user",
        when=[User(u).status == "active"],
        ports={"user": u},
        repr="active user %user",
    )

rule.when         # (PredAtom("User:exists", ...), PredAtom("user:status", ...))
rule.render_repr()                 # "active user <user>"
rule.render_repr({"user": "u-1"})  # "active user u-1"

Entity-DSL atom forms#

The items in a when list are built from entity classes (subclasses of Entity) and logic variables. Each form lowers to one or more core where-AST atoms inside the resulting Rule.when. Comparisons are written with Python operators; Entity(var) and Entity(var).field are callable/attribute forms produced by the entity metaclass.

Existence: `Entity(var)`#

Asserts that an entity of that type exists, bound to var. Lowers to a PredAtom with pred_id "<EntityType>:exists" and the var as its single term.

with vars("u") as (u,):
    build_application_rule(id="any_user", when=[User(u)], ports={"user": u})
# rule.when -> (PredAtom("User:exists", [Var]),)

Anonymous existence: `Entity(...)`#

Entity(...) (with Ellipsis) creates an existence atom bound to a fresh anonymous LogicVar. An anonymous var cannot be used as a port: passing it in ports raises DSLToApplicationRuleError.

with vars("u") as (u,):
    build_application_rule(
        id="named_users_exist",
        when=[User(u), User(...).name == "alice"],
        ports={"user": u},
    )
# rule.when pred_ids -> ["User:exists", "User:exists", "user:name"]

Identity equality: `Entity(var).identity_field == literal`#

Binds an entity by an identity value. Emits the entity existence predicate plus a field predicate "<entitytype>:<field>" (entity type lowercased) with the var and the literal.

with vars("u") as (u,):
    build_application_rule(
        id="user_by_id",
        when=[User(u).user_id == "u-2"],
        ports={"user": u},
    )
# rule.when pred_ids -> ["User:exists", "user:user_id"]

Field equality to a literal: `Entity(var).field == literal`#

Emits the entity existence predicate plus a field predicate "<entitytype>:<field>" with the var and the literal as terms.

with vars("u") as (u,):
    build_application_rule(
        id="active_user",
        when=[User(u).status == "active"],
        ports={"user": u},
    )
# rule.when pred_ids -> ["User:exists", "user:status"]

Field equality to a named var: `Entity(var).field == other_var`#

Binds the field value to a logic variable, which may be exposed as a port.

with vars("li", "country") as (li, country):
    build_application_rule(
        id="lives_in_country",
        when=[LivesIn(li).country == country],
        ports={"edge": li, "country": country},
    )
# rule.when pred_ids -> ["LivesIn:exists", "livesin:country"]

Cross-entity reference: `EntityA(a).field == EntityB(b)`#

Relates two entities through a reference field. Emits both existence predicates and the reference field predicate with both vars as terms.

with vars("li", "u") as (li, u):
    build_application_rule(
        id="lives_in_user",
        when=[LivesIn(li).user == User(u)],
        ports={"edge": li, "user": u},
    )
# rule.when pred_ids -> ["LivesIn:exists", "User:exists", "livesin:user"]

Ordering comparisons on logic variables#

The comparison operators ==, !=, >, >=, <, <= on a LogicVar produce a comparison. ==, >, >=, <, <= lower to a CmpAtom with the corresponding op (eq, gt, ge, lt, le); != lowers to a negated equality (NotAtom over an equality). Entity attribute comparison sugar (Entity(var).field <op> value) supports only ==; other operators on an attribute raise DSLToApplicationRuleError during lowering.

Linear arithmetic on logic variables#

+, -, and constant * on a LogicVar build an arithmetic expression that lowers to a BuiltinAtom (ops include add, sub, addc, mulc) feeding a temporary var, followed by a comparison. Non-linear multiplication of two variables (x * y) is not supported and raises DSLToApplicationRuleError.

with vars("u") as (u,):
    build_application_rule(id="user_plus_one", when=[(u + 1) == 3], ports={"user": u})
# rule.when atom types -> [BuiltinAtom, CmpAtom]

Aggregates#

agg_count(*, where), agg_sum(target, *, where), agg_min(target, *, where), agg_max(target, *, where), and agg_mean(target, *, where) build a correlated aggregate reference, compared against a result var. The aggregate lowers into an AggregateAtom carried by the comparison CmpAtom. where must be a non-empty list. agg_count takes no target. The target must be an Entity(var).field form, not a bare var.field; a bare attribute target or an unbound target var raises DSLToApplicationRuleError. A legacy Pred(...) inside an aggregate filter is rejected.

with vars("u", "o", "total") as (u, o, total):
    build_application_rule(
        id="user_order_total",
        when=[
            User(u),
            total == agg_sum(Order(o).amount, where=[Order(o).buyer == u]),
            total > 4,
        ],
        ports={"user": u, "total": total},
    )
# rule.when[1] -> CmpAtom whose rhs is an AggregateAtom(kind="sum")

`vars(*names) -> _VarsContext`#

Creates named logic variables for rule bodies. Used as a context manager.

Parameters:

*names (str): variable names. Each must be a non-empty string; duplicates are rejected. Names beginning with __ are reserved.

Behavior:

with vars("u", "tag") as (u, tag): yields a tuple of LogicVar values, one per name.
with vars() as V: (no names) yields a factory; call V("u", "tag") inside the block to build vars. Calling the factory with no names raises SDKDSLError.
Iterating the no-name form directly (with vars() as (...)) raises SDKDSLError.

Raises: SDKDSLError for empty/duplicate names, the unsupported iteration form, or a factory call with no names.

Example:

from factgraph.sdk import vars

with vars("u", "tag") as (u, tag):
    ...  # u and tag are LogicVar values

with vars() as V:
    u, tag = V("u", "tag")

`Not(body) -> NotExpr`#

Negates a body fragment inside a when clause, for absence or anti-join checks. Lowers to a core NotAtom. The wrapped body may correlate with variables bound by earlier atoms in the conjunction.

Parameters:

body (list[Any]): a non-empty list of Entity-DSL atoms.

Returns: NotExpr.

Raises: SDKDSLError when body is not a non-empty list.

from factgraph.sdk import Not, vars

with vars("u") as (u,):
    build_application_rule(
        id="user_without_status",
        when=[User(u), Not([User(u).status == "banned"])],
        ports={"user": u},
    )

`Rule`#

Application protocol Rule DTO. Returned by build_application_rule. Exported from factgraph.sdk as both Rule and ApplicationRule. Frozen dataclass. (This is distinct from the legacy factgraph.sdk.dsl.Rule; see Legacy below.)

Fields:

Field	Type	Default	Description
`id`	`str`	required	Rule id. Must be non-empty.
`when`	`tuple[Atom, ...]`	required	Non-empty tuple of core where-AST atoms (`PredAtom`, `CmpAtom`, `InAtom`, `BuiltinAtom`, `NotAtom`).
`ports`	`Mapping[str, Var]`	required	Non-empty mapping from port name to a core `Var`; each var must appear in `when`. Stored read-only.
`version`	`str \| None`	`None`	Optional version; when not `None`, must be non-empty.
`repr`	`str \| None`	`None`	Optional `%name` interpolation template; when not `None`, must be non-empty and reference only declared ports.

Raises: RuleValidationError for any violated invariant (empty id/version/repr, empty when, empty ports, a non-Var port value, a port var absent from when, a disallowed atom type such as RuleRefAtom, or a repr referencing an undeclared port or containing malformed %).

Properties:

atom_ids -> tuple[str, ...]: ids "<id>:atom_<idx>" for each atom in when.
port_types -> Mapping[str, PortType]: per-port PortType(kind, entity_type), where kind is "entity_ref" (with entity_type) when the port var is bound by an "<Entity>:exists" predicate, otherwise "value". PortType is importable only from factgraph.application.protocol.rule.
content_digest -> str: SHA-256 hex over the canonical serialization of ports and when.

Methods:

render_repr(bindings=None) -> str: render the repr template. Unbound %name tokens render as <name>. Returns "" when repr is None. bindings must be a Mapping[str, Any] or None.
as_(alias=None) -> RuleOccurrence: create an occurrence of this rule. alias defaults to the rule id; it must match [A-Za-z][A-Za-z0-9_]*.
Rule.projection(*port_names) -> Rule (classmethod): build a projection rule exposing the given port names. Requires at least one name; names must be non-empty and unique.

Operators:

rule & other and rule | other return a RuleExpr (AND / OR combination). See RuleExpr.
bool(rule) raises ExplicitBoolError: rule values do not support Python truthiness; use & or | instead of and/or.

Occurrence access:

A RuleOccurrence (from rule.as_(alias)) exposes ports as attributes or via occurrence.port(name), each returning a RulePortRef. Accessing an undeclared port name raises RuleValidationError (or AttributeError via attribute access). A RulePortRef provides .eq(other) -> RuleJoinConstraint to build a join across two distinct occurrences. RuleOccurrence and RulePortRef are importable from factgraph.application.protocol (defining module factgraph.application.protocol.rule), not from factgraph.sdk.

occurrence = rule.as_("a")
occurrence.alias            # "a"
occurrence.user.port_name   # "user"

`RuleExpr`#

Public namespace for combining rules into AND/OR expressions. Rule & Rule, Rule | Rule, and the factories below all return an internal RuleExpr value (frozen, hashable, equality-comparable by canonical form). RuleExpr values are passed to fg.eval.evaluate(...) and fg.rules.inspect(...).

Factories:

RuleExpr.all(*operands) -> RuleExpr: conjunction (same as chaining &). Requires at least one operand.
RuleExpr.any(*operands) -> RuleExpr: disjunction (same as chaining |). Requires at least one operand.

Operands are Rule, RuleOccurrence, or RuleExpr values. Same-kind groups flatten and are commutative: (a & b) & c == a & (b & c) and a & b == b & a.

Operators:

& / __and__, | / __or__ combine operands.
bool(expr) raises ExplicitBoolError.

Aliasing and occurrence rules:

A bare Rule enters an expression with its id as the default alias, which must be identifier-shaped ([A-Za-z][A-Za-z0-9_]*); otherwise use rule.as_(alias). A rule appearing more than once must use explicit .as_(...) aliases on every occurrence. Duplicate aliases across operands are rejected.

Raises: RuleExprError for: empty factory calls; non-Rule/non-RuleExpr operands; a bare rule id that is not identifier-shaped; duplicate aliases; a rule used multiple times without explicit aliases; and (see below) malformed joins. Passing a legacy factgraph.sdk.dsl.Rule raises RuleExprError instructing use of build_application_rule(...).

Example:

expr = active_user.as_("a") & lives_in_country.as_("b")
also  = RuleExpr.all(active_user.as_("a"), lives_in_country.as_("b"))
either = active_user.as_("a") | lives_in_country.as_("b")

Joins#

Joins constrain ports across distinct occurrences and attach only to AND groups (_AndGroup).

and_group.join(*constraints) -> _AndGroup: attach RuleJoinConstraint values (built with portref.eq(other_portref)). Requires at least one constraint. Returns a new immutable group; duplicate and symmetric constraints are normalized.
and_group.join_by_ports(*port_names) -> _AndGroup: pairwise-join all direct occurrences that declare each named port. Requires at least one name; names must be non-empty and unique. A name absent from every occurrence, or present on fewer than two occurrences, raises RuleExprError.

Calling join or join_by_ports on an OR group raises RuleExprError (joins must be attached to AND groups). A constraint connecting the same occurrence to itself raises RuleExprError (put self-constraints in Rule.when). A join endpoint not reachable from the AND spine, or whose var/port type does not match the rule port, raises RuleExprError.

RuleJoinConstraint:

Frozen value with fields left: RulePortRef, right: RulePortRef, op: Literal["eq"] = "eq". Both endpoints must be RulePortRef; op must be "eq". Equality and hashing are canonical (order-independent across left/right).

a = active_user.as_("a")
b = lives_in_country.as_("b")

joined = (a & b).join(a.user.eq(b.user))
same   = (a & b).join_by_ports("user")     # equivalent when both declare "user"

`fg.rules.inspect(obj) -> RuleExprInspect`#

Read-only structural inspection of a rule or rule expression, without executing it. fg.rules is read-only (assigning attributes raises FrozenSnapshotError).

Parameters:

obj: an application Rule or a RuleExpr value. For a Rule, the schema index bound to the graph is used to compute closed-head information.

Returns: RuleExprInspect.

Raises: SDKStoreError (message rules.inspect(...) expects SDK Rule or Inference) when the argument is neither an application Rule nor a RuleExpr value.

inspect = fg.rules.inspect(rule)
inspect.templates              # tuple of template (rule) ids
inspect.render_compact()       # compact AST string

`RuleExprInspect`#

Frozen result of fg.rules.inspect.

Fields and properties:

Member	Type	Description
`ast`	`tuple[object, ...]`	Nested AST tuple: `("rule", alias, rule_id)`, `("and", children, joins)`, or `("or", children)`.
`occurrences`	`tuple[OccurrenceInspect, ...]`	One entry per rule occurrence, sorted by `(alias, template_id)`.
`joins`	`tuple[RuleJoinConstraint, ...]`	Join constraints across the expression, sorted.
`unjoined_same_name_ports`	`tuple[dict[str, object], ...]`	Hints (`{"port_name", "occurrences"}`) for same-named ports on sibling AND occurrences that are not fully joined.
`ports` (property)	`tuple[PortInspect, ...]`	Distinct port descriptors across all occurrences.
`is_closed`	`bool`	Whether every port is closed (bound to a literal, or, for entity refs, by all identity predicates).
`unbound_ports`	`tuple[str, ...]`	Port names that are not closed.
`templates` (property)	`tuple[str, ...]`	Distinct template (rule) ids, in occurrence order.
`port_visibility` (property)	`Mapping[str, tuple[str, ...]]`	Per-occurrence-alias mapping to the occurrence’s port names.

Methods:

render(bindings=None) -> str: human-readable rendering combining the compact AST, occurrence labels (with %name repr interpolation, resolving alias.port then port), and joins. bindings must be a Mapping[str, object] or None.
render_compact() -> str: compact AST string (for example (a:active_user & b:lives_in_country), with .join(N) appended when an AND group carries N joins).

`OccurrenceInspect`#

Frozen per-occurrence record: template_id: str, alias: str, repr_template: str | None, ports: tuple[str, ...] (sorted port names), atoms: tuple[ConditionDescriptor, ...] (one per when atom).

`ConditionDescriptor`#

Frozen per-atom record. Fields: atom_id: str, kind: str, subject: str | None = None, entity_type: str | None = None, field: str | None = None, op: str | None = None, value: object = None, summary: str = "". kind is one of entity_existence, field_predicate, pred, cmp, in, builtin, not, derived from the atom type.

`PortInspect`#

Frozen port descriptor. Fields: name: str, kind: Literal["entity_ref", "value"], entity_type: str | None = None, field: str | None = None, value_type: str | None = None. entity_type is required when kind == "entity_ref".

Errors#

Error	Module	Raised when
`DSLToApplicationRuleError`	`factgraph.sdk` (defined in `...sdk.dsl`)	`build_application_rule` input cannot be lowered into a `Rule`. Subclass of `SDKDSLError`.
`RuleValidationError`	`factgraph.application.protocol` (defined in `...protocol.rule`); not exported from `factgraph.sdk`	An application `Rule` (or `RuleOccurrence`) violates a shape invariant. Subclass of `ValueError`.
`RuleExprError`	`factgraph.sdk` (defined in `...protocol.rule_expr`)	`RuleExpr` authoring or join input violates the expression contract. Subclass of `SDKDSLError`.
`ExplicitBoolError`	`factgraph.sdk` (defined in `...protocol.rule_expr`)	`bool(...)` is called on a `Rule` or `RuleExpr` value. Subclass of `RuleExprError`.
`SDKDSLError`	`factgraph.sdk`	`vars(...)`, `Not(...)`, and other DSL builders receive invalid input.
`SDKStoreError`	`factgraph.sdk`	`fg.rules.inspect(...)` receives an argument that is neither an application `Rule` nor a `RuleExpr`.

See errors.md for the full error catalogue.

Legacy#

The legacy DSL types Inference, Query, Pred, EmitSpec, RuleRef, and factgraph.sdk.dsl.Rule (the select/where rule) remain importable for backward compatibility but are not part of the documented rule-authoring path and are not covered here.

Rules#

build_application_rule(*, id, when, ports, version=None, repr=None) -> Rule#

Entity-DSL atom forms#

Existence: Entity(var)#

Anonymous existence: Entity(...)#

Identity equality: Entity(var).identity_field == literal#

Field equality to a literal: Entity(var).field == literal#

Field equality to a named var: Entity(var).field == other_var#

Cross-entity reference: EntityA(a).field == EntityB(b)#

Ordering comparisons on logic variables#

Linear arithmetic on logic variables#

Aggregates#

vars(*names) -> _VarsContext#

Not(body) -> NotExpr#

Rule#

RuleExpr#

Joins#

fg.rules.inspect(obj) -> RuleExprInspect#

RuleExprInspect#

OccurrenceInspect#

ConditionDescriptor#

PortInspect#

Errors#

Legacy#

See also#

`build_application_rule(*, id, when, ports, version=None, repr=None) -> Rule`#

Existence: `Entity(var)`#

Anonymous existence: `Entity(...)`#

Identity equality: `Entity(var).identity_field == literal`#

Field equality to a literal: `Entity(var).field == literal`#

Field equality to a named var: `Entity(var).field == other_var`#

Cross-entity reference: `EntityA(a).field == EntityB(b)`#

`vars(*names) -> _VarsContext`#

`Not(body) -> NotExpr`#

`Rule`#

`RuleExpr`#

`fg.rules.inspect(obj) -> RuleExprInspect`#

`RuleExprInspect`#

`OccurrenceInspect`#

`ConditionDescriptor`#

`PortInspect`#