Module interaction 

Asynchronous History Tracking with InteractionObserver

When to Use It

Use InteractionObserver when your plugin needs to monitor conversation history in the background to build long-term memory, distill insights, log dialogue sentiment, or update external tracking systems.

The Finalized, Lossless Queue Pattern

Unlike the main cognitive loop which uses a highly filtered state, background observers receive a lossless, backpressured, private mpsc queue containing finalized interactions. An interaction is considered “finalized” by the core when it falls out of the active sliding window (becoming older than the last N active interactions). This ensures observers only process stable, permanent records, and avoids processing intermediate/unfinished states.

[!IMPORTANT] Implementation Rule: The actual size of the active sliding window (e.g. 8 interactions) is completely arbitrary, controlled by the core, and must remain completely opaque to plugins. Observers must never assume a specific window size or wait for a threshold; they must simply process all incoming finalized interactions from interaction_rx as they are received.

The Batched Draining Queue Pattern

Because LLM distillation or external API requests are slow and expensive, observers should never process every interaction sequentially. Instead, use the Batched Draining Pattern: wake up on recv(), drain any other immediately available interactions using non-blocking try_recv(), package them into a batch, and execute a single combined process or LLM inference.

How to Use It (Example)

Implement the specialized InteractionObserver trait, and register it inside your base Plugin::register hook:

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use synapto_interface::plugin::{Plugin, PluginRegistry, sync::watch, sync::mpsc};
use synapto_interface::core::{Interaction, Timestamp, InteractionMemory};

pub struct MyObserverPlugin {
    // A private sender to populate your internal state
    memory_tx: std::sync::Mutex<Option<watch::Sender<MyCustomState>>>,
}

impl Plugin for MyObserverPlugin {
    async fn create(context: synapto_interface::plugin::PluginContext) -> Result<Self, String> {
        // let config: MyObserverConfig = context.config()?; // Load config if needed
        let (memory_tx, memory_rx) = watch::channel(MyCustomState::default());
        Ok(Self {
            memory_tx: std::sync::Mutex::new(Some(memory_tx)),
        })
    }

    fn register<R: PluginRegistry + ?Sized>(self: Arc<Self>, registry: &mut R)
    where
        Self: Sized,
    {
        // 1. Registers as an InteractionObserver to receive the background queue
        registry.register_interaction_observer(self.clone());
    }
}

#[async_trait]
impl synapto_interface::interaction::InteractionObserver for MyObserverPlugin {
    async fn start(
        &self,
        mut interaction_rx: mpsc::Receiver<synapto_interface::interaction::ObservedInteraction>,
    ) -> Result<(), String> {
        let memory_tx = self.memory_tx.lock().unwrap().take().unwrap();

        tokio::spawn(async move {
            // Wake up when the first new interaction arrives
            while let Some(first) = interaction_rx.recv().await {

                // IMPORTANT BEST PRACTICE: Dynamic Batching
                // Once awake, immediately try to drain any additional interactions
                // that arrived concurrently into a single batch.
                // This is highly token-efficient and produces better summaries
                // during rapid multi-turn conversations or when catching up
                // from a long-running background task.
                let mut batch = vec![first];
                while let Ok(next) = interaction_rx.try_recv() {
                    batch.push(next);
                }

                // Process the batch (e.g. call an LLM to distill insights or log to database)
                // Doing this once per batch rather than per interaction saves API calls.
                tracing::info!("Processing a batch of {} interactions asynchronously...", batch.len());

                // Report progress back to the core to advance the sliding window
                let last_processed_timestamp = batch.last().unwrap().timestamp;
                let _ = rollout_tx.send(last_processed_timestamp);
            }
        });

        Ok(())
    }
}

Structs

CognitiveSpoken

CognitiveWritten

NotClearInteraction

NotClearInteractionMemory

ObservedInteraction

Timestamp

Traits

InteractionObserver

RetrospectiveConsolidationPlugin