crv.core.grammar

Experimental API

Grammar synchronization and EBNF may change as expressivity is extended. Since: 2025-09-25.

crv.core.grammar

Canonical grammar and normalization helpers.

Defines enums (actions, channels, visibility, patch operations, representation edge kinds, exchange kinds) and helpers to normalize/validate enum-like strings. Includes the authoritative lower_snake EBNF. Zero‑IO; source of truth for naming.

See Also

schema.md, versioning.md

crv.core.grammar.EBNF_GRAMMAR module-attribute

EBNF_GRAMMAR: typing.Final[str] = crv.core.grammar._load_ebnf_text()

crv.core.grammar.PARSED_GRAMMAR module-attribute

PARSED_GRAMMAR: typing.Final[crv.core.grammar.ParsedGrammar] = (
    crv.core.grammar.ParsedGrammar.from_text(crv.core.grammar.EBNF_GRAMMAR)
)

crv.core.grammar.GrammarProduction dataclass

Parsed production with convenient accessors.

Source code in src/crv/core/grammar.py

@dataclass(slots=True, frozen=True)
class GrammarProduction:
    """Parsed production with convenient accessors."""

    name: str
    expression: str
    alternatives: tuple[str, ...]
    leading_terminals: tuple[str, ...]

    def lower_snake_terminals(self) -> tuple[str, ...]:
        return tuple(token for token in self.leading_terminals if is_lower_snake(token))

crv.core.grammar.GrammarProduction.name instance-attribute

name: str

crv.core.grammar.GrammarProduction.expression instance-attribute

expression: str

crv.core.grammar.GrammarProduction.alternatives instance-attribute

alternatives: tuple[str, ...]

crv.core.grammar.GrammarProduction.leading_terminals instance-attribute

leading_terminals: tuple[str, ...]

crv.core.grammar.GrammarProduction.lower_snake_terminals

lower_snake_terminals() -> tuple[str, ...]

Source code in src/crv/core/grammar.py

def lower_snake_terminals(self) -> tuple[str, ...]:
    return tuple(token for token in self.leading_terminals if is_lower_snake(token))

crv.core.grammar.ParsedGrammar dataclass

Container for parsed grammar productions.

Source code in src/crv/core/grammar.py

@dataclass(slots=True, frozen=True)
class ParsedGrammar:
    """Container for parsed grammar productions."""

    productions: dict[str, GrammarProduction]

    def production(self, name: str) -> GrammarProduction:
        try:
            return self.productions[name]
        except KeyError as exc:  # pragma: no cover - defensive
            raise KeyError(f"Unknown grammar production: {name}") from exc

    def lower_snake_terminals(self, name: str) -> tuple[str, ...]:
        return self.production(name).lower_snake_terminals()

    @classmethod
    def from_text(cls, text: str) -> ParsedGrammar:
        stripped = _strip_ebnf_comments(text)
        productions: dict[str, GrammarProduction] = {}
        for match in _RULE_RE.finditer(stripped):
            rule_name = match.group(1)
            expression = match.group(2).strip()
            alternatives = _split_alternatives(expression)
            leading = tuple(
                literal
                for literal in (_first_literal(part) for part in alternatives)
                if literal is not None
            )
            productions[rule_name] = GrammarProduction(
                name=rule_name,
                expression=expression,
                alternatives=alternatives,
                leading_terminals=_dedupe_preserving_order(leading),
            )
        return cls(productions=productions)

crv.core.grammar.ParsedGrammar.productions instance-attribute

productions: dict[str, crv.core.grammar.GrammarProduction]

crv.core.grammar.ParsedGrammar.production

production(name: str) -> GrammarProduction

Source code in src/crv/core/grammar.py

def production(self, name: str) -> GrammarProduction:
    try:
        return self.productions[name]
    except KeyError as exc:  # pragma: no cover - defensive
        raise KeyError(f"Unknown grammar production: {name}") from exc

crv.core.grammar.ParsedGrammar.lower_snake_terminals

lower_snake_terminals(name: str) -> tuple[str, ...]

Source code in src/crv/core/grammar.py

def lower_snake_terminals(self, name: str) -> tuple[str, ...]:
    return self.production(name).lower_snake_terminals()

crv.core.grammar.ParsedGrammar.from_text classmethod

from_text(text: str) -> ParsedGrammar

Source code in src/crv/core/grammar.py

@classmethod
def from_text(cls, text: str) -> ParsedGrammar:
    stripped = _strip_ebnf_comments(text)
    productions: dict[str, GrammarProduction] = {}
    for match in _RULE_RE.finditer(stripped):
        rule_name = match.group(1)
        expression = match.group(2).strip()
        alternatives = _split_alternatives(expression)
        leading = tuple(
            literal
            for literal in (_first_literal(part) for part in alternatives)
            if literal is not None
        )
        productions[rule_name] = GrammarProduction(
            name=rule_name,
            expression=expression,
            alternatives=alternatives,
            leading_terminals=_dedupe_preserving_order(leading),
        )
    return cls(productions=productions)

crv.core.grammar.ActionKind

Bases: enum.Enum

All concrete action verbs an agent (or venue) may emit.

Serialized values are used in

• DecisionHead.action_candidates[].action_type (schema.DecisionHead)
• Event envelopes payloads written by world
• EBNF terminals

Notes

Table mappings: * messages (send_chat_message, publish_announcement) * exchange (post/cancel/order, trade, swap, peer exchange, vote, gift) * scenarios_seen (as part of context reconstruction) * decisions (chosen_action and action_candidates)

