fathom.FleetEngine

fathom.FleetEngine

Manages multiple session-scoped Engines backed by a shared :class:FactStore.

Fleet-scoped facts (those whose template has scope="fleet") are visible across all sessions via :meth:sync_fleet_facts. Every method that addresses a session takes its id and raises KeyError for an id with no session.

Known limitation: fleet retraction does not propagate. FleetEngine has no retract method and never calls :meth:FactStore.subscribe, and :meth:sync_fleet_facts only ever asserts. A fact removed from the shared store (explicitly, or by Redis TTL expiry) therefore stays in the working memory of every session that already synced it. To drop it from a session, rebuild that session with :meth:replace_session and sync it again.

Source code in src/fathom/fleet.py

class FleetEngine:
    """Manages multiple session-scoped Engines backed by a shared :class:`FactStore`.

    Fleet-scoped facts (those whose template has ``scope="fleet"``) are
    visible across all sessions via :meth:`sync_fleet_facts`. Every method
    that addresses a session takes its id and raises ``KeyError`` for an id
    with no session.

    Known limitation: fleet retraction does not propagate. FleetEngine has no
    ``retract`` method and never calls :meth:`FactStore.subscribe`, and
    :meth:`sync_fleet_facts` only ever asserts. A fact removed from the shared
    store (explicitly, or by Redis TTL expiry) therefore stays in the working
    memory of every session that already synced it. To drop it from a session,
    rebuild that session with :meth:`replace_session` and sync it again.
    """

    def __init__(
        self,
        fact_store: FactStore,
        rules_path: str,
        **engine_kwargs: Any,
    ) -> None:
        self._fact_store = fact_store
        self._rules_path = rules_path
        self._engine_kwargs = engine_kwargs
        self._sessions: dict[str, Engine] = {}

    @property
    def sessions(self) -> dict[str, Engine]:
        """Return the mapping of session IDs to Engine instances."""
        return dict(self._sessions)

    def create_session(self, session_id: str) -> Engine:
        """Create and return an isolated :class:`Engine` for *session_id*.

        The engine is loaded from the configured rules path and stored
        internally so it can be retrieved later.

        Raises:
            FleetError: *session_id* already has a session. Use
                :meth:`replace_session` to rebuild it.
        """
        if session_id in self._sessions:
            raise FleetError(
                f"session '{session_id}' already exists; use replace_session() to rebuild it",
                session_id=session_id,
            )
        return self._new_session(session_id)

    def replace_session(self, session_id: str) -> Engine:
        """Discard *session_id*'s engine and return a freshly loaded one.

        The new engine starts with empty working memory; call
        :meth:`sync_fleet_facts` to pull the fleet state back in.

        Raises:
            KeyError: *session_id* has no session.
        """
        self._session(session_id)
        return self._new_session(session_id)

    def remove_session(self, session_id: str) -> None:
        """Forget *session_id*'s engine.

        Raises:
            KeyError: *session_id* has no session.
        """
        self._session(session_id)
        del self._sessions[session_id]

    def _new_session(self, session_id: str) -> Engine:
        engine = Engine.from_rules(self._rules_path, **self._engine_kwargs)
        self._sessions[session_id] = engine
        return engine

    def _session(self, session_id: str) -> Engine:
        try:
            return self._sessions[session_id]
        except KeyError:
            raise KeyError(f"unknown session '{session_id}'") from None

    async def assert_fact(
        self,
        session_id: str,
        template: str,
        data: FactData,
    ) -> FactId | None:
        """Assert a fact into *session_id*, routing fleet-scoped facts to the store.

        Returns the assigned ``fact_id`` for fleet-scoped templates, or ``None``
        for session-scoped templates. The store receives the fact as the
        engine accepted it, defaults applied and values coerced, so
        :meth:`query` filters match the values this session's rules see.

        Raises:
            KeyError: if *session_id* has no associated Engine.
            ValidationError: if the template is not registered on that Engine.
        """
        engine = self._session(session_id)

        tmpl_def = engine.template_registry.get(template)
        if tmpl_def is None:
            from fathom.errors import ValidationError

            raise ValidationError(
                f"Unknown template '{template}'",
                template=template,
            )

        if tmpl_def.scope == "fleet":
            # Validate locally FIRST: a fact this engine rejects must never reach
            # the shared store, where it would abort every peer's sync.
            # Write what the engine accepted (defaults applied, values
            # coerced), not the raw caller dict, so store rows, peers and
            # `query` filters see the same values this node's rules match on.
            validated, handles = engine._assert_local(template, data)
            try:
                return await self._fact_store.assert_fact(template, validated)
            except Exception:
                # The store write failed after the local assert; undo it so this
                # session does not hold a fleet fact the store never accepted.
                #
                # Roll back BY HANDLE, never by value. A retract-by-value
                # filter is wrong twice over: it misses a value FactManager
                # coerced on the way in (int 123 -> "123" for a STRING slot),
                # silently leaving the rejected fact behind; and when CLIPS
                # de-duplicated this assert onto an identical fact that was
                # already committed, it deletes THAT one — silent local data
                # loss on a transient store blip during a repeated heartbeat.
                # `handles` is empty in the de-dup case, which is exactly
                # right: we created nothing, so we withdraw nothing.
                try:
                    engine._retract_local_handles(handles)
                except Exception:
                    logger.warning(
                        "failed to roll back local assert of '%s' after fact store error",
                        template,
                        exc_info=True,
                    )
                raise

        # session scope — local-only, so the public Engine entry point applies
        # (its fleet-scope guard cannot trigger here) and metrics stay accurate.
        engine.assert_fact(template, data)
        return None

    async def sync_fleet_facts(self, session_id: str) -> None:
        """Pull fleet-scoped facts from the :class:`FactStore` into *session_id*.

        Queries the fact store for every fleet-scoped template and
        asserts matching facts into the session's working memory. A fact the
        session rejects is logged and skipped, so one malformed row in the
        shared store cannot stop the rest of the fleet state from syncing.

        Raises:
            KeyError: *session_id* has no session.
        """
        session = self._session(session_id)
        # Discover fleet-scoped templates from the session's loaded templates
        for tmpl_name, tmpl_def in session.template_registry.items():
            scope = tmpl_def.scope
            if scope != "fleet":
                continue
            facts = await self._fact_store.query(tmpl_name)
            for fact in facts:
                # Remove fact_id metadata before asserting into CLIPS
                data = {k: v for k, v in fact.items() if k != "fact_id"}
                try:
                    # Pulled from the authoritative FactStore; bypass the Engine-level
                    # scope guard since this is the legitimate path for fleet facts into
                    # a local engine. Still takes the engine lock and records the metric.
                    session._assert_local(tmpl_name, data)
                except Exception:
                    logger.warning(
                        "skipping unsyncable fleet fact %s of template '%s'",
                        fact.get("fact_id", "<unknown>"),
                        tmpl_name,
                        exc_info=True,
                    )

    async def query(
        self,
        template: str,
        fact_filter: FactFilter | None = None,
    ) -> list[FactData]:
        """Query the shared :class:`FactStore` for facts matching *template*.

        Raises:
            ValidationError: *fact_filter* names a template or slot the rules
                do not declare. The store would match a null-valued unknown
                key against every row.
        """
        if fact_filter:
            # Every session loads the same rules, so any one's registry serves.
            # ponytail: with no session yet, loads the rules per filtered query.
            engine = next(iter(self._sessions.values()), None) or Engine.from_rules(
                self._rules_path, **self._engine_kwargs
            )
            filter_template(engine.template_registry, template, fact_filter)
        return await self._fact_store.query(template, fact_filter)

