How to Add a New Memory Ingestion Source to OpenHuman's Memory System

To add a new memory ingestion source to OpenHuman, you implement a parser in parse.rs that converts raw data into NamespaceDocumentInput objects, register a dispatch handler in queue.rs that forwards to ingest_doc, and expose a JSON-RPC endpoint via schemas.rs.

OpenHuman's unified memory system stores knowledge in a SQLite-backed UnifiedMemory, accepting data from external sources like Slack, email, and Notion through a standardized ingestion pipeline. Adding a custom source—whether CSV exports, REST API payloads, or proprietary log formats—requires extending three specific Rust modules in the tinyhumansai/openhuman repository to transform, queue, and expose your data. This guide provides the exact file paths, method signatures, and working code examples needed to integrate a new ingestion source.

Step 1: Build a Source-Specific Parser

All ingestion logic begins in src/openhuman/memory/ingestion/parse.rs. A parser receives raw bytes or structured payloads alongside a MemoryIngestionConfig, then returns a ParsedIngestion containing a list of NamespaceDocumentInput objects. The parser must not write directly to the database; it only transforms input into the canonical document format the unified store expects.

Here is a complete parser implementation that ingests CSV-formatted contacts:

// src/openhuman/memory/ingestion/parse.rs
use crate::openhuman::memory::ingestion::{MemoryIngestionConfig, ParsedIngestion};

/// Parse CSV lines like “name,email,phone” into memory documents.
pub fn parse_csv_contacts(
    raw: &str,
    config: &MemoryIngestionConfig,
) -> Result<ParsedIngestion, String> {
    let mut docs = Vec::new();

    for (i, line) in raw.lines().enumerate() {
        let fields: Vec<_> = line.split(',').collect();
        if fields.len() != 3 {
            return Err(format!("CSV line {} malformed", i + 1));
        }
        let content = format!("Contact: {}\nEmail: {}\nPhone: {}", fields[0], fields[1], fields[2]);

        docs.push(NamespaceDocumentInput {
            namespace: "contacts".into(),
            title: fields[0].to_string(),
            content,
            metadata: Default::default(),
            // optional: attach tags, owners, etc.
        });
    }

    Ok(ParsedIngestion { docs })
}

The config parameter allows you to respect user-configurable limits such as maximum chunk size or content filters. Errors return as String values so the surrounding ingestion machinery surfaces clear RPC error messages to callers.

Step 2: Hook the Parser into the Ingestion Queue

The singleton IngestionQueue in src/openhuman/memory/ingestion/queue.rs manages all background ingestion jobs. To activate your parser, add a match arm to the dispatch function that maps a unique method name—such as "memory_csv_ingest"—to your parsing logic, then forwards the result to the generic ingest_doc helper.

// src/openhuman/memory/ingestion/queue.rs
use super::{parse_csv_contacts, MemoryIngestionRequest};

pub async fn dispatch(
    method: &str,
    req: MemoryIngestionRequest,
) -> Result<MemoryIngestionResult, String> {
    match method {
        // Existing methods …
        "memory_csv_ingest" => {
            // `req.payload` is the raw CSV string supplied by the caller.
            let parsed = parse_csv_contacts(&req.payload, &req.config)?;
            // Forward to the generic document ingest routine.
            crate::openhuman::memory::ops::documents::ingest_doc(parsed, req).await
        }
        _ => Err(format!("unknown ingestion method `{}`", method)),
    }
}

The ingest_doc function in src/openhuman/memory/ops/documents.rs handles the actual database persistence. Routing through this path guarantees your new source inherits existing infrastructure including [memory:ingest] logging, event-bus notifications (MemoryIngestionStarted and MemoryIngestionCompleted), and automatic idempotency checks.

Step 3: Expose a JSON-RPC Endpoint

OpenHuman generates its RPC surface from controller schemas defined in src/openhuman/memory/schemas.rs. Add a ControllerSchema entry that declares the method name, request payload structure, and a thin handler wrapper that forwards to the queue dispatcher.

// src/openhuman/memory/schemas.rs
use crate::core::event_bus::DomainEvent;