Source code in src/crv/core/grammar.py

class ActionKind(Enum):
    """
    All concrete action verbs an agent (or venue) may emit.

    Serialized values are used in:
      - DecisionHead.action_candidates[].action_type (schema.DecisionHead)
      - Event envelopes payloads written by world
      - EBNF terminals

    Notes:
      Table mappings:
        * messages           (send_chat_message, publish_announcement)
        * exchange           (post/cancel/order, trade, swap, peer exchange, vote, gift)
        * scenarios_seen     (as part of context reconstruction)
        * decisions          (chosen_action and action_candidates)
    """

    ACQUIRE_TOKEN = "acquire_token"
    RELINQUISH_TOKEN = "relinquish_token"
    RELATE_AGENT = "relate_agent"
    RELATE_OTHER_AGENTS = "relate_other_agents"
    ENDORSE_TOKEN = "endorse_token"
    COENDORSE_TOKEN_WITH_AGENTS = "coendorse_token_with_agents"
    EXPOSE_SIGNAL_ABOUT_TOKEN = "expose_signal_about_token"
    DECLARE_COOCCURRENCE = "declare_cooccurrence_between_tokens"

    PROPOSE_PEER_EXCHANGE = "propose_peer_exchange"
    ACCEPT_PEER_EXCHANGE = "accept_peer_exchange"
    REJECT_PEER_EXCHANGE = "reject_peer_exchange"
    SETTLE_PEER_EXCHANGE = "settle_peer_exchange"

    POST_ORDER_TO_VENUE = "post_order_to_venue"
    CANCEL_ORDER_AT_VENUE = "cancel_order_at_venue"
    SETTLE_TRADE_FROM_VENUE = "settle_trade_from_venue"
    SWAP_AT_VENUE = "swap_at_venue"

    CAST_VOTE_AT_VENUE = "cast_vote_at_venue"
    PUBLISH_VOTE_OUTCOME = "publish_vote_outcome_from_venue"
    GIFT_TOKEN = "gift_token"

    SEND_CHAT_MESSAGE = "send_chat_message"
    PUBLISH_ANNOUNCEMENT = "publish_announcement"

crv.core.grammar.ActionKind.ACQUIRE_TOKEN class-attribute instance-attribute

ACQUIRE_TOKEN = 'acquire_token'

crv.core.grammar.ActionKind.RELINQUISH_TOKEN class-attribute instance-attribute

RELINQUISH_TOKEN = 'relinquish_token'

crv.core.grammar.ActionKind.RELATE_AGENT class-attribute instance-attribute

RELATE_AGENT = 'relate_agent'

crv.core.grammar.ActionKind.RELATE_OTHER_AGENTS class-attribute instance-attribute

RELATE_OTHER_AGENTS = 'relate_other_agents'

crv.core.grammar.ActionKind.ENDORSE_TOKEN class-attribute instance-attribute

ENDORSE_TOKEN = 'endorse_token'

crv.core.grammar.ActionKind.COENDORSE_TOKEN_WITH_AGENTS class-attribute instance-attribute

COENDORSE_TOKEN_WITH_AGENTS = 'coendorse_token_with_agents'

crv.core.grammar.ActionKind.EXPOSE_SIGNAL_ABOUT_TOKEN class-attribute instance-attribute

EXPOSE_SIGNAL_ABOUT_TOKEN = 'expose_signal_about_token'

crv.core.grammar.ActionKind.DECLARE_COOCCURRENCE class-attribute instance-attribute

DECLARE_COOCCURRENCE = 'declare_cooccurrence_between_tokens'

crv.core.grammar.ActionKind.PROPOSE_PEER_EXCHANGE class-attribute instance-attribute

PROPOSE_PEER_EXCHANGE = 'propose_peer_exchange'

crv.core.grammar.ActionKind.ACCEPT_PEER_EXCHANGE class-attribute instance-attribute

ACCEPT_PEER_EXCHANGE = 'accept_peer_exchange'

crv.core.grammar.ActionKind.REJECT_PEER_EXCHANGE class-attribute instance-attribute

REJECT_PEER_EXCHANGE = 'reject_peer_exchange'

crv.core.grammar.ActionKind.SETTLE_PEER_EXCHANGE class-attribute instance-attribute

SETTLE_PEER_EXCHANGE = 'settle_peer_exchange'

crv.core.grammar.ActionKind.POST_ORDER_TO_VENUE class-attribute instance-attribute

POST_ORDER_TO_VENUE = 'post_order_to_venue'

crv.core.grammar.ActionKind.CANCEL_ORDER_AT_VENUE class-attribute instance-attribute

CANCEL_ORDER_AT_VENUE = 'cancel_order_at_venue'

crv.core.grammar.ActionKind.SETTLE_TRADE_FROM_VENUE class-attribute instance-attribute

SETTLE_TRADE_FROM_VENUE = 'settle_trade_from_venue'

crv.core.grammar.ActionKind.SWAP_AT_VENUE class-attribute instance-attribute

SWAP_AT_VENUE = 'swap_at_venue'

crv.core.grammar.ActionKind.CAST_VOTE_AT_VENUE class-attribute instance-attribute

CAST_VOTE_AT_VENUE = 'cast_vote_at_venue'

crv.core.grammar.ActionKind.PUBLISH_VOTE_OUTCOME class-attribute instance-attribute