sessions property

Return the mapping of session IDs to Engine instances.

create_session(session_id)

Create and return an isolated :class:Engine for session_id.

The engine is loaded from the configured rules path and stored internally so it can be retrieved later.

Raises:

Type	Description
FleetError	session_id already has a session. Use :meth:replace_session to rebuild it.

Source code in src/fathom/fleet.py

def create_session(self, session_id: str) -> Engine:
    """Create and return an isolated :class:`Engine` for *session_id*.

    The engine is loaded from the configured rules path and stored
    internally so it can be retrieved later.

    Raises:
        FleetError: *session_id* already has a session. Use
            :meth:`replace_session` to rebuild it.
    """
    if session_id in self._sessions:
        raise FleetError(
            f"session '{session_id}' already exists; use replace_session() to rebuild it",
            session_id=session_id,
        )
    return self._new_session(session_id)

replace_session(session_id)

Discard session_id's engine and return a freshly loaded one.

The new engine starts with empty working memory; call :meth:sync_fleet_facts to pull the fleet state back in.

Raises:

Type	Description
KeyError	session_id has no session.

Source code in src/fathom/fleet.py

def replace_session(self, session_id: str) -> Engine:
    """Discard *session_id*'s engine and return a freshly loaded one.

    The new engine starts with empty working memory; call
    :meth:`sync_fleet_facts` to pull the fleet state back in.

    Raises:
        KeyError: *session_id* has no session.
    """
    self._session(session_id)
    return self._new_session(session_id)

remove_session(session_id)

Forget session_id's engine.

Raises:

Type	Description
KeyError	session_id has no session.

Source code in src/fathom/fleet.py

def remove_session(self, session_id: str) -> None:
    """Forget *session_id*'s engine.

    Raises:
        KeyError: *session_id* has no session.
    """
    self._session(session_id)
    del self._sessions[session_id]

assert_fact(session_id, template, data) async

Assert a fact into session_id, routing fleet-scoped facts to the store.

Returns the assigned fact_id for fleet-scoped templates, or None for session-scoped templates. The store receives the fact as the engine accepted it, defaults applied and values coerced, so :meth:query filters match the values this session's rules see.

Raises:

Type	Description
KeyError	if session_id has no associated Engine.
ValidationError	if the template is not registered on that Engine.

