Ingestion Pipeline

Relevant source files

The following files were used as context for generating this wiki page:

• ethos-server/src/subsystems/ingest.rs
• ethos-server/src/http.rs
• ethos-ingest-ts/src/client.ts
• migrations/002_story_004_ingest.sql
• ethos-server/tests/ingest_integration.rs
• README.md

The Ingestion Pipeline is the primary entry point for capturing and persisting conversation data and memory traces within the Ethos neuromorphic memory engine. It passively records every conversation turn and embeds them using Gemini to create a foundation for long-term memory. The pipeline is designed to be high-velocity and non-blocking, utilizing asynchronous background tasks for heavy operations like embedding generation.

The pipeline spans multiple components, starting from client-side hooks (such as those for OpenClaw) that send data via Unix domain sockets or HTTP, to the core server subsystems that handle atomic database transactions and background embedding loops.

Sources: README.md:1-20, ethos-server/src/subsystems/ingest.rs:1-10

Architecture and Data Flow

The ingestion architecture follows a "fire-and-forget" pattern where raw conversation data is captured instantly and more complex processing (like semantic embedding) happens asynchronously.

Ingestion Sequence

1. Client Request: A client (e.g., the TypeScript EthosClient or an HTTP client) sends an ingest payload containing content, source, and metadata.
2. IPC/HTTP Entry: The server receives the request via the Unix socket listener or the Axum HTTP /ingest endpoint.
3. Atomic Persistence: The ingest_payload_with_embedding function executes a database transaction that writes simultaneously to two tables: session_events and memory_vectors.
4. Async Embedding: After the transaction commits, a background task is spawned to generate vector embeddings using the Gemini API.

Core Components

Ingest Subsystem

The logic for processing incoming payloads resides in ethos-server/src/subsystems/ingest.rs. This subsystem is responsible for extracting content and metadata, mapping sources to specific database roles, and managing the atomic database transaction.

Key logic includes:

• Role Mapping: Normalizes source types (user, assistant, system, tool) into database roles.
• Transactional Integrity: Ensures that a conversation turn is never partially recorded; it must exist in both the raw session log and the vector memory table.
• Background Spawning: Uses embedder::spawn_embed_task to offload API calls.

Sources: ethos-server/src/subsystems/ingest.rs:36-50, ethos-server/src/subsystems/ingest.rs:77-83

Client Protocol

Clients communicate with the pipeline using either Little-Endian (LE) framed MessagePack over Unix sockets or standard JSON over HTTP.

Field	Type	Description
content	String	The actual text of the message or memory.
source	String	The origin: user, assistant, system, or tool.
metadata	JSON/Map	Optional object containing session_id, agent_id, or author.

Sources: ethos-ingest-ts/src/client.ts:10-20, ethos-server/src/http.rs:85-95

Data Models

The ingestion pipeline populates two primary tables that act as the raw input for the memory engine.

Table: session_events

This table serves as a high-velocity, unlogged raw turn log.

Column	Type	Description
id	UUID	Primary key.
session_id	TEXT	Identifier for the conversation session.
agent_id	TEXT	Identifier for the AI agent involved.
role	TEXT	Checked: 'user', 'assistant', 'system', 'tool'.
content	TEXT	The raw message text.
metadata	JSONB	Additional contextual data.

Table: memory_vectors

This table stores the data that will eventually be used for semantic search. During ingestion, the vector column is temporarily null until the background embedder completes.

Column	Type	Description
id	UUID	Primary key (returned to client as memory_id).
content	TEXT	Snapshot of the content for vector matching.
vector	vector(768)	Gemini embedding (nullable during initial ingest).
importance	FLOAT	Default 0.5; used for decay calculations.
source	TEXT	Author or source of the memory.

Sources: migrations/002_story_004_ingest.sql:5-30, ethos-server/src/subsystems/ingest.rs:65-75

Integration and Implementation

The pipeline is tested and validated through integration tests and specialized client libraries.

Implementation Snippet: payload Ingestion

The following logic illustrates how the server handles the dual-write to the database:

// ethos-server/src/subsystems/ingest.rs:55-75
let mut tx = pool.begin().await?;

sqlx::query!(
    "INSERT INTO session_events (session_id, agent_id, role, content, metadata) VALUES ($1, $2, $3, $4, $5)",
    session_id, agent_id, role, content, serde_json::to_value(metadata).unwrap_or(serde_json::json!({}))
).execute(&mut *tx).await?;

let row = sqlx::query!(
    "INSERT INTO memory_vectors (content, source, metadata) VALUES ($1, $2, $3) RETURNING id",
    content, author, serde_json::to_value(metadata).unwrap_or(serde_json::json!({}))
).fetch_one(&mut *tx).await?;

tx.commit().await?;

Sources: ethos-server/src/subsystems/ingest.rs:55-75

Verification

Integration tests verify that the pipeline correctly maps roles and persists data to the correct tables. For example, the test_ingest_session_events test ensures that payloads with metadata result in searchable rows in session_events.

Sources: ethos-server/tests/ingest_integration.rs:10-35

Summary

The Ingestion Pipeline provides a robust and scalable method for capturing AI interactions. By separating the high-velocity database writes from the high-latency embedding generation, it ensures that conversational agents remain responsive while building a rich, semantically searchable long-term memory. The use of atomic transactions ensures data consistency between raw logs and memory vectors.

Sources: README.md:80-95, ethos-server/src/subsystems/ingest.rs:1-10