diff --git a/.changeset/hermes-memory-provider.md b/.changeset/hermes-memory-provider.md new file mode 100644 index 0000000..abcff22 --- /dev/null +++ b/.changeset/hermes-memory-provider.md @@ -0,0 +1,5 @@ +--- +"upstash-hermes-memory": minor +--- + +First release: an Upstash Redis memory provider for Hermes Agent. It recalls related memories into each turn, saves every session as a searchable transcript, mirrors the built-in `memory` tool, and gives the model `upstash_memory_save` / `_search` / `_forget` and `upstash_chat_search`. It reads and writes the same keys and Redis Search indexes as `AgentMemory` and `ChatHistory` in `@upstash/agentkit-sdk`, so memories are shared with AgentKit apps on the same database. diff --git a/CLAUDE.md b/CLAUDE.md index f18357c..c1fb7e8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,7 +20,7 @@ embeddings — keep that in mind when naming/among scoring. | `@upstash/agentkit-tanstack-ai` (`packages/tanstack-ai`) | TanStack AI backends: persistence stores, `StreamDurability`, `LockStore`, `MemoryAdapter`, middlewares, search tools. | | `@upstash/agentkit-eve-extension` (`packages/eve-extension`) | AgentKit as a mountable **eve extension** (eve ≥0.24): one `agent/extensions/.ts` file composes memory tools, search tools, a chat-history hook, and an instructions fragment under `__*`. | | `@upstash/mcp-toolkit` (`packages/mcp-toolkit`) | Durable building blocks for the official `@modelcontextprotocol/server` v2: `/tasks` (**long-running MCP tools** — native MCP Tasks for clients that declare the extension, else start + `task_status` + `task_cancel`) and `/events` (**MCP Events**, webhook delivery). `/tasks` and `/events` are generic (the `McpTasks` / `McpEvents` classes, interfaces, memory backends); every Upstash backend for both is in `/upstash` (no `/tasks/upstash`, no `/events/upstash`). `@upstash/redis`, `@upstash/qstash` and `@upstash/workflow` are regular dependencies (the package "just works"); only `@modelcontextprotocol/server` is a peer. Renamed from `@upstash/mcp-tasks` before its first release. **Not an `agentkit-*` package** — separate name, versioned independently (the changesets `linked` glob only covers `@upstash/agentkit-*`), and it depends on none of the others. | -| `upstash-hermes-memory` (`python/hermes-memory`, **PyPI**) | Hermes Agent memory provider on Redis Search. **Scaffold at `0.0.0`** (unreleased) until the port lands (DX-3182). See **Python packages** below. | +| `upstash-hermes-memory` (`python/hermes-memory`, **PyPI**) | Hermes Agent memory provider (`UpstashMemoryProvider`, entry point `hermes_agent.memory_providers: upstash`) plus Python ports of `AgentMemory` / `ChatHistory` that write the TS keys and indexes byte for byte. See **Python packages** below. | Examples (`examples/`): `ai-sdk-demo` (hand-written Next.js), `eve-demo` (a real `eve` CLI scaffold), `eve-extension-demo` (a minimal eve scaffold that mounts the extension), and `mcp-toolkit-demo` @@ -1434,6 +1434,11 @@ Python lives **beside** the pnpm workspace, not instead of any of it: non-zero version merges. Nothing else to touch. - Python tests share the test Redis (10-index cap) with vitest: run them **after** the Node tests, never in parallel, and reuse the TS key prefixes/index names when the data is meant to be shared. + Live tests take a unique `prefix` fixture and drop their indexes; call `index.wait_indexing()` + after a write or delete before asserting on search results (`SEARCH.COUNT` lags otherwise). +- `upstash-hermes-memory` e2e: `uv run python python/hermes-memory/e2e/hermes_e2e.py` drives a real + Hermes (temporary `HERMES_HOME`, package on `PYTHONPATH`, `gpt-5-mini` via `OPENAI_API_KEY`; Hermes + sends `reasoning.effort`, so non-reasoning models like `gpt-4o-mini` fail) and checks Redis. ## Commands - **Setup gotcha — use pnpm 11.5.3, matching the `packageManager` pin.** Root `package.json` pins diff --git a/README.md b/README.md index 83f10b0..766cf62 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ are powered by [Upstash Redis Search](https://upstash.com/docs/redis/search/intr | [`@upstash/agentkit-tanstack-ai`](./packages/tanstack-ai) | Production backends for [TanStack AI](https://tanstack.com/ai) — chat persistence, resumable streams, distributed locks, memory, tool caching, rate limiting and search tools. | | [`@upstash/agentkit-eve-extension`](./packages/eve-extension) | The same capabilities as a mountable [Eve extension](https://eve.dev/docs/extensions) — one file in `agent/extensions/` adds memory tools, search tools, and durable chat history the agent can search. | | [`@upstash/mcp-toolkit`](./packages/mcp-toolkit) | Durable building blocks for MCP servers on the official TypeScript SDK. `/tasks`: long-running tools that answer at once, as a native MCP task for clients with the Tasks extension (Claude Code) or a task id the model polls for everyone else, with the record in Redis and the work on QStash or Workflow. `/events`: MCP Events, with subscriptions in Redis and signed webhook deliveries retried by QStash. | -| [`upstash-hermes-memory`](./python/hermes-memory) (PyPI) | Memory provider for [Hermes Agent](https://hermes-agent.nousresearch.com) on Upstash Redis Search. Work in progress. | +| [`upstash-hermes-memory`](./python/hermes-memory) (PyPI) | Memory provider for [Hermes Agent](https://hermes-agent.nousresearch.com) on Upstash Redis Search, sharing memories with `@upstash/agentkit-sdk`. | ## Core features diff --git a/python/hermes-memory/README.md b/python/hermes-memory/README.md index a66444e..3cdfa52 100644 --- a/python/hermes-memory/README.md +++ b/python/hermes-memory/README.md @@ -1,7 +1,65 @@ # upstash-hermes-memory An Upstash Redis memory provider for [Hermes Agent](https://hermes-agent.nousresearch.com), built on -the same Redis Search layout as `@upstash/agentkit-sdk`'s `AgentMemory` and `ChatHistory`. +the same Redis Search layout as `@upstash/agentkit-sdk`'s `AgentMemory` and `ChatHistory`. A memory +your AI SDK app saves is recalled by your Hermes agent, and the other way round. -**Status: scaffold.** The package is at `0.0.0`, which `scripts/publish-python.mjs` treats as -unreleased, so nothing is published to PyPI yet. The provider itself is tracked in DX-3182. +## What it does in Hermes + +- **Recall:** before each turn, the memories closest to the user's message are added to the prompt. +- **Chat history:** after each turn, the session's transcript is saved as a searchable document. +- **Mirroring:** facts the built-in `memory` tool adds are also saved to Redis. +- **Tools:** `upstash_memory_save`, `upstash_memory_search`, `upstash_memory_forget` and + `upstash_chat_search` (search past conversations). + +Subagents, cron jobs and context flushes can read memory but never write it. + +## Setup + +```bash +# into the Python environment Hermes runs in +pip install upstash-hermes-memory +``` + +Then select it in Hermes's `config.yaml`: + +```yaml +memory: + provider: upstash +``` + +And give it a database, in your environment or `~/.hermes/.env` (`hermes memory setup` asks for +these): + +| Variable | | +| --- | --- | +| `UPSTASH_REDIS_REST_URL`, `UPSTASH_REDIS_REST_TOKEN` | Required. From the [Upstash console](https://console.upstash.com/redis). | +| `UPSTASH_HERMES_USER_ID` | Whose memories these are. Defaults to the gateway user, then the Hermes profile name, then `default`. | +| `UPSTASH_HERMES_MEMORY_PREFIX`, `UPSTASH_HERMES_CHAT_PREFIX` | Key prefixes. Default `agentkit:memory` and `agentkit:chat`, the TS SDK's defaults, so the two share memories. | + +## Redis layout + +| | Key | Index | +| --- | --- | --- | +| Memory | `agentkit:memory::` → JSON `{text, userId, createdAt, ...metadata}` | `agentkit_memory` (`text`, `userId`) | +| Chat | `agentkit:chat::` → JSON `{userId, sessionId, userMessages, modelMessages, messages, createdAt, updatedAt}` | `agentkit_chat` | + +User and session ids are used as they are, so the same id reads the same keys from Hermes and the +TS SDK. Like the TS SDK, the provider refuses a user id with `:` (the key separator), and keeps no +transcript for a session id with one. Indexes are created on first use. `AgentMemory` and `ChatHistory` are importable on their own for +use outside Hermes. + +## Development + +```bash +uv sync --all-packages && pnpm py:lint && pnpm py:test # Redis tests need UPSTASH_REDIS_REST_* set +uv run python python/hermes-memory/e2e/hermes_e2e.py # a real Hermes session, needs OPENAI_API_KEY +``` + +The e2e script installs the package into a temporary directory and runs Hermes with a temporary +`HERMES_HOME`, so your own Hermes install is not touched. Hermes's built-in memory is turned off +there. One session saves two facts; a fresh session started right after must name them without +calling any tool (checked in Hermes's `state.db`), so they can only come from the automatic recall; +and a control session as another user must not know them. The script then checks the keys and +transcripts in Redis (and, if `packages/sdk` is built, that the TS `AgentMemory` recalls them too) +before cleaning up. diff --git a/python/hermes-memory/e2e/hermes_e2e.py b/python/hermes-memory/e2e/hermes_e2e.py new file mode 100644 index 0000000..9f21caf --- /dev/null +++ b/python/hermes-memory/e2e/hermes_e2e.py @@ -0,0 +1,296 @@ +"""End-to-end check of the provider inside a real Hermes Agent, against a real Upstash Redis. + + uv run python python/hermes-memory/e2e/hermes_e2e.py [--keep] [--model gpt-5-mini] + +What it does: +1. Builds the wheel and installs it (plus upstash-redis) into a temporary directory that is put on + Hermes's PYTHONPATH. Hermes's own install and ~/.hermes are not touched: Hermes runs with a + temporary HERMES_HOME whose config.yaml selects `memory.provider: upstash`. +2. Session 1 asks the agent to remember two facts. Session 2, a fresh session started right away, + asks for them back. Hermes's built-in memory is off, and session 2 must answer without calling + a tool (read from Hermes's state.db), so the facts can only come from the provider's automatic + recall. A control session as another user must not know them. +3. Checks Redis: the memories and both transcripts landed under a unique prefix, in the documents + and indexes the TS SDK uses. If packages/sdk is built, the TS AgentMemory recalls the same + memory too. +4. Drops the e2e indexes and deletes the keys (unless --keep). + +Needs: Hermes installed (`hermes` on PATH), UPSTASH_REDIS_REST_URL / UPSTASH_REDIS_REST_TOKEN and +OPENAI_API_KEY in the environment or the repo's .env. +""" + +import argparse +import json +import os +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +import time +import uuid +from pathlib import Path + +REPO = Path(__file__).resolve().parents[3] +PACKAGE = REPO / "python" / "hermes-memory" +USER_ID = "e2e-user" +FACT_PROMPT = ( + "Please remember this for future conversations: my cat is called Pamuk and my favourite " + "programming language is Elixir. Save it to your long-term memory." +) +RECALL_PROMPT = ( + "What is my cat called, and what is my favourite programming language? " + "Answer from what you remember about me, in one sentence." +) + + +def load_dotenv(path: Path) -> None: + if not path.exists(): + return + for line in path.read_text().splitlines(): + if "=" in line and not line.lstrip().startswith("#"): + key, value = line.split("=", 1) + os.environ.setdefault(key.strip(), value.strip().strip('"')) + + +def hermes_python() -> list[str]: + """Hermes's own interpreter and entry script. Its launcher unsets PYTHONPATH, so skip it.""" + launcher = shutil.which("hermes") + if not launcher: + sys.exit("hermes is not on PATH: install Hermes Agent first") + home = Path(os.environ.get("HERMES_INSTALL_DIR", Path.home() / ".hermes" / "hermes-agent")) + python, script = home / "venv" / "bin" / "python", home / "hermes" + if not (python.exists() and script.exists()): + sys.exit(f"cannot find Hermes's venv under {home} (set HERMES_INSTALL_DIR)") + return [str(python), str(script)] + + +def step(title: str) -> None: + print(f"\n== {title} ==", flush=True) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--model", default="gpt-5-mini") + parser.add_argument("--keep", action="store_true", help="leave the keys and indexes in Redis") + args = parser.parse_args() + + load_dotenv(REPO / ".env") + for name in ("UPSTASH_REDIS_REST_URL", "UPSTASH_REDIS_REST_TOKEN", "OPENAI_API_KEY"): + if not os.environ.get(name): + sys.exit(f"{name} is not set") + + from upstash_hermes_memory import AgentMemory, ChatHistory + from upstash_redis import Redis + + run = uuid.uuid4().hex[:8] + memory_prefix, chat_prefix = f"e2e:hermes:{run}:memory", f"e2e:hermes:{run}:chat" + redis = Redis.from_env() + memory = AgentMemory(redis, prefix=memory_prefix, enable_telemetry=False) + chats = ChatHistory(redis, prefix=chat_prefix, enable_telemetry=False) + work = Path(tempfile.mkdtemp(prefix="hermes-e2e-")) + failures: list[str] = [] + + def check(ok: bool, what: str) -> None: + print(f" {'ok ' if ok else 'FAIL'} {what}") + if not ok: + failures.append(what) + + try: + step("install the provider next to Hermes (temporary)") + dist, site, home = work / "dist", work / "site", work / "home" + subprocess.run( + ["uv", "build", "--wheel", "--package", "upstash-hermes-memory", "--out-dir", dist], + cwd=REPO, + check=True, + capture_output=True, + ) + (wheel,) = dist.glob("*.whl") + # --no-deps: httpx and friends come from Hermes's venv, only upstash-redis is missing there. + subprocess.run( + [ + "uv", + "pip", + "install", + "--quiet", + "--no-deps", + "--target", + site, + wheel, + "upstash-redis==1.8.0", + ], + check=True, + ) + home.mkdir() + # Hermes's own stores (MEMORY.md / USER.md) default to on. Off, they cannot carry the + # facts from session 1 to session 2: the provider is the only way across. + (home / "config.yaml").write_text( + f"model:\n provider: openai-api\n default: {args.model}\n" + "memory:\n provider: upstash\n memory_enabled: false\n user_profile_enabled: false\n" + ) + env = { + **os.environ, + "HERMES_HOME": str(home), + "PYTHONPATH": str(site), + "UPSTASH_HERMES_MEMORY_PREFIX": memory_prefix, + "UPSTASH_HERMES_CHAT_PREFIX": chat_prefix, + } + print(f" HERMES_HOME={home}\n prefixes {memory_prefix} / {chat_prefix}") + + def hermes( + prompt: str, user: str = USER_ID, hermes_home: Path = home + ) -> tuple[str, list[str]]: + """Run one fresh one-shot session; return its answer and the tools it called.""" + cmd = [*hermes_python(), "-z", prompt, "--provider", "openai-api", "-m", args.model] + done = subprocess.run( + cmd, + check=False, + env={**env, "HERMES_HOME": str(hermes_home), "UPSTASH_HERMES_USER_ID": user}, + capture_output=True, + text=True, + timeout=300, + ) + if done.returncode != 0: + print(done.stdout, done.stderr, sep="\n") + raise SystemExit(f"hermes exited with {done.returncode}") + # Hermes records every session in $HERMES_HOME/state.db, tool results included. + with sqlite3.connect(hermes_home / "state.db") as db: + (session,) = db.execute( + "SELECT session_id FROM messages WHERE role = 'user' AND content = ?" + " ORDER BY id DESC LIMIT 1", + (prompt,), + ).fetchone() + tools = [ + row[0] + for row in db.execute( + "SELECT tool_name FROM messages WHERE session_id = ? AND role = 'tool'" + " ORDER BY id", + (session,), + ) + ] + answer = done.stdout.strip() + print(" < " + answer.replace("\n", "\n ")) + print(f" tools called: {tools or 'none'}") + return answer, tools + + # The sessions run back to back, with no waiting in between: a fact the agent saved must + # be recallable as soon as the save tool returns. + step("session 1: tell the agent two facts") + print(" > " + FACT_PROMPT) + _, tools_1 = hermes(FACT_PROMPT) + check( + tools_1.count("upstash_memory_save") >= 1, "session 1 saved through upstash_memory_save" + ) + check("memory" not in tools_1, "session 1 did not use Hermes's built-in memory tool") + + step("session 2: a fresh session asks for them back") + print(" > " + RECALL_PROMPT) + answer, tools_2 = hermes(RECALL_PROMPT) + check("Pamuk" in answer, "the answer names the cat (Pamuk)") + check("Elixir" in answer, "the answer names the language (Elixir)") + # With no tool calls, the facts can only have come from the automatic recall (prefetch). + check(tools_2 == [], "session 2 called no tools: the facts came from automatic recall") + + step("control: the same question as a user with no memories") + print(" > " + RECALL_PROMPT) + # Its own HERMES_HOME: Hermes's built-in session_search reads the sessions in state.db, + # whoever they belonged to, so a shared home would let it find sessions 1 and 2. This + # checks the Redis side only, which is what the provider controls. + control_home = work / "control-home" + control_home.mkdir() + shutil.copy(home / "config.yaml", control_home / "config.yaml") + control, _ = hermes(RECALL_PROMPT, user=f"{USER_ID}-control", hermes_home=control_home) + check("Pamuk" not in control, "another user's session does not know the cat's name") + + step("Hermes's built-in stores stayed empty") + builtin = [ + p for p in (home / "memories").glob("*") if p.is_file() and p.read_text().strip() + ] + print(f" {home / 'memories'}: {[p.name for p in builtin] or 'empty'}") + check(not builtin, "MEMORY.md / USER.md hold nothing") + + step("Redis: memories") + memory.index.wait_indexing() + saved = memory.list(user_id=USER_ID) + for record in saved: + print( + f" memory {memory_prefix}:{USER_ID}:{record.id} -> {record.text!r} {record.metadata}" + ) + text = " ".join(r.text for r in saved) + check("Pamuk" in text, "a memory mentioning Pamuk was saved") + check("Elixir" in text, "a memory mentioning Elixir was saved") + + step("Redis: chat history") + chats.index.wait_indexing() + sessions = chats.list_chats(user_id=USER_ID) + for doc in sessions: + roles = [m["role"] for m in doc["messages"]] + print(f" chat {chat_prefix}:{USER_ID}:{doc['sessionId']} messages={roles}") + check(len(sessions) == 2, "both sessions were saved as transcripts") + check( + any("Pamuk" in d.get("userMessages", "") for d in sessions), + "session 1's transcript holds the user's message", + ) + hits = chats.search_chats("cat Pamuk", user_id=USER_ID) + check(bool(hits), "the transcripts are searchable (search_chats 'cat Pamuk')") + + step("Redis: layout shared with the TS SDK") + key = f"{memory_prefix}:{USER_ID}:{saved[0].id}" if saved else "" + doc = redis.execute(["JSON.GET", key]) if key else None + print(f" JSON.GET {key} -> {json.dumps(doc)}") + check( + isinstance(doc, dict) and {"text", "userId", "createdAt"} <= set(doc), + "memory documents have the AgentMemory shape", + ) + ts_dist = REPO / "packages" / "sdk" / "dist" / "index.js" + if ts_dist.exists() and shutil.which("node"): + script = ( + "const { AgentMemory } = await import(process.argv[1]);" + "const { Redis } = await import('@upstash/redis');" + "const m = new AgentMemory({ redis: Redis.fromEnv(), prefix: process.argv[2]," + " enableTelemetry: false });" + "const hits = await m.recall({ query: 'cat Pamuk', userId: process.argv[3] });" + "console.log(JSON.stringify(hits.map((h) => h.text)));" + ) + out = subprocess.run( + ["node", "--input-type=module", "-e", script, str(ts_dist), memory_prefix, USER_ID], + check=False, + cwd=REPO / "packages" / "sdk", + capture_output=True, + text=True, + ) + print(f" TS AgentMemory.recall -> {out.stdout.strip() or out.stderr.strip()[:300]}") + check("Pamuk" in out.stdout, "the TS SDK recalls the memory Hermes saved") + else: + print(" (skipped: build packages/sdk to cross-check with the TS SDK)") + finally: + if args.keep: + print( + f"\n--keep: left {memory_prefix}* and {chat_prefix}* in Redis, HERMES_HOME {work}" + ) + else: + for index in (memory.index, chats.index): + try: + index.drop() + except Exception as error: # noqa: BLE001 - cleanup is best effort + print(f" could not drop {index.name}: {error}") + for prefix in (memory_prefix, chat_prefix): + cursor = 0 + while True: + cursor, keys = redis.scan(cursor, match=f"{prefix}*", count=200) + if keys: + redis.delete(*keys) + if int(cursor) == 0: + break + shutil.rmtree(work, ignore_errors=True) + + print(f"\n{'E2E PASSED' if not failures else 'E2E FAILED: ' + '; '.join(failures)}") + sys.exit(1 if failures else 0) + + +if __name__ == "__main__": + start = time.monotonic() + try: + main() + finally: + print(f"({time.monotonic() - start:.0f}s)") diff --git a/python/hermes-memory/pyproject.toml b/python/hermes-memory/pyproject.toml index d3f41ab..9d7d621 100644 --- a/python/hermes-memory/pyproject.toml +++ b/python/hermes-memory/pyproject.toml @@ -7,6 +7,11 @@ license = "MIT" requires-python = ">=3.11" dependencies = ["upstash-redis>=1.8.0"] +# Hermes discovers pip-installed memory providers through this group; the entry-point name is the +# value of `memory.provider` in Hermes's config.yaml. +[project.entry-points."hermes_agent.memory_providers"] +upstash = "upstash_hermes_memory:register" + [project.urls] Homepage = "https://github.com/upstash/agentkit" Repository = "https://github.com/upstash/agentkit" diff --git a/python/hermes-memory/src/upstash_hermes_memory/__init__.py b/python/hermes-memory/src/upstash_hermes_memory/__init__.py index b4a85bf..86b15f2 100644 --- a/python/hermes-memory/src/upstash_hermes_memory/__init__.py +++ b/python/hermes-memory/src/upstash_hermes_memory/__init__.py @@ -8,5 +8,20 @@ """ from ._version import VERSION +from .chat_history import ChatHistory +from .memory import AgentMemory, MemoryRecord +from .provider import UpstashMemoryProvider -__all__ = ["VERSION"] +__all__ = [ + "VERSION", + "AgentMemory", + "ChatHistory", + "MemoryRecord", + "UpstashMemoryProvider", + "register", +] + + +def register(ctx) -> None: + """Hermes plugin entry point (`hermes_agent.memory_providers`).""" + ctx.register_memory_provider(UpstashMemoryProvider()) diff --git a/python/hermes-memory/src/upstash_hermes_memory/_search.py b/python/hermes-memory/src/upstash_hermes_memory/_search.py new file mode 100644 index 0000000..231fb35 --- /dev/null +++ b/python/hermes-memory/src/upstash_hermes_memory/_search.py @@ -0,0 +1,91 @@ +"""A Redis Search index that is created on first use, shared by AgentMemory and ChatHistory. + +The commands are the ones `ReactiveSearchIndex` sends from the TS SDK, argument for argument, so +an index created from either language is the same index. +""" + +import json +import threading +from collections.abc import Sequence +from typing import Any + +# (field, type, no_tokenize) +Field = tuple[str, str, bool] + + +def index_name_for(prefix: str) -> str: + """The TS default: every character outside [a-zA-Z0-9_] becomes `_`.""" + return "".join(c if c.isascii() and (c.isalnum() or c == "_") else "_" for c in prefix) + + +def dumps(value: Any) -> str: + """JSON the way JSON.stringify writes it, so stored documents match the TS SDK's.""" + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +class SearchIndex: + def __init__(self, redis: Any, *, name: str, prefix: str, schema: Sequence[Field]): + self.redis = redis + self.name = name + self.prefix = prefix + self.schema = list(schema) + self._ready = False + self._lock = threading.Lock() + + def create_command(self) -> list[str]: + cmd = ["SEARCH.CREATE", self.name, "EXISTSOK", "ON", "JSON", "PREFIX", "1", self.prefix] + cmd.append("SCHEMA") + for field, type_, no_tokenize in self.schema: + cmd += [field, type_] + (["NOTOKENIZE"] if no_tokenize else []) + return cmd + + def ensure(self) -> None: + """Create the index if it does not exist yet. A no-op after the first call.""" + if self._ready: + return + with self._lock: + if self._ready: + return + self.redis.execute(self.create_command()) + self.redis.execute(["SEARCH.WAITINDEXING", self.name]) + self._ready = True + + def query(self, filter: dict[str, Any], limit: int | None = None) -> list[Any]: + self.ensure() + rows = self._query(filter, limit) + if not rows and self._raw_count(filter) < 0: + # The Python client turns a missing index into [], so an empty result is ambiguous. + # COUNT says -1 when the index is gone (another client dropped it): recreate, retry once. + self._ready = False + self.ensure() + rows = self._query(filter, limit) + return rows + + def count(self, filter: dict[str, Any]) -> int: + self.ensure() + count = self._raw_count(filter) + if count < 0: + # -1 means the index is gone (dropped after we created it): recreate it and retry once. + self._ready = False + self.ensure() + count = self._raw_count(filter) + return max(count, 0) + + def _query(self, filter: dict[str, Any], limit: int | None) -> list[Any]: + cmd = ["SEARCH.QUERY", self.name, dumps(filter)] + if limit is not None: + cmd += ["LIMIT", str(limit)] + return list(self.redis.execute(cmd) or []) + + def _raw_count(self, filter: dict[str, Any]) -> int: + result = self.redis.execute(["SEARCH.COUNT", self.name, dumps(filter)]) + return int(getattr(result, "count", result)) + + def wait_indexing(self) -> None: + """Block until every write so far is searchable (and every delete has left the index).""" + self.ensure() + self.redis.execute(["SEARCH.WAITINDEXING", self.name]) + + def drop(self) -> None: + self.redis.execute(["SEARCH.DROP", self.name]) + self._ready = False diff --git a/python/hermes-memory/src/upstash_hermes_memory/_telemetry.py b/python/hermes-memory/src/upstash_hermes_memory/_telemetry.py new file mode 100644 index 0000000..4a19fc2 --- /dev/null +++ b/python/hermes-memory/src/upstash_hermes_memory/_telemetry.py @@ -0,0 +1,27 @@ +"""Tags the redis client's telemetry header, the Python counterpart of the TS SDK's addTelemetry.""" + +import os +from typing import Any + +from ._version import VERSION + +SDK_TELEMETRY = f"upstash-hermes-memory@{VERSION}" + +_HEADER = "Upstash-Telemetry-Sdk" + + +def add_telemetry(redis: Any, *, enabled: bool = True) -> None: + """Append this package's tag to the `Upstash-Telemetry-Sdk` header, once per client. + + upstash-redis only sends the header when the client allows telemetry, so a client created + with `allow_telemetry=False` stays untagged, as does any process with + `UPSTASH_DISABLE_TELEMETRY` set. + """ + if not enabled or os.environ.get("UPSTASH_DISABLE_TELEMETRY"): + return + headers = getattr(redis, "_headers", None) + if not isinstance(headers, dict) or _HEADER not in headers: + return + tags = headers[_HEADER].split(",") + if SDK_TELEMETRY not in tags: + headers[_HEADER] = ",".join([*tags, SDK_TELEMETRY]) diff --git a/python/hermes-memory/src/upstash_hermes_memory/chat_history.py b/python/hermes-memory/src/upstash_hermes_memory/chat_history.py new file mode 100644 index 0000000..861001d --- /dev/null +++ b/python/hermes-memory/src/upstash_hermes_memory/chat_history.py @@ -0,0 +1,170 @@ +"""Python port of `ChatHistory` from `@upstash/agentkit-sdk` (packages/sdk/src/chat-history.ts). + + key :: + value JSON {"userId", "sessionId", "userMessages", "modelMessages", "messages", + "createdAt", "updatedAt" (epoch ms), "title"?} + index SEARCH.CREATE EXISTSOK ON JSON PREFIX 1 : SCHEMA + userId TEXT NOTOKENIZE sessionId TEXT NOTOKENIZE userMessages TEXT modelMessages TEXT + +`userMessages` / `modelMessages` are the searchable text of each side, extracted the same way as +the TS default extractor; `messages` keeps the raw transcript and is not indexed. +""" + +import time +from dataclasses import dataclass +from typing import Any, Literal + +from ._search import SearchIndex, dumps, index_name_for +from ._telemetry import add_telemetry +from .memory import _check_id + +DEFAULT_PREFIX = "agentkit:chat" + + +def extract_text(messages: list[dict[str, Any]]) -> dict[str, str]: + """Split a transcript into searchable user and model text (the TS `defaultExtractText`).""" + user: list[str] = [] + model: list[str] = [] + for message in messages: + parts = message.get("parts") + if isinstance(parts, list): + text = " ".join( + p["text"] + for p in parts + if isinstance(p, dict) + and p.get("type") == "text" + and isinstance(p.get("text"), str) + ).strip() + elif isinstance(message.get("content"), str): + text = message["content"].strip() + else: + text = "" + if not text: + continue + if message.get("role") == "user": + user.append(text) + elif message.get("role") in ("assistant", "model"): + model.append(text) + return {"userMessages": "\n".join(user), "modelMessages": "\n".join(model)} + + +@dataclass +class ChatSearchHit: + session_id: str + score: float + title: str | None + updated_at: int + user_text: str = "" + model_text: str = "" + + +class ChatHistory: + def __init__( + self, + redis: Any, + *, + prefix: str = DEFAULT_PREFIX, + index_name: str | None = None, + ttl_seconds: int | None = None, + min_score: float = 0, + enable_telemetry: bool = True, + ): + self.redis = redis + self.prefix = prefix + self.ttl_seconds = ttl_seconds + self.min_score = min_score + self.index = SearchIndex( + redis, + name=index_name or index_name_for(prefix), + prefix=f"{prefix}:", + schema=[ + ("userId", "TEXT", True), + ("sessionId", "TEXT", True), + ("userMessages", "TEXT", False), + ("modelMessages", "TEXT", False), + ], + ) + add_telemetry(redis, enabled=enable_telemetry) + + def _key(self, user_id: str, session_id: str) -> str: + return f"{self.prefix}:{user_id}:{session_id}" + + def get_chat(self, session_id: str, *, user_id: str) -> dict[str, Any] | None: + _check_id(user_id, "user_id") + _check_id(session_id, "session_id") + found = self.redis.execute(["JSON.GET", self._key(user_id, session_id), "$"]) + return found[0] if isinstance(found, list) and found else None + + def save_chat( + self, + session_id: str, + messages: list[dict[str, Any]], + *, + user_id: str, + title: str | None = None, + ) -> dict[str, Any]: + """Write the whole transcript of a session, keeping `createdAt` (and `title`) of a prior save.""" + existing = self.get_chat(session_id, user_id=user_id) or {} + now = int(time.time() * 1000) + doc: dict[str, Any] = { + "userId": user_id, + "sessionId": session_id, + **extract_text(messages), + "messages": messages, + "createdAt": existing.get("createdAt") or now, + "updatedAt": now, + } + title = title or existing.get("title") + if title: + doc["title"] = title + key = self._key(user_id, session_id) + self.redis.execute(["JSON.SET", key, "$", dumps(doc)]) + # Any TTL that was set is applied, 0 included (expire now), as the TS ChatHistory does. + if self.ttl_seconds is not None: + self.redis.expire(key, self.ttl_seconds) + return doc + + def list_chats(self, *, user_id: str, limit: int | None = None) -> list[dict[str, Any]]: + """Chats of a user, most recently updated first.""" + _check_id(user_id, "user_id") + rows = self.index.query({"userId": {"$eq": user_id}}, limit=limit) + docs = [row.data for row in rows if isinstance(row.data, dict)] + return sorted(docs, key=lambda d: d.get("updatedAt") or 0, reverse=True) + + def search_chats( + self, + query: str, + *, + user_id: str, + target: Literal["user", "model", "both"] = "both", + limit: int | None = None, + ) -> list[ChatSearchHit]: + _check_id(user_id, "user_id") + if target == "both": + text: dict[str, Any] = { + "$or": [{"userMessages": {"$smart": query}}, {"modelMessages": {"$smart": query}}] + } + else: + text = {f"{target}Messages": {"$smart": query}} + rows = self.index.query({"$and": [{"userId": {"$eq": user_id}}, text]}, limit=limit) + hits = [] + for row in rows: + data = row.data if isinstance(row.data, dict) else {} + if row.score < self.min_score: + continue + hits.append( + ChatSearchHit( + session_id=data.get("sessionId") or row.key.rsplit(":", 1)[-1], + score=row.score, + title=data.get("title"), + updated_at=data.get("updatedAt") or 0, + user_text=data.get("userMessages") or "", + model_text=data.get("modelMessages") or "", + ) + ) + return hits + + def delete_chat(self, session_id: str, *, user_id: str) -> bool: + _check_id(user_id, "user_id") + _check_id(session_id, "session_id") + return bool(self.redis.delete(self._key(user_id, session_id))) diff --git a/python/hermes-memory/src/upstash_hermes_memory/memory.py b/python/hermes-memory/src/upstash_hermes_memory/memory.py new file mode 100644 index 0000000..8462f67 --- /dev/null +++ b/python/hermes-memory/src/upstash_hermes_memory/memory.py @@ -0,0 +1,139 @@ +"""Python port of `AgentMemory` from `@upstash/agentkit-sdk` (packages/sdk/src/memory.ts). + +Same keys, same documents, same index: a memory saved by the TS SDK is recalled here and the other +way round, as long as both sides use the same prefix. + + key :: + value JSON {"text", "userId", "createdAt" (epoch ms), ...metadata} + index SEARCH.CREATE EXISTSOK ON JSON PREFIX 1 : SCHEMA text TEXT userId TEXT NOTOKENIZE +""" + +import time +import uuid +from dataclasses import dataclass, field +from typing import Any + +from ._search import SearchIndex, dumps, index_name_for +from ._telemetry import add_telemetry + +DEFAULT_PREFIX = "agentkit:memory" + +# The document's own fields. Metadata is spread next to them, so it must not replace them: a +# `userId` in metadata would move the memory into another user's searches. +RESERVED_FIELDS = frozenset({"text", "userId", "createdAt"}) + + +def _check_id(value: str, what: str) -> None: + """User and session ids sit in the middle of the key, so they may not contain `:`.""" + if not value or ":" in value: + raise ValueError(f"{what} must be a non-empty string without ':' (got {value!r})") + + +def _check_memory_id(value: str) -> None: + """The memory id is the last part of the key, so `:` is fine there (the TS SDK allows it).""" + if not value: + raise ValueError("id must be a non-empty string") + + +@dataclass +class MemoryRecord: + id: str + text: str + created_at: int + score: float | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + +class AgentMemory: + def __init__( + self, + redis: Any, + *, + prefix: str = DEFAULT_PREFIX, + index_name: str | None = None, + min_score: float = 0, + enable_telemetry: bool = True, + ): + self.redis = redis + self.prefix = prefix + self.min_score = min_score + self.index = SearchIndex( + redis, + name=index_name or index_name_for(prefix), + prefix=f"{prefix}:", + schema=[("text", "TEXT", False), ("userId", "TEXT", True)], + ) + add_telemetry(redis, enabled=enable_telemetry) + + def _key(self, user_id: str, id: str) -> str: + return f"{self.prefix}:{user_id}:{id}" + + def add( + self, + text: str, + *, + user_id: str, + id: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> MemoryRecord: + _check_id(user_id, "user_id") + id = id or str(uuid.uuid4()) + _check_memory_id(id) + reserved = RESERVED_FIELDS & set(metadata or {}) + if reserved: + raise ValueError(f"metadata may not set {sorted(reserved)}") + created_at = int(time.time() * 1000) + doc = {"text": text, "userId": user_id, "createdAt": created_at, **(metadata or {})} + self.redis.execute(["JSON.SET", self._key(user_id, id), "$", dumps(doc)]) + return MemoryRecord(id=id, text=text, created_at=created_at, metadata=metadata or {}) + + def recall(self, query: str, *, user_id: str, top_k: int = 5) -> list[MemoryRecord]: + """The `top_k` memories closest to `query` ($smart: full-text plus fuzzy matching).""" + _check_id(user_id, "user_id") + filter: dict[str, Any] = {"userId": {"$eq": user_id}} + has_query = bool(query and query.strip()) + if has_query: + filter["text"] = {"$smart": query} + rows = self.index.query(filter, limit=top_k) + min_score = self.min_score if has_query else 0 + return [ + r for r in (self._record(row, user_id) for row in rows) if (r.score or 0) >= min_score + ] + + def list(self, *, user_id: str, limit: int = 100) -> list[MemoryRecord]: + _check_id(user_id, "user_id") + rows = self.index.query({"userId": {"$eq": user_id}}, limit=limit) + return [self._record(row, user_id) for row in rows] + + def count(self, *, user_id: str) -> int: + _check_id(user_id, "user_id") + return self.index.count({"userId": {"$eq": user_id}}) + + def get(self, id: str, *, user_id: str) -> MemoryRecord | None: + _check_id(user_id, "user_id") + _check_memory_id(id) + doc = self.redis.execute(["JSON.GET", self._key(user_id, id)]) + if not isinstance(doc, dict): + return None + return self._from_doc(id, doc) + + def forget(self, id: str, *, user_id: str) -> bool: + _check_id(user_id, "user_id") + _check_memory_id(id) + return bool(self.redis.delete(self._key(user_id, id))) + + def _record(self, row: Any, user_id: str) -> MemoryRecord: + id = row.key[len(self._key(user_id, "")) :] + record = self._from_doc(id, row.data if isinstance(row.data, dict) else {}) + record.score = row.score + return record + + @staticmethod + def _from_doc(id: str, doc: dict[str, Any]) -> MemoryRecord: + metadata = {k: v for k, v in doc.items() if k not in RESERVED_FIELDS} + return MemoryRecord( + id=id, + text=doc.get("text") or "", + created_at=doc.get("createdAt") or 0, + metadata=metadata, + ) diff --git a/python/hermes-memory/src/upstash_hermes_memory/provider.py b/python/hermes-memory/src/upstash_hermes_memory/provider.py new file mode 100644 index 0000000..c1e43c3 --- /dev/null +++ b/python/hermes-memory/src/upstash_hermes_memory/provider.py @@ -0,0 +1,456 @@ +"""The Hermes Agent memory provider: long-term memory and searchable chat history on Upstash Redis. + +Enable it with `memory.provider: upstash` in Hermes's config.yaml. Hermes finds it through the +`hermes_agent.memory_providers` entry point this package declares. + +What it does inside Hermes: +- Before each turn, recalls the memories closest to the user's message into the prompt. +- After each turn, saves the session's transcript as a searchable ChatHistory document. +- Mirrors facts the built-in `memory` tool adds, so they are searchable in Redis too. +- Gives the model four tools: upstash_memory_save / _search / _forget and upstash_chat_search. + +Configuration (environment, or Hermes's `$HERMES_HOME/.env`): +- UPSTASH_REDIS_REST_URL, UPSTASH_REDIS_REST_TOKEN (required) +- UPSTASH_HERMES_USER_ID: whose memories these are. Defaults to the gateway user, then the Hermes + profile name, then "default". +- UPSTASH_HERMES_MEMORY_PREFIX / UPSTASH_HERMES_CHAT_PREFIX: key prefixes. The defaults are the + TS SDK's (`agentkit:memory`, `agentkit:chat`), so memories are shared with AgentKit apps that + use the same database. +""" + +import json +import logging +import os +import threading +from typing import Any + +from .chat_history import DEFAULT_PREFIX as CHAT_PREFIX +from .chat_history import ChatHistory +from .memory import DEFAULT_PREFIX as MEMORY_PREFIX +from .memory import AgentMemory, MemoryRecord + +try: # Inside Hermes. Outside it (tests, plain library use) the provider is a plain class. + from agent.memory_provider import MemoryProvider +except ImportError: # pragma: no cover - depends on the host + MemoryProvider = object # type: ignore[assignment,misc] + +try: + from agent.secret_scope import get_secret +except ImportError: # pragma: no cover - depends on the host + + def get_secret(name: str, default: str | None = None) -> str | None: + return os.environ.get(name, default) + + +logger = logging.getLogger(__name__) + +# Hermes only lets the primary agent write; subagents, cron jobs and flushes read. +_WRITING_CONTEXTS = {"", "primary"} +_PREFETCH_TOP_K = 5 +_EXCERPT_CHARS = 300 + +SAVE_SCHEMA = { + "name": "upstash_memory_save", + "description": ( + "Save one durable fact about the user or their work to long-term memory (preferences, " + "decisions, people, projects). Write it as a short standalone sentence." + ), + "parameters": { + "type": "object", + "properties": {"text": {"type": "string", "description": "The fact to remember."}}, + "required": ["text"], + }, +} +SEARCH_SCHEMA = { + "name": "upstash_memory_search", + "description": "Search long-term memory for facts related to a query. Returns ids and text.", + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to look for."}, + "limit": {"type": "integer", "description": "Maximum results (default 5)."}, + }, + "required": ["query"], + }, +} +FORGET_SCHEMA = { + "name": "upstash_memory_forget", + "description": "Delete one memory by the id that upstash_memory_search returned.", + "parameters": { + "type": "object", + "properties": {"id": {"type": "string", "description": "The memory id."}}, + "required": ["id"], + }, +} +CHAT_SEARCH_SCHEMA = { + "name": "upstash_chat_search", + "description": ( + "Search the transcripts of past conversations with this user. Returns matching sessions " + "with an excerpt of each side." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to look for."}, + "limit": {"type": "integer", "description": "Maximum sessions (default 5)."}, + }, + "required": ["query"], + }, +} + + +def _key_part(value: str, what: str) -> str: + """Check a user or session id for the middle of a `::` key. + + Ids are used as they are, so they match the keys the TS SDK writes for the same id (`team%ops` + included). `:` separates the parts of the key and the TS SDK refuses it, so it is refused here + too: any encoding of it could equal another, literal TS id and share that id's memories. + """ + if ":" in value: + raise ValueError(f"{what} may not contain ':' (it is the key separator), got {value!r}") + return value + + +def _message_text(message: dict[str, Any]) -> str: + """A message's text the way Hermes flattens it for `sync_turn` (text parts, image count).""" + content = message.get("content") + if isinstance(content, str): + return content.strip() + if not isinstance(content, list): + return "" + texts, images = [], 0 + for part in content: + if isinstance(part, str): + texts.append(part) + elif isinstance(part, dict): + kind = str(part.get("type") or "").lower() + if kind in ("text", "input_text", "output_text") and isinstance(part.get("text"), str): + texts.append(part["text"]) + elif kind in ("image_url", "input_image"): + images += 1 + text = "\n".join(t for t in texts if t).strip() + if images: + note = f"[{images} image{'s' if images != 1 else ''}]" + text = f"{note} {text}" if text else note + return text + + +def _turns_of(messages: list[dict[str, Any]]) -> list[tuple[str, str]]: + """(user text, reply text) for each user message; the reply is the turn's last assistant text.""" + turns: list[tuple[str, str]] = [] + for message in messages: + role = message.get("role") + if role == "user": + turns.append((_message_text(message), "")) + elif role == "assistant" and turns: + text = _message_text(message) + if text: + turns[-1] = (turns[-1][0], text) + return turns + + +def rewind_transcript( + transcript: list[dict[str, str]], history: list[dict[str, Any]] +) -> list[dict[str, str]]: + """Cut a saved transcript back to what Hermes still has after `/undo`. + + `history` is Hermes's conversation for the session after the undo. Hermes does not say how + many turns were undone, so the cut is found by content: the transcript keeps everything up to + the latest saved turn matching the latest turn still in `history`, and drops the rest. When no + turn of the transcript is left, every saved turn was undone. + + In-place compaction keeps the session id but summarizes older messages, so `history` can be + much shorter than the transcript, and counting occurrences across the two would land on an + older turn. A turn is matched by its user text and its reply instead, and the latest match + wins, so a repeated "yes" lands on the right turn and retained history is never cut. + """ + saved: list[tuple[int, str, str]] = [] + for i, entry in enumerate(transcript): + if entry.get("role") != "user": + continue + reply = transcript[i + 1] if i + 1 < len(transcript) else {} + answer = str(reply.get("content", "")).strip() if reply.get("role") == "assistant" else "" + saved.append((i, str(entry.get("content", "")).strip(), answer)) + # Walk back past turns that were never saved, such as a compaction summary. + for user, reply in reversed([t for t in _turns_of(history) if t[0]]): + same_user = [s for s in saved if s[1] == user] + matches = [s for s in same_user if s[2] == reply] if reply else same_user + if not matches: + continue + i = matches[-1][0] + end = i + 1 + if end < len(transcript) and transcript[end].get("role") == "assistant": + end += 1 + return transcript[:end] + return [] + + +class UpstashMemoryProvider(MemoryProvider): + def __init__(self, redis: Any = None): + self._redis = redis + self._memory: AgentMemory | None = None + self._chats: ChatHistory | None = None + self._user_id = "default" + self._session_id = "" + self._write_enabled = True + # Transcript per session, loaded from Redis on first use so a resumed session appends. + self._transcripts: dict[str, list[dict[str, str]]] = {} + # Sessions Hermes rewound (/undo): their transcript is cut back on the next sync. + self._rewound: set[str] = set() + self._lock = threading.Lock() + + # -- identity and setup ------------------------------------------------------------------ + + @property + def name(self) -> str: + return "upstash" + + def is_available(self) -> bool: + if self._redis is not None: + return True + return bool( + get_secret("UPSTASH_REDIS_REST_URL", "") and get_secret("UPSTASH_REDIS_REST_TOKEN", "") + ) + + def unavailable_reason(self) -> str: + return "Set UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN (hermes memory setup)." + + def get_config_schema(self) -> list[dict[str, Any]]: + return [ + { + "key": "url", + "description": "Upstash Redis REST URL", + "required": True, + "env_var": "UPSTASH_REDIS_REST_URL", + "url": "https://console.upstash.com/redis", + }, + { + "key": "token", + "description": "Upstash Redis REST token", + "secret": True, + "required": True, + "env_var": "UPSTASH_REDIS_REST_TOKEN", + }, + { + "key": "user_id", + "description": "Whose memories these are (defaults to the Hermes profile name)", + "required": False, + "env_var": "UPSTASH_HERMES_USER_ID", + }, + ] + + def save_config(self, values: dict[str, Any], hermes_home: str) -> None: + # Every field maps to an env var, which `hermes memory setup` writes to $HERMES_HOME/.env. + pass + + def initialize(self, session_id: str, **kwargs: Any) -> None: + if self._redis is None: + from upstash_redis import Redis + + self._redis = Redis( + url=get_secret("UPSTASH_REDIS_REST_URL", ""), + token=get_secret("UPSTASH_REDIS_REST_TOKEN", ""), + ) + # Like the credentials, the prefixes are read per Hermes profile: a multiplexed gateway + # keeps each profile's .env out of os.environ. + self._memory = AgentMemory( + self._redis, prefix=get_secret("UPSTASH_HERMES_MEMORY_PREFIX", "") or MEMORY_PREFIX + ) + self._chats = ChatHistory( + self._redis, prefix=get_secret("UPSTASH_HERMES_CHAT_PREFIX", "") or CHAT_PREFIX + ) + user = ( + get_secret("UPSTASH_HERMES_USER_ID", "") + or kwargs.get("user_id") + or kwargs.get("agent_identity") + or "default" + ) + self._user_id = _key_part(str(user), "user id") + self._session_id = self._session_key(session_id) + self._write_enabled = (kwargs.get("agent_context") or "") in _WRITING_CONTEXTS + + def system_prompt_block(self) -> str: + return ( + "# Upstash Memory\n" + f"Long-term memory is stored in Upstash Redis for user `{self._user_id}`. Memories " + "related to each message are recalled automatically into . Use " + "upstash_memory_save for durable facts worth keeping across sessions, " + "upstash_memory_search to look one up, upstash_memory_forget to delete one, and " + "upstash_chat_search to find something said in an earlier conversation." + ) + + # -- lifecycle ----------------------------------------------------------------------------- + + def prefetch(self, query: str, *, session_id: str = "") -> str: + if not self._memory or not query.strip(): + return "" + try: + hits = self._memory.recall(query, user_id=self._user_id, top_k=_PREFETCH_TOP_K) + except Exception as error: # noqa: BLE001 - a Redis outage must never break a turn + logger.warning("upstash memory recall failed: %s", error) + return "" + if not hits: + return "" + lines = "\n".join(f"- {hit.text}" for hit in hits) + return f"## Upstash Memory\n{lines}" + + def sync_turn( + self, + user_content: str, + assistant_content: str, + *, + session_id: str = "", + messages: list[dict[str, Any]] | None = None, + ) -> None: + if not self._chats or not self._write_enabled: + return + sid = self._session_key(session_id) if session_id else self._session_id + if not sid: + return + with self._lock: + transcript = self._load_transcript(sid) + if sid in self._rewound and messages is not None: + # `messages` ends with this turn, which Hermes appended after the undo. + history = list(messages) + while history and history[-1].get("role") != "user": + history.pop() + transcript[:] = rewind_transcript(transcript, history[:-1]) + self._rewound.discard(sid) + if user_content: + transcript.append({"role": "user", "content": user_content}) + if assistant_content: + transcript.append({"role": "assistant", "content": assistant_content}) + self._save_transcript(sid, transcript) + + def on_session_switch( + self, new_session_id: str, *, rewound: bool = False, **kwargs: Any + ) -> None: + self._session_id = self._session_key(new_session_id) + if rewound: + # /undo keeps the session id but truncates Hermes's history. The next sync (or the + # session end) cuts the saved transcript back to match, before anything is appended. + with self._lock: + self._rewound.add(self._session_id) + + def on_session_end(self, messages: list[dict[str, Any]]) -> None: + """A rewind that no later turn followed still has to drop the undone turns from Redis.""" + sid = self._session_id + if not self._chats or not self._write_enabled or sid not in self._rewound: + return + with self._lock: + transcript = self._load_transcript(sid) + transcript[:] = rewind_transcript(transcript, list(messages or [])) + self._rewound.discard(sid) + self._save_transcript(sid, transcript) + + @staticmethod + def _session_key(session_id: str | None) -> str: + """The session id for transcript keys, or "" (no transcript) when it can't be one.""" + try: + return _key_part(session_id or "", "session id") + except ValueError as error: + logger.warning("upstash chat history skipped: %s", error) + return "" + + def _load_transcript(self, sid: str) -> list[dict[str, str]]: + transcript = self._transcripts.get(sid) + if transcript is None: + assert self._chats is not None + existing = self._chats.get_chat(sid, user_id=self._user_id) or {} + transcript = list(existing.get("messages") or []) + self._transcripts[sid] = transcript + return transcript + + def _save_transcript(self, sid: str, transcript: list[dict[str, str]]) -> None: + assert self._chats is not None + try: + # Create the index before the first write rather than leaving it to the first + # search, so no transcript depends on the index's initial backfill. + self._chats.index.ensure() + self._chats.save_chat(sid, transcript, user_id=self._user_id) + except Exception as error: # noqa: BLE001 - a Redis outage must never break a turn + logger.warning("upstash chat history save failed: %s", error) + + def on_memory_write( + self, action: str, target: str, content: str, metadata: dict[str, Any] | None = None + ) -> None: + """Mirror facts the built-in `memory` tool adds, so Redis holds them too.""" + if action != "add" or not content or not self._memory or not self._write_enabled: + return + try: + self._remember(content, {"source": "hermes", "target": target}) + except Exception as error: # noqa: BLE001 - as above + logger.warning("upstash memory mirror failed: %s", error) + + def _remember(self, text: str, metadata: dict[str, Any]) -> MemoryRecord: + """Save a fact so that it is searchable once this returns. + + Recall reads Redis Search, which indexes a write shortly after it lands. Waiting here + means a fact saved in one turn is recalled in the next one, or in a session started right + after. The index is created first, so the write is indexed as a write, not by a backfill. + """ + assert self._memory is not None + self._memory.index.ensure() + record = self._memory.add(text, user_id=self._user_id, metadata=metadata) + self._memory.index.wait_indexing() + return record + + def shutdown(self) -> None: + self._transcripts.clear() + self._rewound.clear() + + # -- tools --------------------------------------------------------------------------------- + + def get_tool_schemas(self) -> list[dict[str, Any]]: + return [SAVE_SCHEMA, SEARCH_SCHEMA, FORGET_SCHEMA, CHAT_SEARCH_SCHEMA] + + def handle_tool_call(self, tool_name: str, args: dict[str, Any], **kwargs: Any) -> str: + try: + return json.dumps(self._call(tool_name, args or {}), ensure_ascii=False) + except Exception as error: # noqa: BLE001 - Hermes expects errors as JSON results + return json.dumps({"error": str(error)}) + + def _call(self, tool_name: str, args: dict[str, Any]) -> dict[str, Any]: + if not self._memory or not self._chats: + raise RuntimeError("upstash memory is not initialized") + if tool_name == "upstash_memory_save": + if not self._write_enabled: + raise RuntimeError("memory is read-only in this agent context") + text = str(args.get("text") or "").strip() + if not text: + raise ValueError("text is required") + record = self._remember(text, {"source": "hermes"}) + return {"saved": True, "id": record.id} + if tool_name == "upstash_memory_search": + hits = self._memory.recall( + str(args.get("query") or ""), + user_id=self._user_id, + top_k=int(args.get("limit") or 5), + ) + return {"results": [{"id": h.id, "text": h.text, "score": h.score} for h in hits]} + if tool_name == "upstash_memory_forget": + if not self._write_enabled: + raise RuntimeError("memory is read-only in this agent context") + return { + "forgotten": self._memory.forget(str(args.get("id") or ""), user_id=self._user_id) + } + if tool_name == "upstash_chat_search": + hits = self._chats.search_chats( + str(args.get("query") or ""), + user_id=self._user_id, + limit=int(args.get("limit") or 5), + ) + return { + "sessions": [ + { + "session_id": h.session_id, + "score": h.score, + "title": h.title, + "updated_at": h.updated_at, + "user_excerpt": h.user_text[:_EXCERPT_CHARS], + "assistant_excerpt": h.model_text[:_EXCERPT_CHARS], + } + for h in hits + ] + } + raise ValueError(f"unknown tool {tool_name}") diff --git a/python/hermes-memory/tests/conftest.py b/python/hermes-memory/tests/conftest.py index a8edd97..78a25ca 100644 --- a/python/hermes-memory/tests/conftest.py +++ b/python/hermes-memory/tests/conftest.py @@ -1,4 +1,5 @@ import os +import uuid from pathlib import Path import pytest @@ -39,3 +40,21 @@ def redis(): from upstash_redis import Redis return Redis.from_env() + + +@pytest.fixture +def prefix(redis): + """A fresh key prefix per test (like the TS `uniquePrefix`); its keys are deleted afterwards. + + Indexes built on it must be dropped by the test's own fixtures: the test database allows only + 10 search indexes, shared with the vitest suites. + """ + value = f"test:py:{uuid.uuid4().hex[:8]}" + yield value + cursor = 0 + while True: + cursor, keys = redis.scan(cursor, match=f"{value}*", count=200) + if keys: + redis.delete(*keys) + if int(cursor) == 0: + break diff --git a/python/hermes-memory/tests/test_layout.py b/python/hermes-memory/tests/test_layout.py new file mode 100644 index 0000000..63f8a91 --- /dev/null +++ b/python/hermes-memory/tests/test_layout.py @@ -0,0 +1,103 @@ +"""The Redis layout must stay byte-compatible with the TS SDK (packages/sdk/src).""" + +from upstash_hermes_memory import AgentMemory, ChatHistory +from upstash_hermes_memory._search import dumps, index_name_for +from upstash_hermes_memory._telemetry import SDK_TELEMETRY, add_telemetry +from upstash_hermes_memory.chat_history import extract_text + + +class Recorder: + def __init__(self): + self.commands = [] + self._headers = {"Upstash-Telemetry-Sdk": "py-upstash-redis@v1.8.0"} + + def execute(self, command): + self.commands.append(command) + return True + + def expire(self, key, seconds): + self.commands.append(["EXPIRE", key, seconds]) + return True + + +def test_index_names_follow_the_ts_default(): + assert index_name_for("agentkit:memory") == "agentkit_memory" + assert index_name_for("agentkit:chat") == "agentkit_chat" + assert index_name_for("my-app.mem:v2") == "my_app_mem_v2" + + +def test_memory_index_matches_reactive_search_index(): + memory = AgentMemory(Recorder(), enable_telemetry=False) + assert " ".join(memory.index.create_command()) == ( + "SEARCH.CREATE agentkit_memory EXISTSOK ON JSON PREFIX 1 agentkit:memory: " + "SCHEMA text TEXT userId TEXT NOTOKENIZE" + ) + + +def test_chat_index_matches_reactive_search_index(): + chats = ChatHistory(Recorder(), enable_telemetry=False) + assert " ".join(chats.index.create_command()) == ( + "SEARCH.CREATE agentkit_chat EXISTSOK ON JSON PREFIX 1 agentkit:chat: SCHEMA " + "userId TEXT NOTOKENIZE sessionId TEXT NOTOKENIZE userMessages TEXT modelMessages TEXT" + ) + + +def test_memory_document_matches_agent_memory(): + redis = Recorder() + record = AgentMemory(redis, enable_telemetry=False).add("likes tea", user_id="u1", id="m1") + (command,) = redis.commands + assert command[:3] == ["JSON.SET", "agentkit:memory:u1:m1", "$"] + assert command[3] == f'{{"text":"likes tea","userId":"u1","createdAt":{record.created_at}}}' + + +def test_json_is_written_like_json_stringify(): + assert dumps({"a": "çay", "b": [1, None]}) == '{"a":"çay","b":[1,null]}' + + +def test_extract_text_matches_the_ts_default_extractor(): + text = extract_text( + [ + {"role": "user", "content": " hi "}, + {"role": "assistant", "parts": [{"type": "text", "text": "hello"}, {"type": "tool"}]}, + {"role": "tool", "content": "ignored"}, + {"role": "model", "content": "again"}, + {"role": "user", "content": " "}, + ] + ) + assert text == {"userMessages": "hi", "modelMessages": "hello\nagain"} + + +def test_ids_with_colons_are_refused(): + memory = AgentMemory(Recorder(), enable_telemetry=False) + for bad in ("", "a:b"): + try: + memory.add("x", user_id=bad) + except ValueError: + continue + raise AssertionError(f"accepted user_id {bad!r}") + + +def test_telemetry_tags_the_client_once(monkeypatch): + monkeypatch.delenv("UPSTASH_DISABLE_TELEMETRY", raising=False) + redis = Recorder() + AgentMemory(redis) + ChatHistory(redis) + assert redis._headers["Upstash-Telemetry-Sdk"] == f"py-upstash-redis@v1.8.0,{SDK_TELEMETRY}" + + +def test_telemetry_opt_outs(monkeypatch): + redis = Recorder() + add_telemetry(redis, enabled=False) + monkeypatch.setenv("UPSTASH_DISABLE_TELEMETRY", "1") + add_telemetry(redis) + assert redis._headers["Upstash-Telemetry-Sdk"] == "py-upstash-redis@v1.8.0" + + +def test_chat_ttl_is_applied_whenever_set_like_the_ts_sdk(): + # TS ChatHistory sends EXPIRE when ttlSeconds is defined, so 0 expires the chat at once. + for ttl, expected in ((None, []), (0, [0]), (3600, [3600])): + redis = Recorder() + ChatHistory(redis, ttl_seconds=ttl, enable_telemetry=False).save_chat( + "s1", [{"role": "user", "content": "hi"}], user_id="u1" + ) + assert [c[2] for c in redis.commands if c[0] == "EXPIRE"] == expected diff --git a/python/hermes-memory/tests/test_package.py b/python/hermes-memory/tests/test_package.py index 518b13b..de64577 100644 --- a/python/hermes-memory/tests/test_package.py +++ b/python/hermes-memory/tests/test_package.py @@ -12,10 +12,28 @@ def test_public_api(): - assert upstash_hermes_memory.__all__ == ["VERSION"] + assert set(upstash_hermes_memory.__all__) == { + "VERSION", + "AgentMemory", + "ChatHistory", + "MemoryRecord", + "UpstashMemoryProvider", + "register", + } assert isinstance(VERSION, str) +def test_register_hands_hermes_one_provider(): + registered = [] + + class Ctx: + def register_memory_provider(self, provider): + registered.append(provider) + + upstash_hermes_memory.register(Ctx()) + assert [p.name for p in registered] == ["upstash"] + + @pytest.mark.skipif(shutil.which("uv") is None, reason="uv not on PATH") def test_wheel_contents(tmp_path): # What publish-python.mjs uploads: build the wheel the same way and check what is inside it. @@ -28,17 +46,20 @@ def test_wheel_contents(tmp_path): (wheel,) = tmp_path.glob("*.whl") assert wheel.name.startswith(f"upstash_hermes_memory-{VERSION}-") + dist_info = f"upstash_hermes_memory-{VERSION}.dist-info" with zipfile.ZipFile(wheel) as zf: names = zf.namelist() - assert "upstash_hermes_memory/__init__.py" in names - assert "upstash_hermes_memory/_version.py" in names + for module in ("__init__", "_version", "memory", "chat_history", "provider"): + assert f"upstash_hermes_memory/{module}.py" in names # Tests and the private package.json are repo plumbing, not part of the distribution. assert not any(n.startswith("tests/") or n.endswith("package.json") for n in names) - metadata = Parser().parsestr( - zf.read(f"upstash_hermes_memory-{VERSION}.dist-info/METADATA").decode() - ) + metadata = Parser().parsestr(zf.read(f"{dist_info}/METADATA").decode()) + entry_points = zf.read(f"{dist_info}/entry_points.txt").decode() assert metadata["Name"] == "upstash-hermes-memory" assert metadata["Version"] == VERSION assert metadata["Requires-Python"] == ">=3.11" assert any(r.startswith("upstash-redis") for r in metadata.get_all("Requires-Dist")) + # How Hermes finds the provider once the package is pip-installed next to it. + assert "[hermes_agent.memory_providers]" in entry_points + assert "upstash = upstash_hermes_memory:register" in entry_points diff --git a/python/hermes-memory/tests/test_provider.py b/python/hermes-memory/tests/test_provider.py new file mode 100644 index 0000000..37fdf3a --- /dev/null +++ b/python/hermes-memory/tests/test_provider.py @@ -0,0 +1,267 @@ +"""The Hermes provider: its contract with Hermes, then its lifecycle against the test Redis.""" + +import json + +import pytest +from upstash_hermes_memory import UpstashMemoryProvider +from upstash_hermes_memory import provider as provider_module +from upstash_hermes_memory.provider import _key_part, rewind_transcript + +# Tools Hermes itself owns; a provider tool with one of these names is dropped (memory_manager.py). +HERMES_CORE_TOOL_NAMES = {"memory", "session_search", "todo", "clarify"} + + +def test_tool_schemas_are_prefixed_and_well_formed(): + schemas = UpstashMemoryProvider().get_tool_schemas() + names = [s["name"] for s in schemas] + assert all(n.startswith("upstash_") for n in names) + assert not HERMES_CORE_TOOL_NAMES & set(names) + for schema in schemas: + assert schema["parameters"]["type"] == "object" + assert set(schema["parameters"]["required"]) <= set(schema["parameters"]["properties"]) + + +def test_is_available_needs_both_credentials(monkeypatch): + monkeypatch.delenv("UPSTASH_REDIS_REST_URL", raising=False) + monkeypatch.setenv("UPSTASH_REDIS_REST_TOKEN", "t") + assert not UpstashMemoryProvider().is_available() + monkeypatch.setenv("UPSTASH_REDIS_REST_URL", "https://example.upstash.io") + assert UpstashMemoryProvider().is_available() + + +def test_tool_errors_come_back_as_json(): + result = json.loads( + UpstashMemoryProvider().handle_tool_call("upstash_memory_save", {"text": "x"}) + ) + assert "error" in result + + +@pytest.fixture +def provider(redis, prefix, monkeypatch): + monkeypatch.setenv("UPSTASH_HERMES_MEMORY_PREFIX", f"{prefix}:memory") + monkeypatch.setenv("UPSTASH_HERMES_CHAT_PREFIX", f"{prefix}:chat") + monkeypatch.delenv("UPSTASH_HERMES_USER_ID", raising=False) + provider = UpstashMemoryProvider(redis) + provider.initialize("20261022_120000_abcd", platform="cli", agent_identity="arda") + provider._memory.index.ensure() + provider._chats.index.ensure() + yield provider + provider._memory.index.drop() + provider._chats.index.drop() + + +def test_user_and_session_ids_are_used_as_they_are(provider): + assert provider._user_id == "arda" + assert provider._session_id == "20261022_120000_abcd" + + +def test_save_then_prefetch(provider): + saved = json.loads( + provider.handle_tool_call( + "upstash_memory_save", {"text": "The user's dog is named Karabas"} + ) + ) + assert saved["saved"] is True + + # No waiting here: the save tool only returns once the fact is searchable. + block = provider.prefetch("what is my dog called?") + assert block.startswith("## Upstash Memory") + assert "Karabas" in block + + found = json.loads(provider.handle_tool_call("upstash_memory_search", {"query": "dog"})) + assert [r["id"] for r in found["results"]] == [saved["id"]] + + forgotten = json.loads(provider.handle_tool_call("upstash_memory_forget", {"id": saved["id"]})) + assert forgotten == {"forgotten": True} + + +def test_turns_become_a_searchable_transcript(provider): + provider.sync_turn("I'm planning a trip to Lisbon", "Great, when?", session_id="s-1") + provider.sync_turn("In May", "May is lovely there.", session_id="s-1") + + doc = provider._chats.get_chat("s-1", user_id="arda") + assert [m["role"] for m in doc["messages"]] == ["user", "assistant", "user", "assistant"] + + provider._chats.index.wait_indexing() + found = json.loads(provider.handle_tool_call("upstash_chat_search", {"query": "Lisbon"}))[ + "sessions" + ] + assert [s["session_id"] for s in found] == ["s-1"] + + +def test_a_resumed_session_appends_instead_of_overwriting(provider, redis): + provider.sync_turn("first", "one", session_id="s-2") + resumed = UpstashMemoryProvider(redis) + resumed.initialize("s-2", platform="cli", agent_identity="arda") + resumed.sync_turn("second", "two", session_id="s-2") + doc = resumed._chats.get_chat("s-2", user_id="arda") + assert [m["content"] for m in doc["messages"]] == ["first", "one", "second", "two"] + + +def test_built_in_memory_writes_are_mirrored(provider): + provider.on_memory_write("add", "user", "Prefers answers in Turkish") + provider.on_memory_write("remove", "user", "Prefers answers in Turkish") # not mirrored + hits = provider._memory.recall("Turkish", user_id="arda") # mirroring waits for indexing + assert [(h.text, h.metadata) for h in hits] == [ + ("Prefers answers in Turkish", {"source": "hermes", "target": "user"}) + ] + + +def test_subagents_read_but_never_write(provider, redis): + sub = UpstashMemoryProvider(redis) + sub.initialize("s-sub", platform="cli", agent_identity="arda", agent_context="subagent") + sub.sync_turn("hello", "hi", session_id="s-sub") + sub.on_memory_write("add", "memory", "should not land") + refused = json.loads(sub.handle_tool_call("upstash_memory_save", {"text": "nope"})) + assert "error" in refused + assert sub._chats.get_chat("s-sub", user_id="arda") is None + assert sub._memory.count(user_id="arda") == 0 + + +def test_key_parts_match_the_ts_sdk(): + # The TS SDK stores these literally, so the provider must too, or the two read different keys. + assert _key_part("team%ops", "user id") == "team%ops" + assert _key_part("alice%3Awork", "user id") == "alice%3Awork" + # The TS SDK refuses `:`, and any encoding of it could equal a literal TS id like the above. + with pytest.raises(ValueError): + _key_part("alice:work", "user id") + + +def test_a_user_id_with_a_colon_is_refused(monkeypatch): + monkeypatch.setenv("UPSTASH_HERMES_USER_ID", "alice:work") + with pytest.raises(ValueError, match="user id"): + UpstashMemoryProvider(object()).initialize("s", platform="cli") + + +def test_a_session_id_with_a_colon_keeps_no_transcript(provider): + provider.sync_turn("hi", "hello", session_id="ab:cd") + assert provider._transcripts == {} + + +def test_ids_with_percent_are_shared_with_the_ts_sdk(provider, redis): + # A TS `AgentMemory.add(text, { userId: "team%ops" })` writes this key and document. + key = f"{provider._memory.prefix}:team%ops:from-ts" + doc = {"text": "the deploy freeze starts friday", "userId": "team%ops", "createdAt": 1} + redis.execute(["JSON.SET", key, "$", json.dumps(doc)]) + provider._user_id = _key_part("team%ops", "user id") + assert provider._memory.get("from-ts", user_id=provider._user_id) is not None + assert json.loads(provider.handle_tool_call("upstash_memory_forget", {"id": "from-ts"})) == { + "forgotten": True + } + + +def test_prefixes_come_from_the_hermes_profile(monkeypatch): + # In a multiplexed gateway a profile's .env is only reachable through get_secret. + profile = { + "UPSTASH_HERMES_MEMORY_PREFIX": "team:memory", + "UPSTASH_HERMES_CHAT_PREFIX": "team:chat", + } + monkeypatch.delenv("UPSTASH_HERMES_MEMORY_PREFIX", raising=False) + monkeypatch.delenv("UPSTASH_HERMES_CHAT_PREFIX", raising=False) + monkeypatch.setattr( + provider_module, "get_secret", lambda name, default=None: profile.get(name, default) + ) + provider = UpstashMemoryProvider(object()) + provider.initialize("s", platform="cli") + assert (provider._memory.prefix, provider._chats.prefix) == ("team:memory", "team:chat") + + +def _turns(*texts): + out = [] + for user, assistant in texts: + out += [{"role": "user", "content": user}, {"role": "assistant", "content": assistant}] + return out + + +def test_rewind_drops_the_undone_turns(): + saved = _turns(("a", "1"), ("b", "2"), ("c", "3")) + after_undo = _turns(("a", "1"), ("b", "2")) + assert rewind_transcript(saved, after_undo) == _turns(("a", "1"), ("b", "2")) + assert rewind_transcript(saved, _turns(("a", "1"))) == _turns(("a", "1")) + assert rewind_transcript(saved, []) == [] + + +def test_rewind_matches_repeated_messages_by_their_reply(): + saved = _turns(("yes", "1"), ("yes", "2"), ("no", "3")) + assert rewind_transcript(saved, _turns(("yes", "1"))) == _turns(("yes", "1")) + assert rewind_transcript(saved, _turns(("yes", "1"), ("yes", "2"))) == _turns( + ("yes", "1"), ("yes", "2") + ) + + +# In-place compaction keeps the session id: the opening stays, the middle becomes a summary. +_SAVED = _turns( + ("opening goal", "g"), ("yes", "1"), ("middle details", "m"), ("yes", "2"), ("undo this", "u") +) +_COMPACTED = _turns(("opening goal", "g")) + [ + {"role": "user", "content": "[CONTEXT SUMMARY] the user agreed twice and gave details"}, + {"role": "assistant", "content": "noted"}, + *_turns(("yes", "2")), +] + + +def test_rewind_after_in_place_compaction_keeps_the_retained_turns(): + assert rewind_transcript(_SAVED, _COMPACTED) == _SAVED[:8] + + +def test_undo_after_in_place_compaction_then_a_new_turn(provider): + for user, reply in zip(_SAVED[::2], _SAVED[1::2]): + provider.sync_turn(user["content"], reply["content"], session_id="s-compact") + provider.on_session_switch("s-compact", rewound=True) + history = _COMPACTED + _turns(("next step", "n")) + provider.sync_turn("next step", "n", session_id="s-compact", messages=history) + doc = provider._chats.get_chat("s-compact", user_id="arda") + assert doc["messages"] == _SAVED[:8] + _turns(("next step", "n")) + + +def test_undo_after_in_place_compaction_then_exit(provider): + for user, reply in zip(_SAVED[::2], _SAVED[1::2]): + provider.sync_turn(user["content"], reply["content"], session_id="s-compact-exit") + provider.on_session_switch("s-compact-exit", rewound=True) + provider.on_session_end(_COMPACTED) + doc = provider._chats.get_chat("s-compact-exit", user_id="arda") + assert doc["messages"] == _SAVED[:8] + + +def test_rewind_matches_multimodal_messages_as_hermes_flattens_them(): + saved = _turns(("[1 image] what is this?", "a cat"), ("thanks", "np")) + history = [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": "data:"}}, + {"type": "text", "text": "what is this?"}, + ], + }, + {"role": "assistant", "content": "a cat"}, + {"role": "tool", "content": "ignored"}, + ] + assert rewind_transcript(saved, history) == _turns(("[1 image] what is this?", "a cat")) + + +def test_undo_removes_the_undone_turn_from_redis(provider): + provider.sync_turn("plan Kyoto", "ok", session_id="s-undo") + provider.sync_turn("book the secret hotel", "done", session_id="s-undo") + # /undo of the second turn, then a new turn: Hermes passes its post-undo history plus the turn. + provider.on_session_switch("s-undo", rewound=True) + history = _turns(("plan Kyoto", "ok")) + _turns(("plan Osaka instead", "sure")) + provider.sync_turn("plan Osaka instead", "sure", session_id="s-undo", messages=history) + + doc = provider._chats.get_chat("s-undo", user_id="arda") + assert [m["content"] for m in doc["messages"]] == [ + "plan Kyoto", + "ok", + "plan Osaka instead", + "sure", + ] + provider._chats.index.wait_indexing() + assert provider._chats.search_chats("secret hotel", user_id="arda") == [] + + +def test_undo_then_exit_still_cleans_redis(provider): + provider.sync_turn("first", "one", session_id="s-exit") + provider.sync_turn("second", "two", session_id="s-exit") + provider.on_session_switch("s-exit", rewound=True) + provider.on_session_end(_turns(("first", "one"))) + doc = provider._chats.get_chat("s-exit", user_id="arda") + assert [m["content"] for m in doc["messages"]] == ["first", "one"] diff --git a/python/hermes-memory/tests/test_redis.py b/python/hermes-memory/tests/test_redis.py index 880b9ee..f8a02ae 100644 --- a/python/hermes-memory/tests/test_redis.py +++ b/python/hermes-memory/tests/test_redis.py @@ -1,5 +1,10 @@ +"""AgentMemory and ChatHistory against the shared test Redis (skipped without secrets).""" + import uuid +import pytest +from upstash_hermes_memory import AgentMemory, ChatHistory + def test_round_trip(redis): # Smoke test for the CI wiring: the pinned client reaches the shared test database. It uses a @@ -11,3 +16,125 @@ def test_round_trip(redis): finally: redis.delete(key) assert redis.get(key) is None + + +@pytest.fixture +def memory(redis, prefix): + memory = AgentMemory(redis, prefix=f"{prefix}:memory", enable_telemetry=False) + # Index first, then seed: writes are indexed as writes, never left to the create-time backfill. + memory.index.ensure() + yield memory + memory.index.drop() + + +@pytest.fixture +def chats(redis, prefix): + chats = ChatHistory(redis, prefix=f"{prefix}:chat", enable_telemetry=False) + chats.index.ensure() + yield chats + chats.index.drop() + + +def test_memory_lifecycle(memory): + memory.add("Arda prefers dark roast coffee", user_id="u1", id="coffee") + memory.add("Arda's cat is called Pamuk", user_id="u1") + memory.add("Someone else likes coffee too", user_id="u2") + memory.index.wait_indexing() + + hits = memory.recall("coffee", user_id="u1") + assert [h.id for h in hits] == ["coffee"] + assert hits[0].text == "Arda prefers dark roast coffee" + assert hits[0].score > 0 + + assert memory.count(user_id="u1") == 2 + assert len(memory.list(user_id="u1")) == 2 + assert memory.get("coffee", user_id="u1").text == "Arda prefers dark roast coffee" + assert memory.get("coffee", user_id="u2") is None + + assert memory.forget("coffee", user_id="u1") + assert memory.get("coffee", user_id="u1") is None + memory.index.wait_indexing() + assert memory.count(user_id="u1") == 1 + assert memory.recall("coffee", user_id="u1") == [] + + +def test_memory_reads_documents_written_by_the_ts_sdk(redis, memory): + # Exactly what AgentMemory.add writes from TypeScript, metadata spread at the top level. + redis.execute( + [ + "JSON.SET", + f"{memory.prefix}:u1:from-ts", + "$", + '{"text":"deploys happen on Fridays","userId":"u1","createdAt":1700000000000,"team":"core"}', + ] + ) + memory.index.wait_indexing() + hits = memory.recall("deploys", user_id="u1") + assert [(h.id, h.created_at, h.metadata) for h in hits] == [ + ("from-ts", 1700000000000, {"team": "core"}) + ] + + +def test_chat_history_lifecycle(chats): + first = chats.save_chat( + "s1", + [ + {"role": "user", "content": "plan a trip to Kyoto"}, + {"role": "assistant", "content": "sure"}, + ], + user_id="u1", + title="Kyoto", + ) + second = chats.save_chat( + "s1", + [ + {"role": "user", "content": "plan a trip to Kyoto"}, + {"role": "assistant", "content": "done"}, + ], + user_id="u1", + ) + assert second["createdAt"] == first["createdAt"] + assert second["title"] == "Kyoto" # kept from the first save + chats.save_chat("s2", [{"role": "user", "content": "fix my bike"}], user_id="u1") + chats.index.wait_indexing() + + doc = chats.get_chat("s1", user_id="u1") + assert doc["userMessages"] == "plan a trip to Kyoto" + assert doc["modelMessages"] == "done" + + hits = chats.search_chats("Kyoto", user_id="u1") + assert [h.session_id for h in hits] == ["s1"] + assert chats.search_chats("Kyoto", user_id="u2") == [] + + listed = chats.list_chats(user_id="u1") + assert [d["sessionId"] for d in listed] == ["s2", "s1"] # most recently updated first + + assert chats.delete_chat("s1", user_id="u1") + assert chats.get_chat("s1", user_id="u1") is None + + +def test_metadata_cannot_replace_the_document_fields(memory): + for field in ("userId", "text", "createdAt"): + with pytest.raises(ValueError, match=field): + memory.add("x", user_id="alice", metadata={field: "bob"}) + memory.index.wait_indexing() + assert memory.count(user_id="alice") == 0 + assert memory.count(user_id="bob") == 0 + + +def test_memory_ids_may_contain_colons_like_in_the_ts_sdk(memory): + memory.add("ship the beta", user_id="u1", id="project:42") + memory.index.wait_indexing() + assert [h.id for h in memory.recall("beta", user_id="u1")] == ["project:42"] + assert memory.get("project:42", user_id="u1").text == "ship the beta" + assert memory.forget("project:42", user_id="u1") + + +def test_search_recovers_when_another_client_drops_the_index(redis, memory): + memory.add("the office moved to Kadikoy", user_id="u1") + memory.index.wait_indexing() + assert memory.recall("office", user_id="u1") + # Someone else drops the shared index; this instance still believes it is there. + redis.execute(["SEARCH.DROP", memory.index.name]) + hits = memory.recall("office", user_id="u1") + assert [h.text for h in hits] == ["the office moved to Kadikoy"]