Source code in src/fathom/fleet.py

async def assert_fact(
    self,
    session_id: str,
    template: str,
    data: FactData,
) -> FactId | None:
    """Assert a fact into *session_id*, routing fleet-scoped facts to the store.

    Returns the assigned ``fact_id`` for fleet-scoped templates, or ``None``
    for session-scoped templates. The store receives the fact as the
    engine accepted it, defaults applied and values coerced, so
    :meth:`query` filters match the values this session's rules see.

    Raises:
        KeyError: if *session_id* has no associated Engine.
        ValidationError: if the template is not registered on that Engine.
    """
    engine = self._session(session_id)

    tmpl_def = engine.template_registry.get(template)
    if tmpl_def is None:
        from fathom.errors import ValidationError

        raise ValidationError(
            f"Unknown template '{template}'",
            template=template,
        )

    if tmpl_def.scope == "fleet":
        # Validate locally FIRST: a fact this engine rejects must never reach
        # the shared store, where it would abort every peer's sync.
        # Write what the engine accepted (defaults applied, values
        # coerced), not the raw caller dict, so store rows, peers and
        # `query` filters see the same values this node's rules match on.
        validated, handles = engine._assert_local(template, data)
        try:
            return await self._fact_store.assert_fact(template, validated)
        except Exception:
            # The store write failed after the local assert; undo it so this
            # session does not hold a fleet fact the store never accepted.
            #
            # Roll back BY HANDLE, never by value. A retract-by-value
            # filter is wrong twice over: it misses a value FactManager
            # coerced on the way in (int 123 -> "123" for a STRING slot),
            # silently leaving the rejected fact behind; and when CLIPS
            # de-duplicated this assert onto an identical fact that was
            # already committed, it deletes THAT one — silent local data
            # loss on a transient store blip during a repeated heartbeat.
            # `handles` is empty in the de-dup case, which is exactly
            # right: we created nothing, so we withdraw nothing.
            try:
                engine._retract_local_handles(handles)
            except Exception:
                logger.warning(
                    "failed to roll back local assert of '%s' after fact store error",
                    template,
                    exc_info=True,
                )
            raise

    # session scope — local-only, so the public Engine entry point applies
    # (its fleet-scope guard cannot trigger here) and metrics stay accurate.
    engine.assert_fact(template, data)
    return None

sync_fleet_facts(session_id) async

Pull fleet-scoped facts from the :class:FactStore into session_id.

Queries the fact store for every fleet-scoped template and asserts matching facts into the session's working memory. A fact the session rejects is logged and skipped, so one malformed row in the shared store cannot stop the rest of the fleet state from syncing.

Raises:

Type	Description
KeyError	session_id has no session.

Source code in src/fathom/fleet.py

async def sync_fleet_facts(self, session_id: str) -> None:
    """Pull fleet-scoped facts from the :class:`FactStore` into *session_id*.

    Queries the fact store for every fleet-scoped template and
    asserts matching facts into the session's working memory. A fact the
    session rejects is logged and skipped, so one malformed row in the
    shared store cannot stop the rest of the fleet state from syncing.

    Raises:
        KeyError: *session_id* has no session.
    """
    session = self._session(session_id)
    # Discover fleet-scoped templates from the session's loaded templates
    for tmpl_name, tmpl_def in session.template_registry.items():
        scope = tmpl_def.scope
        if scope != "fleet":
            continue
        facts = await self._fact_store.query(tmpl_name)
        for fact in facts:
            # Remove fact_id metadata before asserting into CLIPS
            data = {k: v for k, v in fact.items() if k != "fact_id"}
            try:
                # Pulled from the authoritative FactStore; bypass the Engine-level
                # scope guard since this is the legitimate path for fleet facts into
                # a local engine. Still takes the engine lock and records the metric.
                session._assert_local(tmpl_name, data)
            except Exception:
                logger.warning(
                    "skipping unsyncable fleet fact %s of template '%s'",
                    fact.get("fact_id", "<unknown>"),
                    tmpl_name,
                    exc_info=True,
                )

query(template, fact_filter=None) async

Query the shared :class:FactStore for facts matching template.

Raises:

Type	Description
ValidationError	fact_filter names a template or slot the rules do not declare. The store would match a null-valued unknown key against every row.

Source code in src/fathom/fleet.py

async def query(
    self,
    template: str,
    fact_filter: FactFilter | None = None,
) -> list[FactData]:
    """Query the shared :class:`FactStore` for facts matching *template*.

    Raises:
        ValidationError: *fact_filter* names a template or slot the rules
            do not declare. The store would match a null-valued unknown
            key against every row.
    """
    if fact_filter:
        # Every session loads the same rules, so any one's registry serves.
        # ponytail: with no session yet, loads the rules per filtered query.
        engine = next(iter(self._sessions.values()), None) or Engine.from_rules(
            self._rules_path, **self._engine_kwargs
        )
        filter_template(engine.template_registry, template, fact_filter)
    return await self._fact_store.query(template, fact_filter)