PUBLISH_VOTE_OUTCOME = 'publish_vote_outcome_from_venue'

crv.core.grammar.ActionKind.GIFT_TOKEN class-attribute instance-attribute

GIFT_TOKEN = 'gift_token'

crv.core.grammar.ActionKind.SEND_CHAT_MESSAGE class-attribute instance-attribute

SEND_CHAT_MESSAGE = 'send_chat_message'

crv.core.grammar.ActionKind.PUBLISH_ANNOUNCEMENT class-attribute instance-attribute

PUBLISH_ANNOUNCEMENT = 'publish_announcement'

crv.core.grammar.ChannelType

Bases: enum.Enum

Logical channel families for routing & visibility.

Serialized values appear in

• messages.channel_name (prefix like 'group:G1', 'room:market')
• event envelopes channel field

Source code in src/crv/core/grammar.py

class ChannelType(Enum):
    """
    Logical channel families for routing & visibility.

    Serialized values appear in:
      - messages.channel_name (prefix like 'group:G1', 'room:market')
      - event envelopes channel field
    """

    PUBLIC = "public"
    GROUP = "group"
    ROOM = "room"
    DM = "dm"

crv.core.grammar.ChannelType.PUBLIC class-attribute instance-attribute

PUBLIC = 'public'

crv.core.grammar.ChannelType.GROUP class-attribute instance-attribute

GROUP = 'group'

crv.core.grammar.ChannelType.ROOM class-attribute instance-attribute

ROOM = 'room'

crv.core.grammar.ChannelType.DM class-attribute instance-attribute

DM = 'dm'

crv.core.grammar.Visibility

Bases: enum.Enum

Rendered visibility on observations or messages.

Serialized values appear in

• messages.visibility_scope
• event envelopes visibility field
• scenarios_seen.visibility_scope

Source code in src/crv/core/grammar.py

class Visibility(Enum):
    """
    Rendered visibility on observations or messages.

    Serialized values appear in:
      - messages.visibility_scope
      - event envelopes visibility field
      - scenarios_seen.visibility_scope
    """

    PUBLIC = "public"
    GROUP = "group"
    ROOM = "room"
    DM = "dm"

crv.core.grammar.Visibility.PUBLIC class-attribute instance-attribute

PUBLIC = 'public'

crv.core.grammar.Visibility.GROUP class-attribute instance-attribute

GROUP = 'group'

crv.core.grammar.Visibility.ROOM class-attribute instance-attribute

ROOM = 'room'

crv.core.grammar.Visibility.DM class-attribute instance-attribute

DM = 'dm'

crv.core.grammar.PatchOp

Bases: enum.Enum

Atomic edit operations over the identity/affect representation.

Canonical-only policy

Only the following operations are valid and accepted: - set_identity_edge_weight - adjust_identity_edge_weight - decay_identity_edges - remove_identity_edge

Notes

Use the canonical operations together with explicit edge_kind and fields, e.g., object_to_object with subject_id/object_id, or object_to_positive_valence / object_to_negative_valence with token_id.

Table mappings: * identity_edges (rows produced AFTER patches are applied) — we log the post-barrier state, not the patch instruction itself.

Source code in src/crv/core/grammar.py

class PatchOp(Enum):
    """
    Atomic edit operations over the identity/affect representation.

    Canonical-only policy:
      Only the following operations are valid and accepted:
        - set_identity_edge_weight
        - adjust_identity_edge_weight
        - decay_identity_edges
        - remove_identity_edge

    Notes:
      Use the canonical operations together with explicit edge_kind and fields, e.g.,
      object_to_object with subject_id/object_id, or object_to_positive_valence /
      object_to_negative_valence with token_id.

      Table mappings:
        * identity_edges (rows produced AFTER patches are applied) — we log the
          post-barrier state, not the patch instruction itself.
    """

    SET_IDENTITY_EDGE_WEIGHT = "set_identity_edge_weight"
    ADJUST_IDENTITY_EDGE_WEIGHT = "adjust_identity_edge_weight"
    DECAY_IDENTITY_EDGES = "decay_identity_edges"
    REMOVE_IDENTITY_EDGE = "remove_identity_edge"

crv.core.grammar.PatchOp.SET_IDENTITY_EDGE_WEIGHT class-attribute instance-attribute

SET_IDENTITY_EDGE_WEIGHT = 'set_identity_edge_weight'

crv.core.grammar.PatchOp.ADJUST_IDENTITY_EDGE_WEIGHT class-attribute instance-attribute

ADJUST_IDENTITY_EDGE_WEIGHT = 'adjust_identity_edge_weight'

crv.core.grammar.PatchOp.DECAY_IDENTITY_EDGES class-attribute instance-attribute

DECAY_IDENTITY_EDGES = 'decay_identity_edges'

crv.core.grammar.PatchOp.REMOVE_IDENTITY_EDGE class-attribute instance-attribute

REMOVE_IDENTITY_EDGE = 'remove_identity_edge'

crv.core.grammar.RepresentationEdgeKind

Bases: enum.Enum

Identity/affect edge kinds INSIDE an agent's representation.

Use these values in

• identity_edges.edge_kind

Notes

Table mappings: * identity_edges: one row per edge snapshot/delta written AFTER the barrier applies representation patches.

