service

service

Persistent memory service: async fact extraction integrated into core.

MemoryService runs fact extraction on a dedicated background thread so it never blocks diapason serve request handling or the diapason chat REPL. Callers hand off an exchange via :meth:submit, which enqueues the work and returns immediately — the slow model call and disk write happen out of band. Ollama admission defers memory while an interactive turn is running. A background thread alone does not prevent competition for the same GPU. The worker swallows every per-job error (including BrokenPipeError when a client disconnects mid-extraction), so a flaky extraction model can never take down the host process.

The service is started and stopped as part of the Diapason lifecycle (see cli/serve.py and cli/chat_cmd.py) and is configured through the [memory] section of config.toml.

Classes

MemoryService

MemoryService(
    store: FactStore,
    extractor: FactExtractor,
    *,
    event_bus: EventBus | None = None,
    max_queue: int = 256,
)

Background long-term-memory extraction and persistence service.

Source code in src/diapason/memory/service.py

def __init__(
    self,
    store: FactStore,
    extractor: FactExtractor,
    *,
    event_bus: EventBus | None = None,
    max_queue: int = 256,
) -> None:
    self._store = store
    self._extractor = extractor
    self._event_bus = event_bus
    self._subscribed = False
    self._queue: "queue.Queue[Any]" = queue.Queue(maxsize=max(1, max_queue))
    self._thread: Optional[threading.Thread] = None
    self._running = threading.Event()
    self._stopping = threading.Event()
    self._pending: set[tuple[str, str]] = set()
    self._pending_lock = threading.Lock()

Methods:

start

start() -> None

Start the background worker thread (idempotent).

Source code in src/diapason/memory/service.py

def start(self) -> None:
    """Start the background worker thread (idempotent)."""
    if self._running.is_set():
        return
    if self._thread is not None and self._thread.is_alive():
        return
    self._stopping.clear()
    self._running.set()
    self._subscribe_events()
    self._thread = threading.Thread(
        target=self._loop,
        name="memory-service",
        daemon=True,
    )
    self._thread.start()
    logger.debug("Memory service started")

stop

stop(timeout: float = 2.0) -> None

Stop pending admission; finish an already sent call, then join.

Source code in src/diapason/memory/service.py

def stop(self, timeout: float = 2.0) -> None:
    """Stop pending admission; finish an already sent call, then join."""
    if not self._running.is_set():
        return
    with self._pending_lock:
        self._running.clear()
        self._stopping.set()
        try:
            self._queue.put_nowait(_STOP)
        except queue.Full:
            pass  # worker will notice the cleared flag on its next loop
    thread = self._thread
    if thread is not None:
        thread.join(timeout=timeout)
    if thread is None or not thread.is_alive():
        self._thread = None
    self._unsubscribe_events()
    logger.debug("Memory service stopped")

submit

submit(user_text: str, assistant_text: str = '') -> bool

Queue an exchange for extraction. Non-blocking; never raises.

Returns True if the job was enqueued, False if the service is not running or the queue is full (in which case the exchange is dropped rather than blocking the caller — extraction is best-effort).

Source code in src/diapason/memory/service.py

def submit(self, user_text: str, assistant_text: str = "") -> bool:
    """Queue an exchange for extraction. Non-blocking; never raises.

    Returns True if the job was enqueued, False if the service is not
    running or the queue is full (in which case the exchange is dropped
    rather than blocking the caller — extraction is best-effort).
    """
    if not self._running.is_set():
        return False
    if not user_text or not user_text.strip():
        return False
    job = (user_text, assistant_text)
    # 19/09/2026: identical lifecycle notifications need one extraction.
    # Never merge different replies or conversations by text similarity.
    with self._pending_lock:
        if not self._running.is_set():
            return False
        if job in self._pending:
            return True
        try:
            self._pending.add(job)
            self._queue.put_nowait(job)
            return True
        except queue.Full:
            self._pending.discard(job)
            logger.debug("Memory service queue full; dropping exchange")
            return False

Functions:

build_memory_service

build_memory_service(
    config: Any,
    engine: Any,
    default_model: str = "",
    *,
    event_bus: EventBus | None = None,
    memory_backend: Any = None,
) -> Optional[MemoryService]

Build a :class:MemoryService from config, or None if disabled.

Reads the [memory] section (config.memory / config.tools.storage) for enabled, backend, extraction_model, max_facts and facts_path. Returns None when memory is disabled or no engine / extraction model is available, so callers can simply do::

svc = build_memory_service(config, engine, model)
if svc is not None:
    svc.start()

Source code in src/diapason/memory/service.py

def build_memory_service(
    config: Any,
    engine: Any,
    default_model: str = "",
    *,
    event_bus: EventBus | None = None,
    memory_backend: Any = None,
) -> Optional[MemoryService]:
    """Build a :class:`MemoryService` from config, or ``None`` if disabled.

    Reads the ``[memory]`` section (``config.memory`` / ``config.tools.storage``)
    for ``enabled``, ``backend``, ``extraction_model``, ``max_facts`` and
    ``facts_path``.  Returns ``None`` when memory is disabled or no engine /
    extraction model is available, so callers can simply do::

        svc = build_memory_service(config, engine, model)
        if svc is not None:
            svc.start()
    """
    mem = getattr(config, "memory", None)
    if mem is None or not getattr(mem, "enabled", False):
        return None
    if engine is None:
        return None

    model = getattr(mem, "extraction_model", "") or default_model
    if not model:
        logger.debug("Memory service disabled: no extraction model available")
        return None

    store: FactStore = create_fact_store(
        getattr(mem, "backend", "local"),
        path=getattr(mem, "facts_path", None),
        max_facts=getattr(mem, "max_facts", 1000),
    )
    # Sans ce raccord, l'extraction écrivait dans le vide : les faits allaient
    # au journal JSONL tandis que l'injection de contexte interrogeait le
    # magasin vectoriel. Diapason distillait correctement « le projet Olala
    # doit être publié avant fin septembre », le rangeait, et ne le retrouvait
    # plus jamais. Voir SearchableFactStore.
    if memory_backend is not None:
        store = SearchableFactStore(store, memory_backend)
    extractor = FactExtractor(
        engine, model, use_active_model=not bool(getattr(mem, "extraction_model", ""))
    )
    return MemoryService(store, extractor, event_bus=event_bus)

publish_completed_exchange

publish_completed_exchange(
    bus: EventBus | None,
    user_text: str,
    assistant_text: str = "",
    *,
    source: str = "",
) -> bool

Publish a completed chat exchange for lifecycle subscribers.

Source code in src/diapason/memory/service.py

def publish_completed_exchange(
    bus: EventBus | None,
    user_text: str,
    assistant_text: str = "",
    *,
    source: str = "",
) -> bool:
    """Publish a completed chat exchange for lifecycle subscribers."""
    if bus is None or not user_text or not user_text.strip():
        return False
    bus.publish(
        EventType.CHAT_EXCHANGE_COMPLETED,
        {
            "user_text": user_text,
            "assistant_text": assistant_text or "",
            "source": source,
        },
    )
    return True