fn schemas() -> Vec<ControllerSchema> {
    vec![
        // Existing memory_* RPCs …
        ControllerSchema {
            function: "memory_csv_ingest".into(),
            // The request shape is the same as any other ingestion request:
            //   { source_id: String, owner: String, tags: Vec<String>, payload: String }
            request: json!({
                "source_id": "string",
                "owner": "string",
                "tags": ["string"],
                "payload": "string"
            }),
            handler: handle_memory_csv_ingest,
        },
    ]
}

// The thin wrapper that forwards to the queue dispatcher.
fn handle_memory_csv_ingest(params: Map<String, Value>) -> ControllerFuture {
    let req: MemoryIngestionRequest = serde_json::from_value(params.into())?;
    async move { queue::dispatch("memory_csv_ingest", req).await?.into_cli_compatible_json() }
}

src/core/all.rs automatically aggregates all controller collections, making the new method reachable via the core RPC server as openhuman.memory_csv_ingest. You can invoke it from the command line:

openhuman-core memory_csv_ingest \
  --source-id contacts-csv \
  --owner alice \
  --tags contacts \
  --payload "$(cat contacts.csv)"

Consuming Your Ingestion Source

Once registered, you can trigger ingestion from Rust code or external JSON-RPC clients.

In-Process Rust Calls

Use MemoryClient::ingest_doc to queue documents directly from within the OpenHuman runtime:

use openhuman_core::openhuman::memory::{MemoryClient, MemoryIngestionRequest};

let client = MemoryClient::from_workspace_dir(workspace_path)?;
let csv = "Bob,bob@example.com,555‑1234\nAlice,alice@example.com,555‑5678";

let request = MemoryIngestionRequest {
    source_id: "contacts-csv".into(),
    owner: "alice".into(),
    tags: vec!["contacts".into()],
    payload: csv.into(),
    config: MemoryIngestionConfig::default(),
};

let result = client.ingest_doc(request).await?;
println!("Ingested {} docs", result.docs_written);

JSON-RPC from External Clients

External agents or the OpenHuman UI can call the endpoint using standard JSON-RPC:

{
  "jsonrpc": "2.0",
  "method": "openhuman.memory_csv_ingest",
  "params": {
    "source_id": "contacts-csv",
    "owner": "alice",
    "tags": ["contacts"],
    "payload": "Bob,bob@example.com,555-1234\nAlice,alice@example.com,555-5678"
  },
  "id": 42
}

The response returns a MemoryIngestionResult containing docs_written, chunks_written, and chunks_dropped, identical to the response shape used by Slack, email, and Notion ingestion.

Summary

• Implement the parser in src/openhuman/memory/ingestion/parse.rs to transform raw source data into NamespaceDocumentInput objects without touching the database.
• Register the dispatcher in src/openhuman/memory/ingestion/queue.rs by adding a match arm that calls your parser and forwards results to ingest_doc in src/openhuman/memory/ops/documents.rs.
• Expose the endpoint in src/openhuman/memory/schemas.rs via a ControllerSchema definition, automatically surfacing the method through src/core/all.rs.
• Consume via JSON-RPC or the Rust MemoryClient to trigger ingestion while inheriting built-in logging, event notifications, and idempotency guarantees.

Frequently Asked Questions

Where does the actual database writing occur?

The parser you write in parse.rs only produces NamespaceDocumentInput objects; it never writes to SQLite directly. Database persistence happens inside ingest_doc in src/openhuman/memory/ops/documents.rs, which is called by the IngestionQueue dispatcher. This separation ensures all sources share the same transaction logic, idempotency checks, and event emission.

Can I reuse existing ingestion configurations for my custom source?

Yes. The MemoryIngestionConfig passed to your parser contains user-configurable limits such as maximum chunk size and content filters. Respect these settings in your parsing logic to ensure your source behaves consistently with built-in sources like Slack and email. The configuration is automatically populated from the MemoryIngestionRequest received via JSON-RPC.

What logging and monitoring is automatically available?

Any source routed through the IngestionQueue automatically emits [memory:ingest] prefixed logs and publishes MemoryIngestionStarted and MemoryIngestionCompleted events to the OpenHuman event bus. These events appear in the UI status pills and system monitoring dashboards without requiring additional instrumentation in your parser code.

How do I handle malformed input during parsing?

Return descriptive String errors from your parser function. The dispatch function in queue.rs propagates these errors as JSON-RPC errors, allowing callers to receive clear failure messages. For CSV parsing, this means validating row lengths and returning specific line numbers where formatting fails.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}