orchestrata_ultimate.session.fact_emitter

Reference › orchestrata_ultimate › session › fact_emitter

FactEmitter turns session activity into stored facts. It implements IFactEmitter from interfaces over the dual-storage layers, with the Prolog engine as a third sink. mcp_tools uses it for the fact tools.

emit(session_id, fact) writes the fact to the RedisSessionStore, which must succeed, measures the latency, and then tries to persist the fact to the GraphSessionStore if one was given; a graph failure is logged as a warning and does not fail the emit. emit_typed builds a Fact of a given FactType and emits it, batch_emit emits many and returns the count, and get_facts, query_facts and get_fact_count read back from Redis. flush_to_prolog(session_id) asserts every Redis fact into Prolog as a session_fact/4 term (session id, <fact_type>:<fact_id>, timestamp, source) through the PrologSessionInterface, and returns 0 if no interface was passed. emit_count and avg_latency_ms expose simple metrics.

At a glance

Class diagram

Dependency diagram

API

Classes

FactEmitter

Bases: IFactEmitter · Decorators: traces(spec='006', story='US-003', reqs=['REQ-006-FR7-006']), traces(spec='006', story='FR-003', reqs=['REQ-006-FR3-001', 'REQ-006-FR3-002', 'REQ-006-FR3-003', 'REQ-006-FR3-004', 'R…, traces(spec='006', story='US-004', reqs=['REQ-006-US-004-001', 'REQ-006-US-004-002', 'REQ-006-US-004-003', 'REQ-006-US-… · Line 59

Emits session facts to dual storage backends.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/fact_emitter.py

# mypy: ignore-errors
"""Session fact emitter — dual-backend with Prolog source-of-truth sync.

Emits session facts to Redis (hot) and optionally graph (cold).
When a ``PrologSessionInterface`` is attached, :meth:`flush_to_prolog`
asserts each fact as ``session_fact(SessionID, FactTerm, Timestamp, Source)``
conforming to the spec 006 Prolog source-of-truth schema.

Target latency: < 10 ms for Redis emission.
"""

from __future__ import annotations

import logging
import time
import uuid
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.interfaces import IFactEmitter
from orchestrata_ultimate.session.redis_store import RedisSessionStore
from orchestrata_ultimate.session.types import Fact, FactFilter, FactType

if TYPE_CHECKING:
    from orchestrata_ultimate.session.graph_store import GraphSessionStore
    from orchestrata_ultimate.session.prolog_interface import PrologSessionInterface

logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="US-003",
    reqs=["REQ-006-FR7-006"],
)
@traces(
    spec="006",
    story="FR-003",
    reqs=[
        "REQ-006-FR3-001",
        "REQ-006-FR3-002",
        "REQ-006-FR3-003",
        "REQ-006-FR3-004",
        "REQ-006-FR3-005",
    ],
)
@traces(
    spec="006",
    story="US-004",
    reqs=[
        "REQ-006-US-004-001",
        "REQ-006-US-004-002",
        "REQ-006-US-004-003",
        "REQ-006-US-004-004",
        "REQ-006-US-004-005",
        "REQ-006-US-004-006",
    ],
)
class FactEmitter(IFactEmitter):
    """Emits session facts to dual storage backends.

    Redis is the primary (synchronous) store for low-latency writes.
    Graph is optional for archival graph persistence.
    A ``PrologSessionInterface`` can be attached so that
    :meth:`flush_to_prolog` pushes facts into the Prolog knowledge base
    using the spec 006 ``session_fact/4`` predicate.

    Args:
        redis_store: The Redis session store for hot fact storage.
        graph_store: Optional graph store for cold graph persistence.
        prolog_interface: Optional ``PrologSessionInterface`` for
            Prolog source-of-truth sync.

    Example:
        >>> emitter = FactEmitter(redis_store=store)
        >>> fact = emitter.emit_typed("sess_001", FactType.DECISION, {"key": "val"})
        >>> emitter.emit_count
        1
    """

    def __init__(
        self,
        redis_store: RedisSessionStore,
        graph_store: GraphSessionStore | None = None,
        prolog_interface: PrologSessionInterface | None = None,
    ) -> None:
        """Initialize the emitter with storage backends.

        Args:
            redis_store: Redis session store (required).
            graph_store: Optional graph session store.
            prolog_interface: Optional Prolog session interface for
                source-of-truth sync.
        """
        self._redis = redis_store
        self._graph = graph_store
        self._prolog = prolog_interface
        self._emit_count = 0
        self._total_latency_ms = 0.0

    @property
    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-003"])
    def emit_count(self) -> int:
        """Total facts emitted."""
        return self._emit_count

    @property
    def avg_latency_ms(self) -> float:
        """Average emit latency in milliseconds."""
        if self._emit_count == 0:
            return 0.0
        return self._total_latency_ms / self._emit_count

    @traces(
        spec="006",
        story="US-004",
        reqs=[
            "REQ-006-FR7-002",
            "REQ-006-FR8-001",
            "REQ-006-FR8-004",
            "REQ-006-FR8-005",
            "REQ-006-US-004-001",
            "REQ-006-US-004-004",
        ],
    )
    def emit(self, session_id: str, fact: Fact) -> None:
        """Emit a single fact to storage."""
        start = time.monotonic()
        self._redis.save_fact(session_id, fact)
        elapsed_ms = (time.monotonic() - start) * 1000
        self._emit_count += 1
        self._total_latency_ms += elapsed_ms

        # Best-effort graph persistence
        if self._graph is not None:
            try:
                self._graph.persist_fact(session_id, fact)
            except Exception as e:
                logger.warning("Graph fact persist failed: %s", e)

    @traces(spec="006", story="US-004", reqs=["REQ-006-FR8-003", "REQ-006-US-004-003"])
    def emit_typed(
        self,
        session_id: str,
        fact_type: FactType,
        data: dict,
        source: str = "",
    ) -> Fact:
        """Create and emit a typed fact."""
        fact = Fact(
            id=f"fact-{uuid.uuid4().hex[:12]}",
            session_id=session_id,
            fact_type=fact_type,
            data=data,
            source=source,
        )
        self.emit(session_id, fact)
        return fact

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-005"])
    def batch_emit(self, session_id: str, facts: list[Fact]) -> int:
        """Emit multiple facts in a batch."""
        start = time.monotonic()
        count = self._redis.save_facts_batch(session_id, facts)
        elapsed_ms = (time.monotonic() - start) * 1000
        self._emit_count += count
        self._total_latency_ms += elapsed_ms

        if self._graph is not None:
            try:
                self._graph.persist_facts_batch(session_id, facts)
            except Exception as e:
                logger.warning("Graph batch persist failed: %s", e)

        return count

    @traces(
        spec="006",
        story="US-003",
        reqs=[
            "REQ-006-FR3-004",
            "REQ-006-US-003-002",
            "REQ-006-US-004-006",
            "REQ-006-US-005-003",
        ],
    )
    def get_facts(
        self, session_id: str, filter: FactFilter | None = None
    ) -> list[Fact]:
        """Get facts for a session, optionally filtered."""
        return self._redis.get_facts(session_id, filter)

    @traces(
        spec="006",
        story="US-003",
        reqs=[
            "REQ-006-FR4-002",
            "REQ-006-FR9-001",
            "REQ-006-FR9-002",
            "REQ-006-FR9-004",
            "REQ-006-FR9-005",
            "REQ-006-US-003-004",
            "REQ-006-US-003-006",
            "REQ-006-US-004-006",
        ],
    )
    def query_facts(self, session_id: str, fact_type: str | None = None) -> list[Fact]:
        """Query facts by type."""
        if fact_type is not None:
            ft = FactType(fact_type)
            filter = FactFilter(fact_type=ft)
            return self._redis.get_facts(session_id, filter)
        return self._redis.get_facts(session_id)

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-005"])
    def get_fact_count(self, session_id: str) -> int:
        """Get total fact count for a session."""
        return self._redis.get_fact_count(session_id)

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-002"])
    def flush_to_prolog(self, session_id: str) -> int:
        """Flush all Redis facts into Prolog as ``session_fact/4`` terms.

        Each fact is asserted as::

            session_fact(SessionID, FactTerm, Timestamp, Source)

        where *FactTerm* is ``"<fact_type>:<fact_id>"``.

        Requires a ``PrologSessionInterface`` (with an active engine)
        to have been passed at construction.

        Args:
            session_id: Session whose facts to flush.

        Returns:
            Number of facts successfully asserted into Prolog.

        Example:
            >>> emitter = FactEmitter(store, prolog_interface=iface)
            >>> emitter.emit_typed("s1", FactType.DECISION, {"k": "v"}, source="user")
            >>> emitter.flush_to_prolog("s1")
            1
        """
        if self._prolog is None:
            return 0
        facts = self._redis.get_facts(session_id)
        count = 0
        for fact in facts:
            try:
                fact_term = f"{fact.fact_type.value}:{fact.id}"
                source = fact.source or ""
                ok = self._prolog.assert_session_fact(session_id, fact_term, source)
                if ok:
                    count += 1
            except Exception as e:
                logger.warning("Prolog fact assertion failed: %s", e)
        return count