Psych/math mapping: These correspond to the following psychological constructs and math symbols. Valence is modeled as dual implicit channels per object (r^+{i,o}, r^-{i,o}) and dual "friend/foe" channels per other agent (u^+{i,j}, u^-{i,j}) from self i's perspective. a_{i,j} is an additional primitive self–other tie; anchors s_i^+, s_i^- are slow self-evaluations.

- self_to_positive_valence      -> s^+_{i}            (self anchor)
- self_to_negative_valence      -> s^-_{i}            (self anchor)
- self_to_object                -> s_{i,o}
- self_to_agent                 -> a_{i,j}            (primitive self–other tie)
- agent_to_positive_valence     -> u^+_{i,j}          (self's positive feeling toward agent j)
- agent_to_negative_valence     -> u^-_{i,j}          (self's negative feeling toward agent j)
- agent_to_object               -> b_{i,j,o}
- agent_to_agent                -> d_{i,k,l}
- agent_pair_to_object          -> q_{i,k,l,o}
- object_to_positive_valence    -> r^+_{i,o}
- object_to_negative_valence    -> r^-_{i,o}
- object_to_object              -> c_{i,o,o'}

Source code in src/crv/core/grammar.py

class RepresentationEdgeKind(Enum):
    """
    Identity/affect edge kinds INSIDE an agent's representation.

    Use these values in:
      - identity_edges.edge_kind

    Notes:
      Table mappings:
        * identity_edges: one row per edge snapshot/delta written AFTER the
          barrier applies representation patches.

      Psych/math mapping:
        These correspond to the following psychological constructs and math symbols.
        Valence is modeled as dual implicit channels per object (r^+_{i,o}, r^-_{i,o}) and dual "friend/foe" channels per other agent (u^+_{i,j}, u^-_{i,j}) from self i's perspective. a_{i,j} is an additional primitive self–other tie; anchors s_i^+, s_i^- are slow self-evaluations.

          - self_to_positive_valence      -> s^+_{i}            (self anchor)
          - self_to_negative_valence      -> s^-_{i}            (self anchor)
          - self_to_object                -> s_{i,o}
          - self_to_agent                 -> a_{i,j}            (primitive self–other tie)
          - agent_to_positive_valence     -> u^+_{i,j}          (self's positive feeling toward agent j)
          - agent_to_negative_valence     -> u^-_{i,j}          (self's negative feeling toward agent j)
          - agent_to_object               -> b_{i,j,o}
          - agent_to_agent                -> d_{i,k,l}
          - agent_pair_to_object          -> q_{i,k,l,o}
          - object_to_positive_valence    -> r^+_{i,o}
          - object_to_negative_valence    -> r^-_{i,o}
          - object_to_object              -> c_{i,o,o'}
    """

    SELF_TO_POSITIVE_VALENCE = "self_to_positive_valence"
    SELF_TO_NEGATIVE_VALENCE = "self_to_negative_valence"
    SELF_TO_OBJECT = "self_to_object"
    SELF_TO_AGENT = "self_to_agent"
    AGENT_TO_POSITIVE_VALENCE = "agent_to_positive_valence"
    AGENT_TO_NEGATIVE_VALENCE = "agent_to_negative_valence"
    AGENT_TO_OBJECT = "agent_to_object"
    AGENT_TO_AGENT = "agent_to_agent"
    AGENT_PAIR_TO_OBJECT = "agent_pair_to_object"
    OBJECT_TO_POSITIVE_VALENCE = "object_to_positive_valence"
    OBJECT_TO_NEGATIVE_VALENCE = "object_to_negative_valence"
    OBJECT_TO_OBJECT = "object_to_object"

crv.core.grammar.RepresentationEdgeKind.SELF_TO_POSITIVE_VALENCE class-attribute instance-attribute

SELF_TO_POSITIVE_VALENCE = 'self_to_positive_valence'

crv.core.grammar.RepresentationEdgeKind.SELF_TO_NEGATIVE_VALENCE class-attribute instance-attribute

SELF_TO_NEGATIVE_VALENCE = 'self_to_negative_valence'

crv.core.grammar.RepresentationEdgeKind.SELF_TO_OBJECT class-attribute instance-attribute

SELF_TO_OBJECT = 'self_to_object'

crv.core.grammar.RepresentationEdgeKind.SELF_TO_AGENT class-attribute instance-attribute

SELF_TO_AGENT = 'self_to_agent'

crv.core.grammar.RepresentationEdgeKind.AGENT_TO_POSITIVE_VALENCE class-attribute instance-attribute

AGENT_TO_POSITIVE_VALENCE = 'agent_to_positive_valence'

crv.core.grammar.RepresentationEdgeKind.AGENT_TO_NEGATIVE_VALENCE class-attribute instance-attribute

AGENT_TO_NEGATIVE_VALENCE = 'agent_to_negative_valence'

crv.core.grammar.RepresentationEdgeKind.AGENT_TO_OBJECT class-attribute instance-attribute

AGENT_TO_OBJECT = 'agent_to_object'

crv.core.grammar.RepresentationEdgeKind.AGENT_TO_AGENT class-attribute instance-attribute

AGENT_TO_AGENT = 'agent_to_agent'

crv.core.grammar.RepresentationEdgeKind.AGENT_PAIR_TO_OBJECT class-attribute instance-attribute

AGENT_PAIR_TO_OBJECT = 'agent_pair_to_object'

crv.core.grammar.RepresentationEdgeKind.OBJECT_TO_POSITIVE_VALENCE class-attribute instance-attribute

OBJECT_TO_POSITIVE_VALENCE = 'object_to_positive_valence'

crv.core.grammar.RepresentationEdgeKind.OBJECT_TO_NEGATIVE_VALENCE class-attribute instance-attribute

OBJECT_TO_NEGATIVE_VALENCE = 'object_to_negative_valence'

crv.core.grammar.RepresentationEdgeKind.OBJECT_TO_OBJECT class-attribute instance-attribute

OBJECT_TO_OBJECT = 'object_to_object'

crv.core.grammar.TopologyEdgeKind

Bases: enum.Enum

World topology edge kinds (STRUCTURE OUTSIDE the agent).

These DO NOT go into identity_edges. In a topology/world table, use these values for that table's edge_kind (e.g., 'world_topology_edges').

Examples:

• is_neighbor: grid/graph adjacency
• follows: directed social following
• in_group: static group membership edge

Notes

Table mappings: * world_topology_edges (future) — NOT part of identity_edges.

Source code in src/crv/core/grammar.py

class TopologyEdgeKind(Enum):
    """
    World topology edge kinds (STRUCTURE OUTSIDE the agent).

    These DO NOT go into identity_edges. In a topology/world table,
    use these values for that table's edge_kind (e.g., 'world_topology_edges').

    Examples:
      - is_neighbor: grid/graph adjacency
      - follows: directed social following
      - in_group: static group membership edge

    Notes:
      Table mappings:
        * world_topology_edges (future) — NOT part of identity_edges.
    """

    IS_NEIGHBOR = "is_neighbor"
    FOLLOWS = "follows"
    IN_GROUP = "in_group"
    CONNECTS_ROOM = "connects_room"

crv.core.grammar.TopologyEdgeKind.IS_NEIGHBOR class-attribute instance-attribute

IS_NEIGHBOR = 'is_neighbor'

crv.core.grammar.TopologyEdgeKind.FOLLOWS class-attribute instance-attribute

FOLLOWS = 'follows'

crv.core.grammar.TopologyEdgeKind.IN_GROUP class-attribute instance-attribute

IN_GROUP = 'in_group'

crv.core.grammar.TopologyEdgeKind.CONNECTS_ROOM class-attribute instance-attribute

CONNECTS_ROOM = 'connects_room'

crv.core.grammar.ExchangeKind

Bases: enum.Enum

Generalized ownership-exchange events (beyond finance).

Serialized values appear in

• exchange.exchange_event_type
• EBNF terminals in the 'exchange' section

Notes

Table mappings: * exchange (one row per event): - price/quantity for trade-like events - baseline_value if venue emits an index (price / poll % / trend) - additional_payload for venue specifics

Source code in src/crv/core/grammar.py

class ExchangeKind(Enum):
    """
    Generalized ownership-exchange events (beyond finance).

    Serialized values appear in:
      - exchange.exchange_event_type
      - EBNF terminals in the 'exchange' section

    Notes:
      Table mappings:
        * exchange (one row per event):
            - price/quantity for trade-like events
            - baseline_value if venue emits an index (price / poll % / trend)
            - additional_payload for venue specifics
    """

    TRADE = "trade"
    ORDER_POST = "order_post"
    ORDER_CANCEL = "order_cancel"
    SWAP = "swap"
    GIFT = "gift"
    PEER_OFFER = "peer_offer"
    PEER_ACCEPT = "peer_accept"
    PEER_REJECT = "peer_reject"
    CAST_VOTE = "cast_vote"
    VOTE_OUTCOME = "vote_outcome"

crv.core.grammar.ExchangeKind.TRADE class-attribute instance-attribute

TRADE = 'trade'

crv.core.grammar.ExchangeKind.ORDER_POST class-attribute instance-attribute

ORDER_POST = 'order_post'

crv.core.grammar.ExchangeKind.ORDER_CANCEL class-attribute instance-attribute

ORDER_CANCEL = 'order_cancel'

crv.core.grammar.ExchangeKind.SWAP class-attribute instance-attribute

SWAP = 'swap'

crv.core.grammar.ExchangeKind.GIFT class-attribute instance-attribute

GIFT = 'gift'

crv.core.grammar.ExchangeKind.PEER_OFFER class-attribute instance-attribute

PEER_OFFER = 'peer_offer'

crv.core.grammar.ExchangeKind.PEER_ACCEPT class-attribute instance-attribute

PEER_ACCEPT = 'peer_accept'

crv.core.grammar.ExchangeKind.PEER_REJECT class-attribute instance-attribute

PEER_REJECT = 'peer_reject'

crv.core.grammar.ExchangeKind.CAST_VOTE class-attribute instance-attribute

CAST_VOTE = 'cast_vote'

crv.core.grammar.ExchangeKind.VOTE_OUTCOME class-attribute instance-attribute

VOTE_OUTCOME = 'vote_outcome'

crv.core.grammar.TableName

Bases: enum.Enum

Canonical Parquet table names. Enforced by crv.io.tables.

Source code in src/crv/core/grammar.py

class TableName(Enum):
    """
    Canonical Parquet table names. Enforced by crv.io.tables.
    """

    EXCHANGE = "exchange"
    IDENTITY_EDGES = "identity_edges"
    SCENARIOS_SEEN = "scenarios_seen"
    MESSAGES = "messages"
    DECISIONS = "decisions"
    ORACLE_CALLS = "oracle_calls"
    HOLDINGS = "holdings"
    HOLDINGS_VALUATION = "holdings_valuation"

crv.core.grammar.TableName.EXCHANGE class-attribute instance-attribute

EXCHANGE = 'exchange'

crv.core.grammar.TableName.IDENTITY_EDGES class-attribute instance-attribute

IDENTITY_EDGES = 'identity_edges'

crv.core.grammar.TableName.SCENARIOS_SEEN class-attribute instance-attribute

SCENARIOS_SEEN = 'scenarios_seen'

crv.core.grammar.TableName.MESSAGES class-attribute instance-attribute

MESSAGES = 'messages'

crv.core.grammar.TableName.DECISIONS class-attribute instance-attribute

DECISIONS = 'decisions'

crv.core.grammar.TableName.ORACLE_CALLS class-attribute instance-attribute

ORACLE_CALLS = 'oracle_calls'

crv.core.grammar.TableName.HOLDINGS class-attribute instance-attribute

HOLDINGS = 'holdings'

crv.core.grammar.TableName.HOLDINGS_VALUATION class-attribute instance-attribute

HOLDINGS_VALUATION = 'holdings_valuation'

crv.core.grammar.TableDescriptor dataclass

Frozen descriptor for a canonical CRV table.

Attributes:

Name	Type	Description
name	crv.core.grammar.TableName	Canonical table identifier (lower_snake serialized).
columns	dict[str, str]	Mapping of lower_snake column_name -> dtype where dtype ∈ {"i64","f64","str","struct","list[struct]"}.
partitioning	list[str]	Partition columns (always ["bucket"]).
required	list[str]	Columns that must exist and be populated (non-null).
nullable	list[str]	Columns permitted to contain nulls. Validations for combinations are enforced by Pydantic row models downstream.
version	crv.core.versioning.SchemaVersion	Schema version pinned to crv.core.versioning.SCHEMA_V.

Notes

• Core remains zero-IO; this type describes the schema contract only.
• IO layers (crv.io) materialize, validate, and append frames against these descriptors.

Source code in src/crv/core/grammar.py

@dataclass(frozen=True)
class TableDescriptor:
    """
    Frozen descriptor for a canonical CRV table.

    Attributes:
        name (TableName): Canonical table identifier (lower_snake serialized).
        columns (dict[str, str]): Mapping of lower_snake column_name -> dtype
            where dtype ∈ {"i64","f64","str","struct","list[struct]"}.
        partitioning (list[str]): Partition columns (always ["bucket"]).
        required (list[str]): Columns that must exist and be populated (non-null).
        nullable (list[str]): Columns permitted to contain nulls. Validations for
            combinations are enforced by Pydantic row models downstream.
        version (SchemaVersion): Schema version pinned to crv.core.versioning.SCHEMA_V.

    Notes:
        - Core remains zero-IO; this type describes the schema contract only.
        - IO layers (crv.io) materialize, validate, and append frames against
          these descriptors.
    """

    name: TableName
    columns: dict[str, str]  # "i64","f64","str","struct","list[struct]"
    partitioning: list[str]  # always ["bucket"]
    required: list[str]
    nullable: list[str]
    version: SchemaVersion

crv.core.grammar.TableDescriptor.name instance-attribute

name: crv.core.grammar.TableName

crv.core.grammar.TableDescriptor.columns instance-attribute

columns: dict[str, str]

crv.core.grammar.TableDescriptor.partitioning instance-attribute

partitioning: list[str]

crv.core.grammar.TableDescriptor.required instance-attribute

required: list[str]

crv.core.grammar.TableDescriptor.nullable instance-attribute

nullable: list[str]

crv.core.grammar.TableDescriptor.version instance-attribute

version: crv.core.versioning.SchemaVersion

crv.core.grammar.is_lower_snake

is_lower_snake(value: str) -> bool

Check whether a string is lower_snake.

Parameters:

Name	Type	Description	Default
value	str	Candidate string to validate.	required

Returns:

Name	Type	Description
bool	bool	True if value matches lower_snake (e.g., "acquire_token"), False otherwise.

Examples:

>>> is_lower_snake("acquire_token")
True
>>> is_lower_snake("AcquireToken")
False

Source code in src/crv/core/grammar.py

def is_lower_snake(value: str) -> bool:
    """
    Check whether a string is lower_snake.

    Args:
      value (str): Candidate string to validate.

    Returns:
      bool: True if value matches lower_snake (e.g., "acquire_token"), False otherwise.

    Examples:
      >>> is_lower_snake("acquire_token")
      True
      >>> is_lower_snake("AcquireToken")
      False
    """
    return bool(_LOWER_SNAKE_RE.match(value or ""))

crv.core.grammar.assert_lower_snake

assert_lower_snake(value: str, what: str = 'value') -> None

Validate that a string is lower_snake.

Parameters:

Name	Type	Description	Default
value	str	Candidate string to validate.	required
what	str	Human-friendly label used in the error message.	'value'

Raises:

Type	Description
ValueError	If value is not lower_snake.

Source code in src/crv/core/grammar.py

def assert_lower_snake(value: str, what: str = "value") -> None:
    """
    Validate that a string is lower_snake.

    Args:
      value (str): Candidate string to validate.
      what (str): Human-friendly label used in the error message.

    Raises:
      ValueError: If value is not lower_snake.
    """
    if not is_lower_snake(value):
        raise ValueError(f"{what} must be lower_snake (got: {value!r})")

crv.core.grammar.action_value

action_value(kind: crv.core.grammar.ActionKind) -> str

Get the serialized (lower_snake) value for an ActionKind.

Parameters:

Name	Type	Description	Default
kind	crv.core.grammar.ActionKind	Action enum.	required

Returns:

Name	Type	Description
str	str	Lower_snake serialized value (e.g., "acquire_token").

Source code in src/crv/core/grammar.py

def action_value(kind: ActionKind) -> str:
    """
    Get the serialized (lower_snake) value for an ActionKind.

    Args:
      kind (ActionKind): Action enum.

    Returns:
      str: Lower_snake serialized value (e.g., "acquire_token").
    """
    return kind.value

crv.core.grammar.action_kind_from_value

action_kind_from_value(s: str) -> ActionKind

Parse a lower_snake action string into an ActionKind.

Parameters:

Name	Type	Description	Default
s	str	Lower_snake action string.	required

Returns:

Name	Type	Description
ActionKind	crv.core.grammar.ActionKind	Parsed action kind.

Raises:

Type	Description
ValueError	If s is not lower_snake or is not a known action.

Source code in src/crv/core/grammar.py

def action_kind_from_value(s: str) -> ActionKind:
    """
    Parse a lower_snake action string into an ActionKind.

    Args:
      s (str): Lower_snake action string.

    Returns:
      ActionKind: Parsed action kind.

    Raises:
      ValueError: If s is not lower_snake or is not a known action.
    """
    assert_lower_snake(s, "action_type")
    return ActionKind(s)

crv.core.grammar.exchange_value

exchange_value(kind: crv.core.grammar.ExchangeKind) -> str

Get the serialized (lower_snake) value for an ExchangeKind.

Parameters:

Name	Type	Description	Default
kind	crv.core.grammar.ExchangeKind	Exchange enum.	required

Returns:

Name	Type	Description
str	str	Lower_snake serialized value (e.g., "trade").

Source code in src/crv/core/grammar.py

def exchange_value(kind: ExchangeKind) -> str:
    """
    Get the serialized (lower_snake) value for an ExchangeKind.

    Args:
      kind (ExchangeKind): Exchange enum.

    Returns:
      str: Lower_snake serialized value (e.g., "trade").
    """
    return kind.value

crv.core.grammar.exchange_kind_from_value

exchange_kind_from_value(s: str) -> ExchangeKind

Parse a lower_snake exchange string into an ExchangeKind.

Parameters:

Name	Type	Description	Default
s	str	Lower_snake exchange kind string.	required

Returns:

Name	Type	Description
ExchangeKind	crv.core.grammar.ExchangeKind	Parsed exchange kind.

Raises:

Type	Description
ValueError	If s is not lower_snake or is not a known exchange kind.

Source code in src/crv/core/grammar.py

def exchange_kind_from_value(s: str) -> ExchangeKind:
    """
    Parse a lower_snake exchange string into an ExchangeKind.

    Args:
      s (str): Lower_snake exchange kind string.

    Returns:
      ExchangeKind: Parsed exchange kind.

    Raises:
      ValueError: If s is not lower_snake or is not a known exchange kind.
    """
    assert_lower_snake(s, "exchange_event_type")
    return ExchangeKind(s)

crv.core.grammar.edge_value

edge_value(kind: crv.core.grammar.RepresentationEdgeKind) -> str

Get the serialized (lower_snake) value for a RepresentationEdgeKind.

Parameters:

Name	Type	Description	Default
kind	crv.core.grammar.RepresentationEdgeKind	Edge kind enum.	required

Returns:

Name	Type	Description
str	str	Lower_snake serialized value (e.g., "object_to_object").

Source code in src/crv/core/grammar.py

def edge_value(kind: RepresentationEdgeKind) -> str:
    """
    Get the serialized (lower_snake) value for a RepresentationEdgeKind.

    Args:
      kind (RepresentationEdgeKind): Edge kind enum.

    Returns:
      str: Lower_snake serialized value (e.g., "object_to_object").
    """
    return kind.value

crv.core.grammar.edge_kind_from_value

edge_kind_from_value(s: str) -> RepresentationEdgeKind

Parse a lower_snake edge_kind string into a RepresentationEdgeKind.

Parameters:

Name	Type	Description	Default
s	str	Lower_snake edge kind string.	required

Returns:

Name	Type	Description
RepresentationEdgeKind	crv.core.grammar.RepresentationEdgeKind	Parsed edge kind.

Raises:

Type	Description
ValueError	If s is not lower_snake or is not a known edge kind.

Source code in src/crv/core/grammar.py

def edge_kind_from_value(s: str) -> RepresentationEdgeKind:
    """
    Parse a lower_snake edge_kind string into a RepresentationEdgeKind.

    Args:
      s (str): Lower_snake edge kind string.

    Returns:
      RepresentationEdgeKind: Parsed edge kind.

    Raises:
      ValueError: If s is not lower_snake or is not a known edge kind.
    """
    assert_lower_snake(s, "edge_kind")
    return RepresentationEdgeKind(s)

crv.core.grammar.normalize_visibility

normalize_visibility(vis: str) -> str

Normalize a free-form visibility token to canonical lower_snake.

Parameters:

Name	Type	Description	Default
vis	str	Candidate visibility token.	required

Returns:

Name	Type	Description
str	str	One of {"public","group","room","dm"}.

Raises:

Type	Description
ValueError	If the token does not match a known visibility class.

Notes

Used by: - messages.visibility_scope - scenarios_seen.visibility_scope - event envelopes visibility field

Source code in src/crv/core/grammar.py

def normalize_visibility(vis: str) -> str:
    """
    Normalize a free-form visibility token to canonical lower_snake.

    Args:
      vis (str): Candidate visibility token.

    Returns:
      str: One of {"public","group","room","dm"}.

    Raises:
      ValueError: If the token does not match a known visibility class.

    Notes:
      Used by:
        - messages.visibility_scope
        - scenarios_seen.visibility_scope
        - event envelopes visibility field
    """
    vis_l = (vis or "").lower()
    allowed = {v.value for v in Visibility}
    if vis_l not in allowed:
        raise ValueError(f"visibility must be one of {sorted(allowed)} (got {vis!r})")
    return vis_l

crv.core.grammar.normalize_channel_type

normalize_channel_type(ch: str) -> str

Normalize a free-form channel type token to canonical lower_snake.

Parameters:

Name	Type	Description	Default
ch	str	Candidate channel type token.	required

Returns:

Name	Type	Description
str	str	One of {"public","group","room","dm"}.

Raises:

Type	Description
ValueError	If the token is not a recognized channel type.

Notes

Typically used only for validation when parsing a channel prefix.

Source code in src/crv/core/grammar.py

def normalize_channel_type(ch: str) -> str:
    """
    Normalize a free-form channel type token to canonical lower_snake.

    Args:
      ch (str): Candidate channel type token.

    Returns:
      str: One of {"public","group","room","dm"}.

    Raises:
      ValueError: If the token is not a recognized channel type.

    Notes:
      Typically used only for validation when parsing a channel prefix.
    """
    ch_l = (ch or "").lower()
    allowed = {c.value for c in ChannelType}
    if ch_l not in allowed:
        raise ValueError(f"channel type must be one of {sorted(allowed)} (got {ch!r})")
    return ch_l

crv.core.grammar.canonical_action_key

canonical_action_key(action_type: crv.core.grammar.ActionKind, **params: object) -> str

Build a compact, human-friendly log key for an action candidate.

Parameters:

Name	Type	Description	Default
action_type	crv.core.grammar.ActionKind	Action type to label.	required
**params	object	Optional parameters to include in the label.	{}

Returns:

Name	Type	Description
str	str	Key formatted as "action:sorted_k=v|k=v".

Examples:

>>> canonical_action_key(ActionKind.ACQUIRE_TOKEN, token_id="Alpha")
'acquire_token:token_id=Alpha'
>>> canonical_action_key(ActionKind.SEND_CHAT_MESSAGE, channel="group:G1")
'send_chat_message:channel=group:G1'

Notes

The key is for logs and dashboards only; program logic must use structured fields rather than parsing this string.

Used by: - decisions.action_candidates[].key (string) - logs / dashboards (viz) as human-readable labels

Source code in src/crv/core/grammar.py

def canonical_action_key(action_type: ActionKind, **params: object) -> str:
    """
    Build a compact, human-friendly log key for an action candidate.

    Args:
      action_type (ActionKind): Action type to label.
      **params (object): Optional parameters to include in the label.

    Returns:
      str: Key formatted as "action:sorted_k=v|k=v".

    Examples:
      >>> canonical_action_key(ActionKind.ACQUIRE_TOKEN, token_id="Alpha")
      'acquire_token:token_id=Alpha'
      >>> canonical_action_key(ActionKind.SEND_CHAT_MESSAGE, channel="group:G1")
      'send_chat_message:channel=group:G1'

    Notes:
      The key is for logs and dashboards only; program logic must use
      structured fields rather than parsing this string.

      Used by:
        - decisions.action_candidates[].key (string)
        - logs / dashboards (viz) as human-readable labels
    """
    parts = [action_type.value]
    if params:
        ordered = ", ".join(f"{k}={v}" for k, v in sorted(params.items()))
        parts.append(ordered.replace(", ", "|"))
    return ":".join(parts)

crv.core.grammar.ensure_all_enum_values_lower_snake

ensure_all_enum_values_lower_snake(
    enums: collections.abc.Iterable[type[enum.Enum]],
) -> None

Assert that every enum member's value is lower_snake.

Parameters:

Name	Type	Description	Default
enums	collections.abc.Iterable[type[enum.Enum]]	Iterable of Enum classes to inspect.	required

Raises:

Type	Description
AssertionError	If any enum member has a non-lower_snake value.

Examples:

>>> ensure_all_enum_values_lower_snake([
...     ActionKind, ChannelType, Visibility,
...     PatchOp, RepresentationEdgeKind, ExchangeKind, TableName
... ])

Source code in src/crv/core/grammar.py

def ensure_all_enum_values_lower_snake(enums: Iterable[type[Enum]]) -> None:
    """
    Assert that every enum member's value is lower_snake.

    Args:
      enums (Iterable[type[Enum]]): Iterable of Enum classes to inspect.

    Raises:
      AssertionError: If any enum member has a non-lower_snake value.

    Examples:
      >>> ensure_all_enum_values_lower_snake([
      ...     ActionKind, ChannelType, Visibility,
      ...     PatchOp, RepresentationEdgeKind, ExchangeKind, TableName
      ... ])
    """
    for E in enums:
        for m in E:
            if not is_lower_snake(m.value):
                raise AssertionError(
                    f"{E.__name__}.{m.name} has non-lower_snake value: {m.value!r}"
                )