From 34d15b1c1508babe0cdb1adc6c29494d6a888de7 Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Sat, 9 May 2026 15:50:30 -0400 Subject: [PATCH 1/4] feat(memory): add Contexto memory provider plugin Wires the self-hosted Contexto memory engine (port 4010 by default) into the MemoryProvider interface. Per-turn ingest of (user, assistant) pairs to /v1/ingest, background-threaded prefetch via /v1/search before each turn, and a contexto_search tool for explicit lookups. Agent slug is one-per-hermes-profile (derived from kwargs.agent_identity); userId is per-platform-user (kwargs.user_id, gateway-supplied). The plugin idempotently registers the agent slug on initialize. Pip dep points at the Python client in the contexto repo's clients/python/ subdirectory via git URL. Co-Authored-By: Claude Opus 4.7 (1M context) --- plugins/memory/contexto/README.md | 37 +++++++ plugins/memory/contexto/__init__.py | 166 ++++++++++++++++++++++++++++ plugins/memory/contexto/plugin.yaml | 5 + 3 files changed, 208 insertions(+) create mode 100644 plugins/memory/contexto/README.md create mode 100644 plugins/memory/contexto/__init__.py create mode 100644 plugins/memory/contexto/plugin.yaml diff --git a/plugins/memory/contexto/README.md b/plugins/memory/contexto/README.md new file mode 100644 index 000000000000..6cf3f4a943de --- /dev/null +++ b/plugins/memory/contexto/README.md @@ -0,0 +1,37 @@ +# Contexto memory plugin (self-hosted) + +Wires the self-hosted Contexto memory engine into hermes-agent's `MemoryProvider` interface. + +> Self-hosted only. The hosted `api.getcontexto.com` API has a completely different surface and is not supported by this plugin. + +## What it does + +- **Per-turn ingest:** every (user, assistant) pair is sent to `/v1/ingest`. The server runs LLM extraction and shards components across episodic, semantic, and procedural sectors. +- **Pre-turn recall:** before each turn, queries `/v1/search` with the user message and injects the top working-memory hits into the system prompt. +- **Tool exposure:** model can call `contexto_search` to look up prior knowledge explicitly. + +## Scoping + +- **Agent slug:** one per hermes profile (`agent_identity`). All sessions of the same profile pool into the same agent's memory. +- **User id:** passed through to Contexto for per-user scoping within an agent. + +## Install + +Bring up the Contexto self-hosted stack first (port 4010 by default), then: + +```bash +pip install "contexto @ git+https://github.com/amiller/contexto.git#subdirectory=clients/python" +``` + +## Activate + +In your hermes config: + +```yaml +memory: + provider: contexto +``` + +## Env vars + +- `CONTEXTO_BASE_URL` (optional — defaults to `http://localhost:4010`) diff --git a/plugins/memory/contexto/__init__.py b/plugins/memory/contexto/__init__.py new file mode 100644 index 000000000000..5cca8f1be448 --- /dev/null +++ b/plugins/memory/contexto/__init__.py @@ -0,0 +1,166 @@ +"""Contexto memory plugin — MemoryProvider backed by self-hosted Contexto. + +Per-turn ingest of (user, assistant) pairs into the Contexto OSS memory +engine, semantic recall via /v1/search before each turn, and a +contexto_search tool for explicit lookups. + +Agent slug is one-per-hermes-profile (derived from agent_identity). +userId is per-platform-user (gateway-supplied). + +Config via environment: + CONTEXTO_BASE_URL — selfhost API base (default http://localhost:4010) +""" + +from __future__ import annotations + +import json +import logging +import os +import threading +from typing import Any, Dict, List + +from agent.memory_provider import MemoryProvider +from tools.registry import tool_error + +logger = logging.getLogger(__name__) + + +SEARCH_SCHEMA = { + "name": "contexto_search", + "description": ( + "Search the Contexto cognitive memory for content relevant to a query. " + "Returns hits ranked across episodic, semantic, and procedural sectors." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to look up."}, + "max_results": {"type": "integer", "description": "Max snippets (default: 5)."}, + }, + "required": ["query"], + }, +} + + +class ContextoMemoryProvider(MemoryProvider): + """Contexto-backed memory provider (self-hosted only).""" + + def __init__(self): + self._client = None + self._agent_slug = "hermes" + self._user_id: str | None = None + self._prefetch_result = "" + self._prefetch_lock = threading.Lock() + self._prefetch_thread: threading.Thread | None = None + self._sync_thread: threading.Thread | None = None + + @property + def name(self) -> str: + return "contexto" + + def is_available(self) -> bool: + # Selfhost has no auth, so just check the package imports. + try: + import contexto # noqa: F401 + return True + except ImportError: + return False + + def get_config_schema(self) -> List[Dict[str, Any]]: + return [ + { + "key": "base_url", + "description": "Contexto self-hosted API base URL", + "default": "http://localhost:4010", + "env_var": "CONTEXTO_BASE_URL", + }, + ] + + def initialize(self, session_id: str, **kwargs) -> None: + from contexto import ContextoClient + + base_url = os.environ.get("CONTEXTO_BASE_URL", "http://localhost:4010") + self._client = ContextoClient(base_url=base_url) + self._agent_slug = kwargs.get("agent_identity") or "hermes" + self._user_id = kwargs.get("user_id") or None + # Idempotent: register the agent slug if it doesn't exist. + self._client.register_agent(self._agent_slug, name=self._agent_slug) + + def system_prompt_block(self) -> str: + scope = f"agent={self._agent_slug}" + if self._user_id: + scope += f", user={self._user_id}" + return ( + "# Contexto Memory\n" + f"Active. Scope: {scope}.\n" + "Use contexto_search to look up prior knowledge by meaning. " + "Recent context is auto-injected before each turn." + ) + + def prefetch(self, query: str, *, session_id: str = "") -> str: + if self._prefetch_thread and self._prefetch_thread.is_alive(): + self._prefetch_thread.join(timeout=3.0) + with self._prefetch_lock: + result = self._prefetch_result + self._prefetch_result = "" + if not result: + return "" + return f"## Contexto Memory\n{result}" + + def queue_prefetch(self, query: str, *, session_id: str = "") -> None: + def _run(): + block = self._client.get_context_for_turn( + query, agent=self._agent_slug, user_id=self._user_id, max_results=5 + ) + with self._prefetch_lock: + self._prefetch_result = block + + self._prefetch_thread = threading.Thread(target=_run, daemon=True, name="contexto-prefetch") + self._prefetch_thread.start() + + def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None: + messages = [ + {"role": "user", "content": user_content}, + {"role": "assistant", "content": assistant_content}, + ] + + def _sync(): + self._client.ingest(messages, agent=self._agent_slug, user_id=self._user_id) + + if self._sync_thread and self._sync_thread.is_alive(): + # Give the previous extraction up to 60s before kicking off a new one; + # otherwise we pile up overlapping ingests on slow turns. + self._sync_thread.join(timeout=60.0) + self._sync_thread = threading.Thread(target=_sync, daemon=True, name="contexto-sync") + self._sync_thread.start() + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + return [SEARCH_SCHEMA] + + def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str: + if tool_name != "contexto_search": + return tool_error(f"Unknown tool: {tool_name}") + query = args.get("query", "") + if not query: + return tool_error("Missing required parameter: query") + max_results = min(int(args.get("max_results", 5)), 25) + result = self._client.search(query, agent=self._agent_slug, user_id=self._user_id) + items = (result or {}).get("workingMemory") or [] + snippets = [ + { + "sector": it.get("sector"), + "content": it.get("content", ""), + "score": it.get("score", 0), + } + for it in items[:max_results] + ] + return json.dumps({"results": snippets, "count": len(snippets)}) + + def shutdown(self) -> None: + for t in (self._prefetch_thread, self._sync_thread): + if t and t.is_alive(): + t.join(timeout=5.0) + + +def register(ctx) -> None: + ctx.register_memory_provider(ContextoMemoryProvider()) diff --git a/plugins/memory/contexto/plugin.yaml b/plugins/memory/contexto/plugin.yaml new file mode 100644 index 000000000000..3415bcdd6ee4 --- /dev/null +++ b/plugins/memory/contexto/plugin.yaml @@ -0,0 +1,5 @@ +name: contexto +version: 0.1.0 +description: "Self-hosted Contexto cognitive memory — episodic/semantic/procedural recall via the OSS memory engine." +pip_dependencies: + - "contexto @ git+https://github.com/amiller/contexto.git#subdirectory=clients/python" From e0735d2ada23d39e2ef5a57caf6fba2c1070c855 Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Sun, 10 May 2026 05:17:42 -0400 Subject: [PATCH 2/4] fix(memory): skip ingest in subagent contexts, capture results via on_delegation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two changes that go together: 1. Honor the MemoryProvider ABC contract: skip writes when agent_context is non-primary (subagent, cron, flush). Subagent turns are tool-output-heavy — 5-10x larger than primary turns — so ingesting them under the parent's slug both corrupts recall semantics and reliably blows the extraction timeout against Gemini Flash. Without this guard, every delegated task triggered a 120s+ Gemini extract that often timed out. 2. Implement on_delegation so the parent agent captures (task, result) pairs as a single curated turn under its own slug. Subagent raw turns are still silenced, but the work they did still reaches memory in a clean, recallable form ("[delegated subtask] ... / [delegation result] ..."). This means delegating "research X" doesn't lose the findings — they're remembered as a parent observation. Adds tests/test_plugin.py covering all branches: subagent + cron + flush silenced, primary still ingests, on_delegation captures and is recallable. Tests skip if the selfhost isn't reachable. Co-Authored-By: Claude Opus 4.7 (1M context) --- plugins/memory/contexto/__init__.py | 32 +++++- plugins/memory/contexto/tests/test_plugin.py | 107 +++++++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) create mode 100644 plugins/memory/contexto/tests/test_plugin.py diff --git a/plugins/memory/contexto/__init__.py b/plugins/memory/contexto/__init__.py index 5cca8f1be448..7523d2c8e8e5 100644 --- a/plugins/memory/contexto/__init__.py +++ b/plugins/memory/contexto/__init__.py @@ -49,6 +49,7 @@ def __init__(self): self._client = None self._agent_slug = "hermes" self._user_id: str | None = None + self._agent_context = "primary" self._prefetch_result = "" self._prefetch_lock = threading.Lock() self._prefetch_thread: threading.Thread | None = None @@ -83,8 +84,14 @@ def initialize(self, session_id: str, **kwargs) -> None: self._client = ContextoClient(base_url=base_url) self._agent_slug = kwargs.get("agent_identity") or "hermes" self._user_id = kwargs.get("user_id") or None - # Idempotent: register the agent slug if it doesn't exist. - self._client.register_agent(self._agent_slug, name=self._agent_slug) + self._agent_context = kwargs.get("agent_context") or "primary" + # Only the primary agent registers the slug and writes — subagents, + # cron, and flush share the parent's slug but must not ingest (their + # turns are tool-output-heavy and would corrupt recall + blow the + # extract timeout). Subagent RESULTS still reach memory via + # on_delegation on the parent. + if self._agent_context == "primary": + self._client.register_agent(self._agent_slug, name=self._agent_slug) def system_prompt_block(self) -> str: scope = f"agent={self._agent_slug}" @@ -108,6 +115,9 @@ def prefetch(self, query: str, *, session_id: str = "") -> str: return f"## Contexto Memory\n{result}" def queue_prefetch(self, query: str, *, session_id: str = "") -> None: + if self._agent_context != "primary": + return + def _run(): block = self._client.get_context_for_turn( query, agent=self._agent_slug, user_id=self._user_id, max_results=5 @@ -119,6 +129,9 @@ def _run(): self._prefetch_thread.start() def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None: + if self._agent_context != "primary": + return + messages = [ {"role": "user", "content": user_content}, {"role": "assistant", "content": assistant_content}, @@ -156,6 +169,21 @@ def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> st ] return json.dumps({"results": snippets, "count": len(snippets)}) + def on_delegation(self, task: str, result: str, *, + child_session_id: str = "", **kwargs) -> None: + """Capture (delegation prompt, subagent result) as one parent turn. + + Subagents are silenced for ingest (their raw turns are noise), but + the curated task→result pair represents real work done on the user's + behalf and IS worth recalling. Tagged so extraction can see the + delegation framing. + """ + if self._agent_context != "primary" or not task or not result: + return + framed_user = f"[delegated subtask] {task}" + framed_assistant = f"[delegation result] {result}" + self.sync_turn(framed_user, framed_assistant, session_id=child_session_id) + def shutdown(self) -> None: for t in (self._prefetch_thread, self._sync_thread): if t and t.is_alive(): diff --git a/plugins/memory/contexto/tests/test_plugin.py b/plugins/memory/contexto/tests/test_plugin.py new file mode 100644 index 000000000000..65153a9bcdff --- /dev/null +++ b/plugins/memory/contexto/tests/test_plugin.py @@ -0,0 +1,107 @@ +"""Regression tests for the Contexto memory provider plugin. + +Requires a running self-hosted Contexto stack on CONTEXTO_BASE_URL +(default http://localhost:4010). Skips if unreachable. + +Covers: + - subagent / cron / flush contexts must not ingest, prefetch, or register + - primary context still ingests normally + - on_delegation captures (task, result) pairs as curated parent turns + - recall finds delegation content + - nested-subagent (subagent calling on_delegation) is also no-op +""" + +from __future__ import annotations + +import os +import time +import uuid + +import httpx +import pytest + +from plugins.memory.contexto import ContextoMemoryProvider + + +_BASE = os.environ.get("CONTEXTO_BASE_URL", "http://localhost:4010") + + +def _selfhost_alive() -> bool: + try: + httpx.get(f"{_BASE}/v1/agents", timeout=2.0).raise_for_status() + return True + except Exception: + return False + + +pytestmark = pytest.mark.skipif( + not _selfhost_alive(), + reason=f"Contexto selfhost not reachable at {_BASE}", +) + + +@pytest.fixture +def slug() -> str: + return f"test-{uuid.uuid4().hex[:8]}" + + +@pytest.fixture +def primary(slug: str) -> ContextoMemoryProvider: + p = ContextoMemoryProvider() + p.initialize(session_id="prim", agent_context="primary", + agent_identity=slug, user_id="alex") + yield p + p.shutdown() + + +@pytest.fixture +def subagent(slug: str) -> ContextoMemoryProvider: + s = ContextoMemoryProvider() + s.initialize(session_id="sub", agent_context="subagent", + agent_identity=slug, user_id="alex") + yield s + s.shutdown() + + +def test_subagent_skips_all_writes(subagent: ContextoMemoryProvider): + """Subagent context must not spawn any threads or fire any HTTP.""" + subagent.queue_prefetch("anything") + subagent.sync_turn("massive subagent output " * 100, "result " * 100) + subagent.on_delegation(task="t", result="r") + assert subagent._sync_thread is None + assert subagent._prefetch_thread is None + + +@pytest.mark.parametrize("ctx", ["cron", "flush"]) +def test_other_nonprimary_contexts_skip(slug: str, ctx: str): + p = ContextoMemoryProvider() + p.initialize(session_id="x", agent_context=ctx, + agent_identity=slug, user_id="alex") + p.sync_turn("hi", "hello") + assert p._sync_thread is None + p.shutdown() + + +def test_primary_ingests(primary: ContextoMemoryProvider): + primary.sync_turn("My favorite color is octarine.", "Got it — octarine.") + assert primary._sync_thread is not None + primary._sync_thread.join(timeout=120) + assert not primary._sync_thread.is_alive() + + +def test_on_delegation_captures_subagent_result(primary: ContextoMemoryProvider, slug: str): + primary.on_delegation( + task="research how the rate limiter handles clock skew", + result="Use time.monotonic() locally and Redis TTL for cross-pod coordination.", + child_session_id="sub-session-id", + ) + primary._sync_thread.join(timeout=120) + time.sleep(2) # let server index + + result = primary._client.search( + "rate limiter clock skew", agent=slug, user_id="alex", + ) + wm = result.get("workingMemory", []) + assert wm, "delegation result should be recallable" + contents = " ".join(it.get("content", "") for it in wm) + assert "clock skew" in contents.lower() or "monotonic" in contents.lower() From 92d8a07122dff415a58a3ae4eb7fc727d266b5d7 Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Sun, 10 May 2026 05:28:55 -0400 Subject: [PATCH 3/4] fix(memory): route contexto sync/prefetch errors to logger, not stderr MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Daemon-thread errors print raw tracebacks to stderr by default. For the contexto plugin running inside an interactive hermes session, that means every transient ingest failure (Gemini extract timeout, network blip, container restart) prints a 30-line traceback right into the user's chat UI mid-stream. Worse, those tracebacks NEVER made it to agent.log, so failures were both maximally visible to the user and invisible to me when debugging. Wrap _sync and queue_prefetch._run in try/except + logger.warning. The error type and message (including the response body, thanks to _raise_with_body in the client) still surface — they just land in agent.log under "contexto sync failed" / "contexto prefetch failed" instead of stderr. Same pattern mem0 and supermemory use. This is logging, not swallowing: the agent's response is non-blocking on memory writes, the error info is preserved, and the user can grep agent.log to see what's actually happening with their ingests. Adds a regression test that captures stderr + caplog and verifies failures route through the logger only. Co-Authored-By: Claude Opus 4.7 (1M context) --- plugins/memory/contexto/__init__.py | 25 ++++++++++--- plugins/memory/contexto/tests/test_plugin.py | 39 ++++++++++++++++++++ 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/plugins/memory/contexto/__init__.py b/plugins/memory/contexto/__init__.py index 7523d2c8e8e5..84926e45005e 100644 --- a/plugins/memory/contexto/__init__.py +++ b/plugins/memory/contexto/__init__.py @@ -119,11 +119,17 @@ def queue_prefetch(self, query: str, *, session_id: str = "") -> None: return def _run(): - block = self._client.get_context_for_turn( - query, agent=self._agent_slug, user_id=self._user_id, max_results=5 - ) - with self._prefetch_lock: - self._prefetch_result = block + # Daemon-thread errors print to stderr by default and pollute the + # chat UI. Route them through the logger instead — error info still + # lands in agent.log, just not in the user's terminal mid-stream. + try: + block = self._client.get_context_for_turn( + query, agent=self._agent_slug, user_id=self._user_id, max_results=5 + ) + with self._prefetch_lock: + self._prefetch_result = block + except Exception as e: + logger.warning("contexto prefetch failed: %s: %s", type(e).__name__, e) self._prefetch_thread = threading.Thread(target=_run, daemon=True, name="contexto-prefetch") self._prefetch_thread.start() @@ -138,7 +144,14 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st ] def _sync(): - self._client.ingest(messages, agent=self._agent_slug, user_id=self._user_id) + # Same logger redirection as _run above — keep daemon-thread + # tracebacks out of the user's terminal. The error message + # (including the response body, thanks to _raise_with_body in + # the client) still surfaces in agent.log under "contexto sync". + try: + self._client.ingest(messages, agent=self._agent_slug, user_id=self._user_id) + except Exception as e: + logger.warning("contexto sync failed: %s: %s", type(e).__name__, e) if self._sync_thread and self._sync_thread.is_alive(): # Give the previous extraction up to 60s before kicking off a new one; diff --git a/plugins/memory/contexto/tests/test_plugin.py b/plugins/memory/contexto/tests/test_plugin.py index 65153a9bcdff..a56b5f3b8aa8 100644 --- a/plugins/memory/contexto/tests/test_plugin.py +++ b/plugins/memory/contexto/tests/test_plugin.py @@ -89,6 +89,45 @@ def test_primary_ingests(primary: ContextoMemoryProvider): assert not primary._sync_thread.is_alive() +def test_failures_route_to_logger_not_stderr(slug: str, caplog): + """Daemon-thread failures must land in the logger, not pollute the chat UI. + + Regression for: previously, sync_turn / queue_prefetch failures printed + raw tracebacks to stderr (visible in the user's terminal mid-chat) and + never made it to agent.log. Now they go through the logger. + """ + import io + import sys + from contexto import ContextoClient + + p = ContextoMemoryProvider() + p.initialize(session_id="t", agent_context="primary", + agent_identity=slug, user_id="alex") + # Point at a closed port to force ConnectError without touching the live host. + p._client = ContextoClient(base_url="http://127.0.0.1:9", timeout=2.0) + + captured_stderr = io.StringIO() + old_stderr = sys.stderr + sys.stderr = captured_stderr + try: + with caplog.at_level("WARNING", logger="plugins.memory.contexto"): + p.sync_turn("hi", "hello") + p._sync_thread.join(timeout=5) + p.queue_prefetch("anything") + p._prefetch_thread.join(timeout=5) + finally: + sys.stderr = old_stderr + p.shutdown() + + assert "Traceback" not in captured_stderr.getvalue(), \ + "daemon-thread traceback leaked to stderr" + msgs = [rec.getMessage() for rec in caplog.records] + assert any("contexto sync failed" in m for m in msgs), \ + f"sync failure not logged; got: {msgs}" + assert any("contexto prefetch failed" in m for m in msgs), \ + f"prefetch failure not logged; got: {msgs}" + + def test_on_delegation_captures_subagent_result(primary: ContextoMemoryProvider, slug: str): primary.on_delegation( task="research how the rate limiter handles clock skew", From c4c7b9288239dee8d56366bc332be1205e3ccfee Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Sun, 10 May 2026 09:44:26 -0400 Subject: [PATCH 4/4] fix(memory): default user_id to agent_slug when no platform user is set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hermes CLI sessions don't have a platform user_id (gateway-supplied), so the plugin was passing user_id=None to Contexto's /v1/ingest. Without a userId, the (now-merged-upstream) extractor user-identity anchoring doesn't kick in — the model produces bare "User" / "I" subjects in extracted triples, which is exactly the contamination root cause we filed ekailabs/contexto#144 for. Fall back to agent_slug as user_id when the platform doesn't supply one. The agent slug is the hermes profile name (e.g. "default", "coder"), which is the closest stable identity we have for a CLI user talking to themselves. Triples will then anchor to that identity: Before (CLI session, user_id=None): User → has favorite color → teal User → is working on → matthammer After: default → has favorite color → teal default → is working on → matthammer Gateway sessions (Telegram/Discord/etc) keep their per-platform user_id unchanged — the fallback only applies when kwargs.get("user_id") is falsy. Co-Authored-By: Claude Opus 4.7 (1M context) --- plugins/memory/contexto/__init__.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/plugins/memory/contexto/__init__.py b/plugins/memory/contexto/__init__.py index 84926e45005e..a08433387fa8 100644 --- a/plugins/memory/contexto/__init__.py +++ b/plugins/memory/contexto/__init__.py @@ -83,7 +83,13 @@ def initialize(self, session_id: str, **kwargs) -> None: base_url = os.environ.get("CONTEXTO_BASE_URL", "http://localhost:4010") self._client = ContextoClient(base_url=base_url) self._agent_slug = kwargs.get("agent_identity") or "hermes" - self._user_id = kwargs.get("user_id") or None + # Prefer the gateway-supplied platform user_id (Telegram, Discord, etc.). + # In CLI sessions there is none, so fall back to the agent slug itself — + # this gives the contexto extractor a stable identity to anchor first- + # person language to, which prevents the extractor from emitting bare + # "User" / "I" subjects (or worse, drifting toward whichever named + # entity is most salient in the conversation context). + self._user_id = kwargs.get("user_id") or self._agent_slug self._agent_context = kwargs.get("agent_context") or "primary" # Only the primary agent registers the slug and writes — subagents, # cron, and flush share the parent's slug but must not ingest (their