diff --git a/AGENTS.md b/AGENTS.md index ba8ae6ac..0192c7fb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -56,8 +56,8 @@ and concise documentation together. `@agentscope-ai/reme_studio` npm static distribution. - `typescript/`: the independently published `@agentscope-ai/reme` package, including the shared TypeScript client and DeepSeek Harness and OpenClaw adapters. -- `plugins/`: installable ReMe extensions, including Auto Fin and LME/BEAM benchmark Steps and application presets. -- `integrations/`: adapters that connect ReMe to external agent hosts, such as Claude Code, DSH, and Hermes Agent. +- `plugins/`: installable ReMe extensions, including Auto Fin and LME/BEAM plugins. +- `integrations/`: adapters that connect ReMe to external agent hosts, including Claude Code and Hermes Agent. - `skills/`: standalone skills; `reme_memory` calls ReMe, while other skills may use separate tools or direct-file conventions. - `benchmark/` and `cookbook/`: runnable evaluations and example workflows. diff --git a/README.md b/README.md index 65c6e773..116c2c16 100644 --- a/README.md +++ b/README.md @@ -185,7 +185,7 @@ lifecycle according to the capabilities of each runtime. | **OpenClaw** | Install [`@agentscope-ai/reme`](typescript/README.md#openclaw) with `openclaw plugins install @agentscope-ai/reme`. | Native memory tools, recall before user-triggered runs, and automatic turn capture. | | **QwenPaw** | Embed ReMe in-process through its Python API. | Reuse the host lifecycle and model config while keeping memory local and file-based. | | **Claude Code** | Start the streamable HTTP MCP service and install [the ReMe plugin](integrations/claude_code/reme). | MCP recall tools, the `reme-memory` skill, and a Stop hook that records sessions automatically. | -| **Hermes** | Start the HTTP service and install [the ReMe provider](integrations/hermes_agent). | Recall before model calls and asynchronous `auto_memory` after each completed turn. | +| **Hermes** | Install [the ReMe provider](integrations/hermes_agent) and choose HTTP or embedded mode. | Recall before model calls and asynchronous `auto_memory` after each completed turn. | | **Codex and other CLI agents** | Install or copy the [ReMe Memory skill](skills/reme_memory/SKILL.md). | Search, read, and write memory through the CLI; automatic capture requires host lifecycle integration. |

Integration demos

diff --git a/README_ZH.md b/README_ZH.md index 11216b1e..c4b65120 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -183,7 +183,7 @@ runtime 的能力，将记忆指引、召回和捕获接入 Agent 生命周期 | **OpenClaw** | 使用 `openclaw plugins install @agentscope-ai/reme` 安装 [`@agentscope-ai/reme`](typescript/README_ZH.md#openclaw)。 | 原生记忆工具、用户触发运行前召回和自动对话捕获。 | | **QwenPaw** | 通过 Python API 在进程内嵌入 ReMe。 | 复用宿主生命周期和模型配置，同时保持记忆本地、文件化。 | | **Claude Code** | 启动 streamable HTTP MCP service，并安装 [ReMe 插件](integrations/claude_code/reme)。 | MCP 召回工具、`reme-memory` skill，以及自动记录会话的 Stop hook。 | -| **Hermes** | 启动 HTTP service，并安装 [ReMe provider](integrations/hermes_agent)。 | 模型调用前召回，每轮对话完成后异步执行 `auto_memory`。 | +| **Hermes** | 安装 [ReMe provider](integrations/hermes_agent)，并选择 HTTP 或 Embedded 模式。 | 模型调用前召回，每轮对话完成后异步执行 `auto_memory`。 | | **Codex 及其他 CLI Agent** | 安装或复制 [ReMe Memory skill](skills/reme_memory/SKILL.md)。 | 通过 CLI 搜索、读取和写入记忆；自动捕获需要显式接入宿主生命周期。 |

集成演示

diff --git a/docs/en/integrations.md b/docs/en/integrations.md index 6e2e828a..fd1653a6 100644 --- a/docs/en/integrations.md +++ b/docs/en/integrations.md @@ -52,7 +52,7 @@ The [`@agentscope-ai/reme` TypeScript package](./integrations/typescript.md) pro ## Hermes Agent -`integrations/hermes_agent/` provides a memory provider that recalls context before model calls and asynchronously invokes `auto_memory` after each turn. +`integrations/hermes_agent/` provides a memory provider with HTTP and embedded modes. It recalls context before model calls and asynchronously invokes `auto_memory` after each turn. Its `config_schema.py` is rendered by Hermes' generic memory settings UI. ## Production guidance diff --git a/docs/en/reme-blog.md b/docs/en/reme-blog.md index 34732191..a0053233 100644 --- a/docs/en/reme-blog.md +++ b/docs/en/reme-blog.md @@ -320,7 +320,7 @@ that best fits their runtime environment and share the same local memory workspa | **OpenClaw** | Install [`@agentscope-ai/reme`](../../typescript/README.md#openclaw) as the native memory plugin. | Recall before conversational root-agent runs, explicit search, automatic turn capture, and scheduled Auto Dream. | | **QwenPaw** | Embed ReMe in-process through the Python API. | Reuse the host application's lifecycle and model configuration while keeping memories local and file-based. | | **Claude Code** | Start the streamable HTTP MCP Service and install [`integrations/claude_code/reme`](../../integrations/claude_code/reme). | MCP memory-recall tools, the `reme-memory` skill, and a Stop hook that automatically records sessions. | -| **Hermes** | Start the HTTP Service and install [`integrations/hermes_agent`](../../integrations/hermes_agent). | Automatically recall relevant memories before model calls and invoke `auto_memory` asynchronously after each conversation turn. | +| **Hermes** | Install [`integrations/hermes_agent`](../../integrations/hermes_agent) and choose HTTP or embedded mode. | Automatically recall relevant memories before model calls and invoke `auto_memory` asynchronously after each conversation turn. | | **Codex and other CLI-capable agents** | Copy or install [`skills/reme_memory/SKILL.md`](../../skills/reme_memory/SKILL.md). | Search, read, and write memories through the CLI; automatic recording requires the host agent to integrate explicitly with the conversation lifecycle. | For installation, configuration, and integration demos, see the [README](../../README.md). diff --git a/docs/zh/integrations.md b/docs/zh/integrations.md index 671ce4af..dba50739 100644 --- a/docs/zh/integrations.md +++ b/docs/zh/integrations.md @@ -76,7 +76,7 @@ Skill 不应： ## Hermes Agent -`integrations/hermes_agent/` 提供 memory provider：模型调用前检索相关记忆，每轮结束后异步调用 `auto_memory`。完整配置见该目录 README。 +`integrations/hermes_agent/` 提供 HTTP 和 Embedded 双模式 memory provider：模型调用前检索相关记忆，每轮结束后异步调用 `auto_memory`，并通过 Hermes 通用配置面板展示设置。完整配置见该目录 README。 ## 生产接入建议 diff --git a/docs/zh/integrations/hermes.md b/docs/zh/integrations/hermes.md index 2371b1f8..85416fec 100644 --- a/docs/zh/integrations/hermes.md +++ b/docs/zh/integrations/hermes.md @@ -1,15 +1,33 @@ --- title: Hermes Agent 集成 -description: 使用 ReMe memory provider 在 Hermes 调用模型前召回、每轮结束后异步记录。 +description: 使用 HTTP 或 Embedded ReMe memory provider，在模型调用前召回、每轮结束后异步记录。 --- # Hermes Agent 集成 -Hermes memory provider 连接到一个已经运行的 ReMe HTTP 服务，在每次模型调用前召回相关记忆，并在用户/助手回合完成后异步调用 `auto_memory`。 +ReMe 的 Hermes memory provider 支持两种运行方式： -## Workspace 隔离 +- HTTP（默认）：连接独立运行的 ReMe 服务，Hermes 环境不需要安装 ReMe SDK； +- Embedded：在 Hermes Python 进程内创建 ReMe `Application`，不需要服务进程和端口。 -ReMe 的搜索范围是一个完整 workspace。多个 Hermes profile 指向同一个 workspace 时会共享召回结果；需要隔离时，为每个 profile 使用独立 workspace 和端点。 +两种模式都会在模型调用前执行 `search`，并在用户/助手回合完成后把 `auto_memory` 放入串行后台队列。 + +## 安装与配置 + +```bash +hermes plugins install agentscope-ai/ReMe/integrations/hermes_agent +hermes memory setup +``` + +Hermes Dashboard 会展示 provider 的模式相关字段和高级 recall、health、write、shutdown 设置；切换模式后只显示对应的 +HTTP endpoint 或 Embedded workspace 字段。 + +配置主路径为 `$HERMES_HOME/reme/config.json`。新文件未包含的字段仍会从旧 `$HERMES_HOME/reme.json` 继承，新文件中的 +值优先；后续 CLI 保存会写入完整的新配置，但不会删除旧文件。 + +## HTTP 模式 + +为当前 Hermes profile 启动独立 workspace： ```bash reme start \ @@ -19,41 +37,55 @@ reme start \ service.port=2333 ``` -自动记忆需要 LLM；默认 BM25 搜索不需要 Embedding。 - -## 安装与配置 - -```bash -hermes plugins install agentscope-ai/ReMe/integrations/hermes_agent -hermes memory setup -``` - -选择 `reme`，接受默认的 `http://127.0.0.1:2333`，或输入上一步使用的端点。Setup 会先调用 `health_check`，只有新端点健康时才替换现有 provider 配置。 - -配置存放在 `$HERMES_HOME/reme.json`： +配置示例： ```json { - "endpoint": "http://127.0.0.1:2333", - "request_timeout": 600.0, - "recall_timeout": 5.0, - "health_timeout": 2.0, - "health_retry_seconds": 30.0, - "shutdown_timeout": 30.0, - "recall_limit": 5 + "mode": "http", + "endpoint": "http://127.0.0.1:2333" } ``` -运行 `hermes memory status` 检查安装和配置。新的 Hermes 会话还会重新检查端点健康状态。 +终端 setup 会在保存前调用 `health_check`。HTTP action 接口没有本集成专用的认证头，不要直接暴露到公网；跨主机使用时 +应放在可信网络、SSH tunnel 或带认证的反向代理后。 + +## Embedded 模式 + +先把 ReMe 安装到 Hermes 使用的同一个 Python 环境： + +```bash +pip install "reme-ai[core]" +``` + +然后配置独立 workspace： + +```json +{ + "mode": "embedded", + "workspace_dir": "~/.reme-hermes-default", + "reme_config": "default" +} +``` + +插件在专用 asyncio loop thread 上构造并启动 `reme.Application`，直接执行 `health_check`、`search` 和 `auto_memory`。 +关闭时会在有界时间内排空写队列、调用 `Application.close()`、停止 loop 并 join thread；不会调用 +`Application.run_app()`，因此不会监听端口。 + +## Workspace 隔离 + +ReMe 搜索覆盖整个 workspace。多个 Hermes profile 指向同一个 workspace 时会共享召回结果；除非明确需要共享，否则为 +每个 profile 配置不同 workspace。HTTP 模式通常也使用不同端口。 + +自动记忆需要可用的 LLM 配置；默认 BM25 搜索不需要 Embedding，除非选用的 ReMe 配置启用了依赖 Embedding 的向量检索。 ## 生命周期和失败行为 -- `prefetch` 调用 ReMe `search`，Hermes 将结果放入受保护的 memory context； -- `sync_turn` 把完成的回合加入串行后台写队列，再调用 `auto_memory`； +- `prefetch` 只返回 ReMe answer，受保护的 memory context 包装由 Hermes 统一添加； +- 有召回内容时，Hermes UI 会显示 ReMe recall indicator 和可获得的结果数量； +- `sync_turn` 只提交最新完整回合，并用 profile 与 session 共同生成安全 ID； - cron、flush 和 subagent context 不写入对话记忆； -- 健康检查失败后，在 cooldown 结束前暂停召回和记录； -- 召回与写入有独立 cooldown，单项失败不会关闭另一项； -- 召回使用较短超时，避免慢搜索长期阻塞模型调用； -- shutdown 会在有限时间内排空写队列，ReMe 服务仍由用户独立管理。 +- backend 健康、召回和写入使用独立 cooldown；错误只记录警告，不中断 Hermes 对话； +- 写队列目前位于内存，进程异常退出时尚未完成的写入可能丢失。 -英文权威安装说明位于 `integrations/hermes_agent/README.md`。 +运行 `hermes memory status` 检查插件状态，然后启动新的 Hermes 会话。完整字段、真实截图和故障排查见 +[`integrations/hermes_agent/README_ZH.md`](../../../integrations/hermes_agent/README_ZH.md)。 diff --git a/docs/zh/reme-blog.md b/docs/zh/reme-blog.md index 1af0d05f..43c0b9a7 100644 --- a/docs/zh/reme-blog.md +++ b/docs/zh/reme-blog.md @@ -338,7 +338,7 @@ ReMe 既可以作为本地记忆服务，通过 CLI、HTTP API 或 MCP Server | **OpenClaw** | 将 [`@agentscope-ai/reme`](../../typescript/README_ZH.md#openclaw) 安装为原生 memory plugin。 | 根 Agent 对话运行前召回、显式搜索、自动捕获对话，以及定时 Auto Dream。 | | **QwenPaw** | 通过 Python API 在进程内嵌入 ReMe。 | 复用宿主应用的生命周期和模型配置，同时保持记忆本地、文件化。 | | **Claude Code** | 启动 streamable HTTP MCP Service，并安装 [`integrations/claude_code/reme`](../../integrations/claude_code/reme)。 | MCP 记忆召回工具、`reme-memory` skill，以及自动记录会话的 Stop hook。 | -| **Hermes** | 启动 HTTP Service，并安装 [`integrations/hermes_agent`](../../integrations/hermes_agent)。 | 在模型调用前自动召回相关记忆，并在每轮对话完成后异步调用 `auto_memory`。 | +| **Hermes** | 安装 [`integrations/hermes_agent`](../../integrations/hermes_agent)，并选择 HTTP 或 Embedded 模式。 | 在模型调用前自动召回相关记忆，并在每轮对话完成后异步调用 `auto_memory`。 | | **Codex 等支持 CLI 的 Agent** | 复制或安装 [`skills/reme_memory/SKILL.md`](../../skills/reme_memory/SKILL.md)。 | 通过 CLI 搜索、读取和写入记忆；自动记录需要宿主 Agent 显式接入会话生命周期。 | 安装、配置与集成演示可查看 [README 中文版](../../README_ZH.md)。 diff --git a/integrations/README.md b/integrations/README.md deleted file mode 100644 index d4ada720..00000000 --- a/integrations/README.md +++ /dev/null @@ -1,11 +0,0 @@ -# Agent Integrations - -This directory contains host-specific adapters that connect external agents to ReMe. An integration may use the host's -plugin API, hooks, MCP configuration, or client interface, but it does not extend ReMe's runtime through the -`reme.plugins` entry-point group. - -The shared TypeScript client and the DeepSeek Harness and OpenClaw adapters live in -[`../typescript`](../typescript/README.md). - -Installable extensions of ReMe itself include [Auto Fin](../plugins/auto-fin/README.md) and -[Daily Paper](../plugins/daily_paper/README.md). diff --git a/integrations/hermes_agent/README.md b/integrations/hermes_agent/README.md index 68fffccb..0e95a1ea 100644 --- a/integrations/hermes_agent/README.md +++ b/integrations/hermes_agent/README.md @@ -1,59 +1,167 @@ # ReMe memory provider for Hermes Agent -This plugin connects Hermes Agent to a running ReMe HTTP service. It recalls -relevant memory before each model call and records each completed turn through -ReMe's automatic memory job. +[中文说明](README_ZH.md) -## Prerequisites +This plugin gives Hermes Agent automatic ReMe recall and recording in either of +two modes: -- Python 3.11 or newer -- A working Hermes Agent installation -- ReMe installed with its core dependencies -- One ReMe workspace and endpoint for each Hermes profile that should remain - isolated +- **HTTP** (default) connects to an independently managed ReMe service and does + not require the ReMe SDK in Hermes' Python environment. +- **Embedded** creates a ReMe `Application` inside Hermes on a dedicated asyncio + loop thread. It needs `reme-ai` installed but no service process or port. -ReMe search currently covers one whole workspace. Pointing multiple Hermes -profiles at the same ReMe workspace therefore shares their recalled memory. Use -a separate ReMe workspace and endpoint when profiles must be isolated. +Both modes search before a model call and queue each completed user/assistant +turn for automatic memory extraction. ReMe remains local-first: workspace files +are the durable source of truth. -## Start ReMe +```text +new Hermes turn + └─ ReMe prefetch → search → protected memory context → model call -Start the HTTP service against a workspace dedicated to the active Hermes -profile: - -```bash -reme start \ - workspace_dir="$HOME/.reme-hermes-default" \ - service.backend=http \ - service.host=127.0.0.1 \ - service.port=2333 +completed user/assistant turn + └─ FIFO writer → auto_memory → workspace daily Markdown ``` -ReMe needs a working LLM configuration for automatic memory extraction. Its -default search uses BM25, so embedding credentials are optional unless vector -retrieval is enabled. Keep the service running while Hermes is active. +Unlike the DSH adapter, this provider does not add a model-visible search tool. +Hermes calls `prefetch()` automatically before every relevant model call and +adds the returned evidence to its protected memory context. + +| Mode | ReMe process | Hermes dependency | Best for | +| --- | --- | --- | --- | +| HTTP | Separate `reme start` service | No ReMe SDK required | Process isolation, shared or independently managed services | +| Embedded | Inside Hermes on a dedicated event-loop thread | `reme-ai[core]` | Simple local setup with no extra service or port | + +## Requirements + +- Python 3.11 or newer. +- Hermes Agent 0.21 or newer. +- A ReMe configuration with `health_check`, `search`, and `auto_memory` jobs. +- A working ReMe model configuration for `auto_memory`; BM25 recall itself does + not require an embedding model. ## Install and configure -Hermes supports installing a plugin from a repository subdirectory: +Hermes can install the plugin directly from this repository subdirectory: ```bash hermes plugins install agentscope-ai/ReMe/integrations/hermes_agent hermes memory setup ``` -Select `reme`, accept `http://127.0.0.1:2333` or enter the endpoint used above. -Setup calls ReMe `health_check` and only replaces an existing provider config -after the endpoint reports healthy. Then start a new Hermes session. -Configuration is stored in -`$HERMES_HOME/reme.json`, so every Hermes profile can point to its own ReMe -workspace. +For development from local ReMe and Hermes checkouts, either copy the integration +into the active profile or link it as a project-local plugin. The link keeps +Hermes on the exact ReMe source being edited: -The file supports these optional settings: +```bash +mkdir -p "$HERMES_HOME/plugins/reme" +cp -R /path/to/ReMe/integrations/hermes_agent/. "$HERMES_HOME/plugins/reme/" +hermes plugins enable reme +hermes config set memory.provider reme +``` + +```bash +cd /path/to/hermes-agent +mkdir -p .hermes/plugins +ln -s /path/to/ReMe/integrations/hermes_agent .hermes/plugins/reme +export HERMES_ENABLE_PROJECT_PLUGINS=1 +hermes config set memory.provider reme +``` + +`.hermes/` is Hermes runtime state and is ignored by the Hermes repository. Do +not commit the link, profile configuration, conversations, or generated memory. + +Verify discovery before starting a conversation: + +```bash +hermes plugins doctor /path/to/ReMe/integrations/hermes_agent --ci +hermes memory status +``` + +The Hermes Dashboard renders the provider's mode-specific fields and advanced +recall, health, write, and shutdown controls. Select **Plugins → Runtime provider +plugins → Memory provider → reme**. The active mode controls whether the HTTP +endpoint or embedded workspace settings are shown. + +![ReMe selected as the active Hermes memory provider](figures/hermes-provider-settings.png) + +Configuration is profile-local at: + +```text +$HERMES_HOME/reme/config.json +``` + +The earlier `$HERMES_HOME/reme.json` location remains a fallback for fields +omitted from the current file. Current values take precedence, and the next CLI +setup save writes a complete current config without deleting the legacy file. + +## HTTP mode + +Install ReMe in the environment that will run its service, then start one +service and workspace for the active Hermes profile: + +```bash +reme start \ + workspace_dir="$HOME/.reme-hermes-default" \ + service.backend=http \ + service.host=127.0.0.1 \ + service.port=3456 +``` + +Use this configuration in Hermes: ```json { + "mode": "http", + "endpoint": "http://127.0.0.1:3456" +} +``` + +Port `3456` is used in this guide so the verification service does not collide +with another ReMe instance on the default `2333` port. It is not a new default. + +The setup wizard checks `health_check` before replacing a valid configuration. +The generic desktop settings endpoint validates field types and choices; a new +Hermes session validates the URL and performs the live health check. ReMe's action HTTP service has no +integration-specific authentication, so keep it on loopback or place it behind +a trusted tunnel or authenticated proxy. + +## Embedded mode + +Install ReMe into the same Python environment used by Hermes: + +```bash +pip install "reme-ai[core]" +``` + +Then configure a dedicated workspace: + +```json +{ + "mode": "embedded", + "workspace_dir": "~/.reme-hermes-default", + "reme_config": "default" +} +``` + +Embedded mode resolves the named ReMe config, overrides its `workspace_dir`, +constructs and starts `reme.Application` on one long-lived event loop, and calls +the `health_check`, `search`, and `auto_memory` jobs directly. Shutdown drains +the Hermes write queue, closes the Application, stops the loop, and joins its +thread within the configured timeout. It never calls `Application.run_app()` and +therefore never opens a service port. + +ReMe needs a working model configuration for automatic memory extraction. The +default search includes BM25, so embedding credentials are optional unless the +selected ReMe config enables vector retrieval that requires them. + +## Full configuration + +```json +{ + "mode": "http", "endpoint": "http://127.0.0.1:2333", + "workspace_dir": "", + "reme_config": "default", "request_timeout": 600.0, "recall_timeout": 5.0, "health_timeout": 2.0, @@ -63,22 +171,148 @@ The file supports these optional settings: } ``` -Run `hermes memory status` to check that the provider is installed and -configured. Starting a Hermes session performs a fresh endpoint health check. +| Field | Default | Meaning | +| --- | --- | --- | +| `mode` | `http` | `http` or `embedded`. Missing values preserve legacy HTTP behavior. | +| `endpoint` | `http://127.0.0.1:2333` | Absolute HTTP(S) service URL; credentials, query strings, and fragments are rejected. | +| `workspace_dir` | empty | Required in embedded mode and normalized to an absolute path. | +| `reme_config` | `default` | Built-in ReMe config name or YAML/JSON path for embedded mode. | +| `recall_limit` | `5` | Maximum number of search results requested before a model call. | +| `recall_timeout` | `5` | Maximum foreground recall time in seconds. | +| `request_timeout` | `600` | Embedded startup and `auto_memory` write timeout in seconds. | +| `health_timeout` | `2` | Health-probe timeout in seconds. | +| `health_retry_seconds` | `30` | Cooldown before retrying an unavailable backend. | +| `shutdown_timeout` | `30` | Total queue-drain and backend-close budget in seconds. | + +All numeric values must be finite and positive. + +ReMe search covers an entire workspace. Give each Hermes profile a different +workspace unless cross-profile recall is intentional. HTTP profiles normally +use different ports as well. ## Lifecycle and failure behavior -- `prefetch` calls ReMe `search` and returns only its recalled text. Hermes wraps - that text in its protected memory-context block. -- `sync_turn` queues the completed user/assistant turn for a serial background - writer, which calls ReMe `auto_memory` with a filename-safe ID derived from - the Hermes profile and conversation. -- Cron, flush, and subagent contexts do not write conversational memory. -- A failed health check disables recall and recording until the retry cooldown - expires. Retrieval and recording failures use independent cooldowns, so one - action cannot disable the other while the ReMe service remains healthy. -- Recall uses its own short timeout so a slow ReMe search cannot stall the - Hermes model call for the longer automatic-memory timeout. -- `shutdown` gives queued writes a bounded drain interval; ReMe remains an - independently managed service. An idempotent process-exit hook uses the same - drain path when a Hermes surface does not call provider shutdown directly. +- `prefetch` returns only ReMe's answer; Hermes adds the protected + `` wrapper. +- Successful recall exposes a Hermes recall indicator using ReMe's returned + result count when available. +- `sync_turn` sends only the latest completed turn and uses a filename-safe ID + derived from both the Hermes profile and session. +- Cron, flush, and subagent contexts do not record conversational memory. +- Health, recall, and write cooldowns are independent. Backend failures log a + warning and do not fail the main Hermes conversation. +- The writer is FIFO and in-memory. A process crash can lose queued turns; + persistent spooling is intentionally outside this first dual-mode version. + +Run `hermes memory status` after installation, then start a new Hermes session. + +## Verified end-to-end behavior + +The screenshots below were captured with Computer Use from real English Hermes +0.21.1 and ReMe Studio 0.4.1.11 interfaces. The conversations used an +OpenAI-compatible model endpoint. Each mode used an isolated temporary Hermes +profile and ReMe workspace. The first session recorded a synthetic fact through +`auto_memory`; a fresh session then recovered it through automatic `prefetch`. +No API keys, `.env` contents, browser chrome, or personal memories appear in the +images. + +### Reproduce the verification + +Run the focused compatibility suite against the Hermes checkout, then validate +the plugin through Hermes' real loader: + +```bash +cd /path/to/ReMe +PYTHONPATH=/path/to/hermes-agent \ + pytest tests/unit/test_hermes_agent_integration.py -v + +cd /path/to/hermes-agent +hermes plugins doctor /path/to/ReMe/integrations/hermes_agent --ci +hermes memory status +``` + +For a live model test, start ReMe on `3456`, select that endpoint in the ReMe +provider settings, and use two new Hermes sessions: the first asks Hermes to +remember a synthetic fact and the second asks for it back. When an existing +ReMe `.env` uses `LLM_*` names, map them only in the process environment used +for verification: + +```bash +set -a +source /path/to/ReMe/.env +set +a +export OPENAI_API_KEY="$LLM_API_KEY" +export OPENAI_BASE_URL="$LLM_BASE_URL" + +hermes --provider openai-api -m "$LLM_MODEL_NAME" -z \ + "Remember this synthetic fact for a later session: Project Juniper's weekly review is Thursday at 14:30 UTC." +hermes --provider openai-api -m "$LLM_MODEL_NAME" -z \ + "From long-term memory, when is Project Juniper's weekly review?" + +unset OPENAI_API_KEY OPENAI_BASE_URL +``` + +The second command must run without `--resume`. Confirm that a new Markdown +note exists under the selected ReMe workspace's `daily/` directory; do not use +the ReMe repository's `.reme/` directory for this check. Never print the loaded +variables or save the mapped credentials in Hermes configuration. + +### Provider configuration + +![ReMe selected as the active HTTP memory provider on port 3456](figures/hermes-provider-settings.png) + +### Two independent Hermes sessions + +![The write session and fresh recall session in the Hermes session overview](figures/hermes-http-sessions.png) + +### HTTP mode recall + +![A fresh Hermes session recalls the HTTP-mode verification fact from ReMe](figures/hermes-http-recall.png) + +### File-native durable result + +![The generated Project Juniper Markdown in the English ReMe Studio interface](figures/hermes-reme-daily-note.png) + +### Embedded mode recall + +![A fresh Hermes session recalls the embedded-mode verification fact from ReMe](figures/hermes-embedded-recall.png) + +## Troubleshooting + +### `hermes memory status` shows built-in memory only + +Enable the plugin and select it explicitly, then start a new session: + +```bash +hermes plugins enable reme +hermes config set memory.provider reme +hermes memory status +``` + +### HTTP mode is unavailable + +- Confirm `reme start` is still running and its port matches `endpoint`. +- Call `POST http://127.0.0.1:3456/health_check` and inspect ReMe service logs. +- In containers or on another host, remember that `127.0.0.1` refers to the + Hermes machine; use a trusted tunnel or authenticated proxy. + +### Embedded mode is unavailable + +- Install `reme-ai[core]` into the exact Python environment that launches Hermes. +- Use an absolute, writable workspace outside the source repository. +- Check that the selected ReMe config exposes all three required jobs. + +### Recall works but completed turns are not written + +- `cron`, `flush`, and `subagent` contexts intentionally skip writes. +- Wait for the asynchronous writer before inspecting `daily/`. +- Check Hermes and ReMe logs for an `auto_memory` error or cooldown warning. +- The current FIFO queue is in memory; an abrupt process exit may lose pending writes. + +### Recall returns no useful context + +- Confirm the expected Markdown exists under the configured workspace's + `daily/` or `digest/` directory. +- Use a focused prompt and increase `recall_limit` only when necessary. +- Rebuild derived ReMe indexes from workspace files; never rewrite source memory + merely to repair an index. diff --git a/integrations/hermes_agent/README_ZH.md b/integrations/hermes_agent/README_ZH.md new file mode 100644 index 00000000..7a8c5303 --- /dev/null +++ b/integrations/hermes_agent/README_ZH.md @@ -0,0 +1,263 @@ +# Hermes Agent 集成 + +[English](README.md) + +ReMe 的 Hermes memory provider 支持两种运行方式： + +- HTTP（默认）：连接独立运行的 ReMe 服务，Hermes 环境不需要安装 ReMe SDK； +- Embedded：在 Hermes Python 进程内创建 ReMe `Application`，不需要服务进程和端口。 + +两种模式都会在模型调用前执行 `search`，并在用户/助手回合完成后把 `auto_memory` 放入串行后台队列。 + +```text +Hermes 新回合 + └─ ReMe prefetch → search → 受保护 memory context → 模型调用 + +完整 user/assistant 回合 + └─ FIFO 写入队列 → auto_memory → workspace daily Markdown +``` + +与 DSH 适配器不同，这个 provider 不增加模型可见的搜索工具。Hermes 会在相关模型调用前自动执行 `prefetch()`，再把 +ReMe 返回的证据加入受保护的 memory context。 + +| 模式 | ReMe 运行位置 | Hermes 环境依赖 | 适用场景 | +| --- | --- | --- | --- | +| HTTP | 独立 `reme start` 服务 | 不需要 ReMe SDK | 进程隔离、共享服务或独立运维 | +| Embedded | Hermes 进程内的专用 event-loop thread | `reme-ai[core]` | 单机使用，不希望维护服务和端口 | + +## 环境要求 + +- Python 3.11 或更高版本； +- Hermes Agent 0.21 或更高版本； +- ReMe 配置包含 `health_check`、`search`、`auto_memory` Job； +- `auto_memory` 需要可用的 ReMe 模型配置；仅 BM25 召回不要求 Embedding 模型。 + +## 安装与配置 + +```bash +hermes plugins install agentscope-ai/ReMe/integrations/hermes_agent +hermes memory setup +``` + +本地同时开发 ReMe 和 Hermes 时，可以复制当前 checkout，也可以把它链接为 project-local plugin。软链接会让 Hermes 始终 +运行当前正在编辑的 ReMe 源码： + +```bash +mkdir -p "$HERMES_HOME/plugins/reme" +cp -R /path/to/ReMe/integrations/hermes_agent/. "$HERMES_HOME/plugins/reme/" +hermes plugins enable reme +hermes config set memory.provider reme +``` + +```bash +cd /path/to/hermes-agent +mkdir -p .hermes/plugins +ln -s /path/to/ReMe/integrations/hermes_agent .hermes/plugins/reme +export HERMES_ENABLE_PROJECT_PLUGINS=1 +hermes config set memory.provider reme +``` + +Hermes 仓库会忽略 `.hermes/` 运行状态。不要提交这个链接、profile 配置、对话记录或生成的记忆文件。 + +启动对话前检查真实发现路径： + +```bash +hermes plugins doctor /path/to/ReMe/integrations/hermes_agent --ci +hermes memory status +``` + +Hermes Dashboard 的 **Plugins → Runtime provider plugins → Memory provider → reme** 会展示模式相关字段以及召回、健康、 +写入和关闭的高级设置。选择不同模式时，只显示 HTTP endpoint 或 Embedded workspace 对应字段。 + +![Hermes 中处于 ready 和 active 状态的 ReMe provider](figures/hermes-provider-settings.png) + +配置主路径为 `$HERMES_HOME/reme/config.json`。新文件未包含的字段仍会从旧 `$HERMES_HOME/reme.json` 继承，新文件中的 +值优先；后续 CLI 保存会写入完整的新配置，但不会删除旧文件。 + +## HTTP 模式 + +为当前 Hermes profile 启动独立 workspace： + +```bash +reme start \ + workspace_dir="/absolute/path/to/reme-hermes-default" \ + service.backend=http \ + service.host=127.0.0.1 \ + service.port=3456 +``` + +配置示例： + +```json +{ + "mode": "http", + "endpoint": "http://127.0.0.1:3456" +} +``` + +本文使用 `3456` 做联调端口，以免与默认 `2333` 上的其他 ReMe 实例冲突；这不是修改 ReMe 默认端口。 + +终端 setup 会在保存前调用 `health_check`。HTTP action 接口没有本集成专用的认证头，不要直接暴露到公网；跨主机使用时 +应放在可信网络、SSH tunnel 或带认证的反向代理后。 + +## Embedded 模式 + +先把 ReMe 安装到 Hermes 使用的同一个 Python 环境： + +```bash +pip install "reme-ai[core]" +``` + +然后配置独立 workspace： + +```json +{ + "mode": "embedded", + "workspace_dir": "~/.reme-hermes-default", + "reme_config": "default" +} +``` + +插件在专用 asyncio loop thread 上构造并启动 `reme.Application`，直接执行 `health_check`、`search` 和 `auto_memory`。 +关闭时会在有界时间内排空写队列、调用 `Application.close()`、停止 loop 并 join thread；不会调用 +`Application.run_app()`，因此不会监听端口。 + +## 完整配置 + +```json +{ + "mode": "http", + "endpoint": "http://127.0.0.1:2333", + "workspace_dir": "", + "reme_config": "default", + "request_timeout": 600.0, + "recall_timeout": 5.0, + "health_timeout": 2.0, + "health_retry_seconds": 30.0, + "shutdown_timeout": 30.0, + "recall_limit": 5 +} +``` + +| 字段 | 默认值 | 说明 | +| --- | --- | --- | +| `mode` | `http` | `http` 或 `embedded`；缺失时保持旧版 HTTP 行为。 | +| `endpoint` | `http://127.0.0.1:2333` | HTTP(S) 绝对地址；拒绝 URL 凭据、query 和 fragment。 | +| `workspace_dir` | 空 | Embedded 模式必填，并规范化为绝对路径。 | +| `reme_config` | `default` | Embedded 使用的内置配置名或 YAML/JSON 路径。 | +| `recall_limit` | `5` | 每次模型调用前请求的最大检索结果数。 | +| `recall_timeout` | `5` | 前台召回最长等待秒数。 | +| `request_timeout` | `600` | Embedded 启动和 `auto_memory` 写入超时。 | +| `health_timeout` | `2` | 健康检查超时。 | +| `health_retry_seconds` | `30` | backend 不可用后的重试冷却时间。 | +| `shutdown_timeout` | `30` | 排空队列和关闭 backend 的总时间预算。 | + +所有数值都必须是有限正数。 + +## Workspace 隔离 + +ReMe 搜索覆盖整个 workspace。多个 Hermes profile 指向同一个 workspace 时会共享召回结果；除非明确需要共享，否则为 +每个 profile 配置不同 workspace。HTTP 模式通常也使用不同端口。 + +自动记忆需要可用的 LLM 配置；默认 BM25 搜索不需要 Embedding，除非选用的 ReMe 配置启用了依赖 Embedding 的向量检索。 + +## 生命周期和失败行为 + +- `prefetch` 只返回 ReMe answer，受保护的 memory context 包装由 Hermes 统一添加； +- 有召回内容时，Hermes UI 会显示 ReMe recall indicator 和可获得的结果数量； +- `sync_turn` 只提交最新完整回合，并用 profile 与 session 共同生成安全 ID； +- cron、flush 和 subagent context 不写入对话记忆； +- backend 健康、召回和写入使用独立 cooldown；错误只记录警告，不中断 Hermes 对话； +- 写队列目前位于内存，进程异常退出时尚未完成的写入可能丢失。 + +## 真实端到端验证 + +以下截图使用 Computer Use 从真实英文 Hermes 0.21.1 与 ReMe Studio 0.4.1.11 界面取得，模型通过 +OpenAI-compatible 接口调用。HTTP 与 Embedded 分别使用隔离的临时 Hermes profile 和 ReMe workspace：第一个会话通过 +`auto_memory` 写入合成事实，第二个全新会话通过自动 `prefetch` 召回。图片不包含 API Key、`.env` 内容、浏览器外框或 +真实个人记忆。 + +### 复现验证 + +先让聚焦单测使用真实 Hermes 源码，再通过 Hermes 自身的 loader 验证插件发现和注册： + +```bash +cd /path/to/ReMe +PYTHONPATH=/path/to/hermes-agent \ + pytest tests/unit/test_hermes_agent_integration.py -v + +cd /path/to/hermes-agent +hermes plugins doctor /path/to/ReMe/integrations/hermes_agent --ci +hermes memory status +``` + +真实模型验证时，在 `3456` 启动 ReMe，把 provider endpoint 指向该服务，并使用两个全新的 Hermes 会话：第一个会话要求 +记住一条虚构事实，第二个会话要求重新回答。如果现有 ReMe `.env` 使用 `LLM_*` 变量，只在验证进程环境中映射它们： + +```bash +set -a +source /path/to/ReMe/.env +set +a +export OPENAI_API_KEY="$LLM_API_KEY" +export OPENAI_BASE_URL="$LLM_BASE_URL" + +hermes --provider openai-api -m "$LLM_MODEL_NAME" -z \ + "Remember this synthetic fact for a later session: Project Juniper's weekly review is Thursday at 14:30 UTC." +hermes --provider openai-api -m "$LLM_MODEL_NAME" -z \ + "From long-term memory, when is Project Juniper's weekly review?" + +unset OPENAI_API_KEY OPENAI_BASE_URL +``` + +第二条命令不要使用 `--resume`。随后确认所选 ReMe workspace 的 `daily/` 下出现新的 Markdown；不要把 ReMe 仓库自身的 +`.reme/` 用作测试 workspace。不要打印这些变量，也不要把映射后的凭据保存到 Hermes 配置。 + +### Provider 配置 + +![Hermes 中使用 3456 端口的 ReMe HTTP provider](figures/hermes-provider-settings.png) + +### 两个相互独立的 Hermes 会话 + +![Hermes 会话总览中的写入会话和全新召回会话](figures/hermes-http-sessions.png) + +### HTTP 模式召回 + +![Hermes 新会话从 ReMe 召回 HTTP 模式验证事实](figures/hermes-http-recall.png) + +### 文件原生的持久化结果 + +![英文 ReMe Studio 中生成的 Project Juniper Markdown](figures/hermes-reme-daily-note.png) + +### Embedded 模式召回 + +![Hermes 新会话从 ReMe 召回 Embedded 模式验证事实](figures/hermes-embedded-recall.png) + +## 常见问题 + +### `hermes memory status` 仍显示 built-in only + +```bash +hermes plugins enable reme +hermes config set memory.provider reme +``` + +配置变更只对新会话生效。 + +### HTTP 模式显示 backend unavailable + +- 确认 `reme start` 仍在运行，端口与 `endpoint` 一致； +- 直接执行 `curl -s http://127.0.0.1:3456/health_check -X POST -H 'Content-Type: application/json' -d '{}'`； +- 容器或跨机器部署时，`127.0.0.1` 指向各自本机； +- 不要把无认证的 ReMe HTTP action 服务直接暴露到公网。 + +### Embedded 模式提示缺少 SDK + +必须把 `reme-ai[core]` 安装到 Hermes 实际使用的 Python 环境，而不是另一个虚拟环境。用 +`python -c 'import reme; print(reme.__version__)'` 核对。 + +### 对话完成但 workspace 尚未出现新记录 + +写入由后台 FIFO 队列执行。正常退出会在 `shutdown_timeout` 内排空，但进程崩溃仍可能丢失尚未写入的回合。检查 Hermes +日志、ReMe 模型配置和 workspace 的 `daily/` 目录。 + +运行 `hermes memory status` 检查插件状态，然后启动新的 Hermes 会话。 diff --git a/integrations/hermes_agent/__init__.py b/integrations/hermes_agent/__init__.py index 8139cb12..2438923e 100644 --- a/integrations/hermes_agent/__init__.py +++ b/integrations/hermes_agent/__init__.py @@ -1,75 +1,30 @@ -"""Hermes Agent memory provider backed by a running ReMe HTTP service.""" +"""Hermes Agent memory provider backed by HTTP or an embedded ReMe SDK.""" from __future__ import annotations import atexit +import contextvars import hashlib -import json +import importlib.util import logging -import os import queue import re -import tempfile import threading import time -from pathlib import Path +from dataclasses import replace from typing import Any, Dict, List, Optional -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, RecallStatus -from .client import ReMeHttpClient, ReMeServiceError +from .backend import ReMeBackend, ReMeBackendError +from .config import ReMeConfig, ReMeConfigError, load_config, save_config +from .embedded_backend import EmbeddedReMeBackend +from .http_backend import HttpReMeBackend logger = logging.getLogger(__name__) - -_CONFIG_FILENAME = "reme.json" -_DEFAULT_CONFIG: dict[str, Any] = { - "endpoint": "http://127.0.0.1:2333", - "request_timeout": 600.0, - "recall_timeout": 5.0, - "health_timeout": 2.0, - "health_retry_seconds": 30.0, - "shutdown_timeout": 30.0, - "recall_limit": 5, -} _NON_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9._-]+") - - -def _config_path(hermes_home: str | Path | None = None) -> Path: - if hermes_home is None: - from hermes_constants import get_hermes_home - - hermes_home = get_hermes_home() - return Path(hermes_home).expanduser() / _CONFIG_FILENAME - - -def _load_config(hermes_home: str | Path | None = None) -> dict[str, Any]: - config = dict(_DEFAULT_CONFIG) - path = _config_path(hermes_home) - if not path.is_file(): - return config - try: - loaded = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: - logger.warning("Unable to read ReMe provider config %s: %s", path, exc) - return config - if isinstance(loaded, dict): - config.update({key: value for key, value in loaded.items() if value is not None and value != ""}) - return config - - -def _positive_float(config: dict[str, Any], key: str) -> float: - try: - return max(0.1, float(config[key])) - except (KeyError, TypeError, ValueError): - return float(_DEFAULT_CONFIG[key]) - - -def _positive_int(config: dict[str, Any], key: str) -> int: - try: - return max(1, int(config[key])) - except (KeyError, TypeError, ValueError): - return int(_DEFAULT_CONFIG[key]) +_SDK_INSTALL_HINT = 'Embedded ReMe mode requires the SDK. Install it with: pip install "reme-ai[core]"' def _slug(value: str, fallback: str, *, limit: int) -> str: @@ -81,22 +36,35 @@ def _scoped_session_id(profile_id: str, session_id: str) -> str: """Create a readable, filename-safe ID without allowing scope collisions.""" profile = str(profile_id or "default") session = str(session_id or "session") - digest = hashlib.sha256(f"{profile}\0{session}".encode("utf-8")).hexdigest()[:12] + digest = hashlib.sha256(f"{profile}\0{session}".encode()).hexdigest()[:12] return f"hermes-{_slug(profile, 'default', limit=32)}-{_slug(session, 'session', limit=64)}-{digest}" +def _backend_for(config: ReMeConfig) -> ReMeBackend: + if config.mode == "embedded": + return EmbeddedReMeBackend( + config.workspace_dir, + reme_config=config.reme_config, + start_timeout=config.request_timeout, + ) + return HttpReMeBackend(config.endpoint, request_timeout=config.request_timeout) + + class ReMeMemoryProvider(MemoryProvider): """Use ReMe for automatic cross-session recall and recording in Hermes.""" def __init__(self) -> None: - self._client: ReMeHttpClient | None = None - self._endpoint = str(_DEFAULT_CONFIG["endpoint"]) - self._recall_timeout = float(_DEFAULT_CONFIG["recall_timeout"]) - self._health_timeout = float(_DEFAULT_CONFIG["health_timeout"]) - self._health_retry_seconds = float(_DEFAULT_CONFIG["health_retry_seconds"]) - self._shutdown_timeout = float(_DEFAULT_CONFIG["shutdown_timeout"]) - self._recall_limit = int(_DEFAULT_CONFIG["recall_limit"]) - self._service_available = False + defaults = ReMeConfig() + self._backend: ReMeBackend | None = None + self._config: ReMeConfig | None = None + self._backend_label = defaults.endpoint + self._recall_timeout = defaults.recall_timeout + self._health_timeout = defaults.health_timeout + self._health_retry_seconds = defaults.health_retry_seconds + self._shutdown_timeout = defaults.shutdown_timeout + self._request_timeout = defaults.request_timeout + self._recall_limit = defaults.recall_limit + self._backend_available = False self._next_health_probe = 0.0 self._next_recall_attempt = 0.0 self._next_write_attempt = 0.0 @@ -107,8 +75,12 @@ class ReMeMemoryProvider(MemoryProvider): self._write_queue: queue.Queue[dict[str, Any] | None] = queue.Queue() self._write_thread: threading.Thread | None = None self._write_thread_lock = threading.Lock() + self._backend_lock = threading.RLock() self._shutdown_started = False + self._deferred_backend_close = False self._atexit_registered = False + self._recall_status: RecallStatus | None = None + self._unavailable_reason = "" @property def name(self) -> str: @@ -116,32 +88,56 @@ class ReMeMemoryProvider(MemoryProvider): return "reme" def is_available(self) -> bool: - """Check local configuration only; network probes belong to initialize().""" + """Check configuration and local dependencies without network or writes.""" try: - config = _load_config() - ReMeHttpClient(str(config["endpoint"]), timeout=_positive_float(config, "request_timeout")) - return True - except (KeyError, TypeError, ValueError, OSError): + config = load_config() + if config.mode == "embedded" and importlib.util.find_spec("reme") is None: + self._unavailable_reason = _SDK_INSTALL_HINT + return False + _backend_for(config) + except (ReMeConfigError, TypeError, ValueError, OSError) as exc: + self._unavailable_reason = str(exc) return False + self._unavailable_reason = "" + return True + + def unavailable_reason(self) -> str: + """Return the last local availability failure as user-facing guidance.""" + return self._unavailable_reason def initialize(self, session_id: str, **kwargs: Any) -> None: - """Load profile configuration and probe ReMe without blocking startup.""" + """Load profile config and start the selected backend best-effort.""" with self._write_thread_lock: if self._write_thread is not None and self._write_thread.is_alive(): - raise RuntimeError("Cannot reinitialize ReMe while its previous writer is still running") + raise RuntimeError( + "Cannot reinitialize ReMe while its previous writer is still running", + ) hermes_home = str(kwargs.get("hermes_home") or "") or None - config = _load_config(hermes_home) - self._endpoint = str(config["endpoint"]) - self._recall_timeout = _positive_float(config, "recall_timeout") - self._health_timeout = _positive_float(config, "health_timeout") - self._health_retry_seconds = _positive_float(config, "health_retry_seconds") - self._shutdown_timeout = _positive_float(config, "shutdown_timeout") - self._recall_limit = _positive_int(config, "recall_limit") + try: + config = load_config(hermes_home) + except ReMeConfigError as exc: + logger.warning("ReMe provider configuration is invalid: %s", exc) + return + + with self._backend_lock: + self._close_backend_locked() + self._config = config + self._backend_label = config.endpoint if config.mode == "http" else f"embedded:{config.workspace_dir}" + self._recall_timeout = config.recall_timeout + self._health_timeout = config.health_timeout + self._health_retry_seconds = config.health_retry_seconds + self._shutdown_timeout = config.shutdown_timeout + self._request_timeout = config.request_timeout + self._recall_limit = config.recall_limit self._session_id = str(session_id or "") self._profile_id = str(kwargs.get("agent_identity") or "default") - self._write_enabled = str(kwargs.get("agent_context") or "primary") not in {"cron", "flush", "subagent"} - self._client = ReMeHttpClient(self._endpoint, timeout=_positive_float(config, "request_timeout")) - self._service_available = False + self._write_enabled = str(kwargs.get("agent_context") or "primary") not in { + "cron", + "flush", + "subagent", + } + self._backend = None + self._backend_available = False self._next_health_probe = 0.0 self._next_recall_attempt = 0.0 self._next_write_attempt = 0.0 @@ -149,77 +145,196 @@ class ReMeMemoryProvider(MemoryProvider): self._write_queue = queue.Queue() self._write_thread = None self._shutdown_started = False + self._deferred_backend_close = False + self._recall_status = None if not self._atexit_registered: atexit.register(self._atexit_shutdown) self._atexit_registered = True - - if not self._ensure_service(force=True): + if not self._ensure_backend(force=True): logger.warning( - "ReMe is unavailable at %s; recall is disabled and completed " - "turns will not be recorded until it recovers", - self._endpoint, + "ReMe is unavailable at %s; recall and recording will retry after cooldown", + self._backend_label, ) def get_config_schema(self) -> List[Dict[str, Any]]: - """Describe interactive setup fields understood by Hermes.""" + """Describe fields used by the terminal setup wizard.""" + defaults = ReMeConfig() return [ { - "key": "endpoint", - "description": "ReMe HTTP service endpoint", - "default": str(_DEFAULT_CONFIG["endpoint"]), + "key": "mode", + "label": "Mode", + "kind": "select", + "description": "How Hermes connects to ReMe", + "default": defaults.mode, + "choices": ["http", "embedded"], "required": True, }, + { + "key": "endpoint", + "label": "HTTP endpoint", + "description": "ReMe service URL (HTTP mode only)", + "default": defaults.endpoint, + "required": True, + "when": {"mode": "http"}, + }, + { + "key": "workspace_dir", + "label": "Workspace directory", + "description": "ReMe workspace (embedded mode only)", + "default": "", + "required": True, + "when": {"mode": "embedded"}, + }, + { + "key": "reme_config", + "label": "ReMe configuration", + "description": "Built-in config name or YAML/JSON path (embedded mode only)", + "default": defaults.reme_config, + "required": True, + "when": {"mode": "embedded"}, + }, + { + "key": "recall_limit", + "label": "Recall limit", + "kind": "integer", + "description": "Maximum search results injected before a model call", + "default": defaults.recall_limit, + "minimum": 1, + }, + { + "key": "recall_timeout", + "label": "Recall timeout (seconds)", + "kind": "number", + "default": defaults.recall_timeout, + "minimum": 0.1, + }, + { + "key": "request_timeout", + "label": "Write/start timeout (seconds)", + "kind": "number", + "default": defaults.request_timeout, + "minimum": 0.1, + }, + { + "key": "health_timeout", + "label": "Health timeout (seconds)", + "kind": "number", + "default": defaults.health_timeout, + "minimum": 0.1, + }, + { + "key": "health_retry_seconds", + "label": "Health retry delay (seconds)", + "kind": "number", + "default": defaults.health_retry_seconds, + "minimum": 0.1, + }, + { + "key": "shutdown_timeout", + "label": "Shutdown timeout (seconds)", + "kind": "number", + "default": defaults.shutdown_timeout, + "minimum": 0.1, + }, ] def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: - """Atomically save non-secret settings inside the active Hermes profile.""" - path = _config_path(hermes_home) - existing = _load_config(hermes_home) - existing.update({key: value for key, value in dict(values or {}).items() if value is not None and value != ""}) - - # Validate before replacing a working configuration. - client = ReMeHttpClient( - str(existing["endpoint"]), - timeout=_positive_float(existing, "request_timeout"), - ) - client.health(timeout=_positive_float(existing, "health_timeout")) - path.parent.mkdir(parents=True, exist_ok=True) - fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) - try: - with os.fdopen(fd, "w", encoding="utf-8") as handle: - json.dump(existing, handle, ensure_ascii=False, indent=2, sort_keys=True) - handle.write("\n") - handle.flush() - os.fsync(handle.fileno()) - os.chmod(tmp_name, 0o600) - os.replace(tmp_name, path) - finally: - if os.path.exists(tmp_name): - os.unlink(tmp_name) + """Validate and save terminal-wizard settings.""" + candidate = dict(values or {}) + current = load_config(hermes_home) + mode = str(candidate.get("mode", current.mode) or current.mode).strip().lower() + if mode == "http": + endpoint = str( + candidate.get("endpoint", current.endpoint) or current.endpoint, + ) + probe = HttpReMeBackend(endpoint, request_timeout=current.request_timeout) + probe.health(timeout=current.health_timeout) + elif importlib.util.find_spec("reme") is None: + raise ReMeConfigError(_SDK_INSTALL_HINT) + save_config(candidate, hermes_home) def get_tool_schemas(self) -> List[Dict[str, Any]]: - """Automatic recall and capture add no model-visible tool schemas.""" + """Automatic recall and capture add no model-visible tools.""" return [] - def prefetch(self, query: str, *, session_id: str = "") -> str: - """Recall relevant memory before Hermes sends a turn to the model.""" - del session_id - query = str(query or "").strip() - if not query or time.monotonic() < self._next_recall_attempt or not self._ensure_service(): - return "" - assert self._client is not None + def backup_paths(self) -> List[str]: + """Expose an embedded workspace to Hermes backup without starting ReMe.""" try: - response = self._client.call( - "search", - {"query": query, "limit": self._recall_limit}, - timeout=self._recall_timeout, - ) - except ReMeServiceError as exc: - self._next_recall_attempt = time.monotonic() + self._health_retry_seconds - logger.warning("ReMe retrieval failed at %s: %s", self._endpoint, exc) + config = load_config() + except ReMeConfigError: + return [] + return [config.workspace_dir] if config.mode == "embedded" and config.workspace_dir else [] + + def recall_status(self) -> Optional[RecallStatus]: + """Describe only the content injected by the latest prefetch call.""" + return self._recall_status + + def prefetch(self, query: str, *, session_id: str = "") -> str: + """Recall relevant memory before Hermes sends the turn to the model.""" + del session_id + self._recall_status = None + query = str(query or "").strip() + if not query or time.monotonic() < self._next_recall_attempt: return "" + deadline = time.monotonic() + self._recall_timeout + if not self._backend_lock.acquire( # pylint: disable=consider-using-with + timeout=max(0.0, deadline - time.monotonic()), + ): + logger.warning( + "ReMe retrieval at %s timed out waiting for the backend", + self._backend_label, + ) + return "" + try: + if not self._ensure_backend(deadline=deadline): + return "" + assert self._backend is not None + try: + remaining = deadline - time.monotonic() + if remaining <= 0: + logger.warning( + "ReMe retrieval at %s exhausted its timeout before search", + self._backend_label, + ) + return "" + response = self._backend.search( + query, + limit=self._recall_limit, + timeout=remaining, + ) + except ReMeBackendError as exc: + self._next_recall_attempt = time.monotonic() + self._health_retry_seconds + logger.warning( + "ReMe retrieval failed at %s: %s", + self._backend_label, + exc, + ) + return "" + finally: + self._close_backend_if_shutdown_locked() + finally: + self._backend_lock.release() answer = response.get("answer") - return answer.strip() if isinstance(answer, str) else "" + answer = answer.strip() if isinstance(answer, str) else "" + if answer: + self._recall_status = RecallStatus( + provider_label="ReMe", + count=self._result_count(response), + ) + return answer + + @staticmethod + def _result_count(response: dict[str, Any]) -> int: + metadata = response.get("metadata") + if not isinstance(metadata, dict): + return 0 + counts = metadata.get("counts") + if isinstance(counts, dict): + returned = counts.get("returned") + if isinstance(returned, int) and not isinstance(returned, bool) and returned >= 0: + return returned + results = metadata.get("results") + return len(results) if isinstance(results, list) else 0 def sync_turn( self, @@ -229,7 +344,7 @@ class ReMeMemoryProvider(MemoryProvider): session_id: str = "", messages: Optional[List[Dict[str, Any]]] = None, ) -> None: - """Queue one completed turn without blocking Hermes on ReMe's LLM.""" + """Queue one completed turn without blocking Hermes on memory extraction.""" del messages user = str(user_content or "").strip() assistant = str(assistant_content or "").strip() @@ -237,15 +352,10 @@ class ReMeMemoryProvider(MemoryProvider): return routed_session = str(session_id or self._session_id) if not routed_session: - logger.warning("ReMe skipped a completed turn because Hermes supplied no session id") - return - if not self._accept_writes: logger.warning( - "ReMe did not record completed turn for session %s because the provider is shutting down", - _scoped_session_id(self._profile_id, routed_session), + "ReMe skipped a completed turn because Hermes supplied no session id", ) return - payload = { "session_id": _scoped_session_id(self._profile_id, routed_session), "messages": [ @@ -255,7 +365,7 @@ class ReMeMemoryProvider(MemoryProvider): } if not self._enqueue_write(payload): logger.warning( - "ReMe did not record completed turn for session %s because the provider is shutting down", + "ReMe did not record session %s because the provider is shutting down", payload["session_id"], ) @@ -268,13 +378,14 @@ class ReMeMemoryProvider(MemoryProvider): rewound: bool = False, **kwargs: Any, ) -> None: - """Update the active conversation boundary after a Hermes switch.""" + """Route future writes to the newly active Hermes conversation.""" del parent_session_id, reset, rewound, kwargs if new_session_id: self._session_id = str(new_session_id) def shutdown(self) -> None: - """Drain queued writes for a bounded interval, then release state.""" + """Drain queued writes, then close the backend within a bounded interval.""" + deadline = time.monotonic() + self._shutdown_timeout with self._write_thread_lock: if self._shutdown_started: return @@ -284,26 +395,37 @@ class ReMeMemoryProvider(MemoryProvider): if thread is not None: self._write_queue.put(None) if thread is not None: - thread.join(timeout=self._shutdown_timeout) + thread.join(timeout=max(0.0, deadline - time.monotonic())) if thread.is_alive(): abandoned = self._discard_queued_writes() logger.warning( - "ReMe shutdown timed out after %.1fs; abandoned %d queued write(s) " - "and the in-flight write may not finish before process exit", + "ReMe shutdown timed out after %.1fs; abandoned %d queued write(s) and an in-flight write", self._shutdown_timeout, abandoned, ) - else: - self._client = None + self._deferred_backend_close = True + # A context manager cannot express the bounded wait required by shutdown. + if self._backend_lock.acquire( # pylint: disable=consider-using-with + timeout=max(0.0, deadline - time.monotonic()), + ): + try: + self._close_backend_locked( + timeout=max(0.0, deadline - time.monotonic()), + ) + self._deferred_backend_close = False + finally: + self._backend_lock.release() else: - self._client = None - self._service_available = False + logger.warning( + "ReMe backend shutdown is deferred until the in-flight operation finishes", + ) + self._backend_available = False self._next_health_probe = 0.0 def _atexit_shutdown(self) -> None: try: self.shutdown() - except Exception as exc: # pragma: no cover - interpreter teardown safety + except Exception as exc: # pragma: no cover logger.debug("ReMe atexit shutdown failed: %s", exc) def _discard_queued_writes(self) -> int: @@ -326,9 +448,10 @@ class ReMeMemoryProvider(MemoryProvider): if not self._accept_writes: return False if self._write_thread is None or not self._write_thread.is_alive(): + context = contextvars.copy_context() self._write_thread = threading.Thread( - target=self._write_loop, - args=(self._write_queue,), + target=context.run, + args=(self._write_loop, self._write_queue), daemon=True, name="reme-memory-writer", ) @@ -345,67 +468,142 @@ class ReMeMemoryProvider(MemoryProvider): return try: self._record_payload(payload) - except Exception as exc: # keep one bad response from killing the writer + except Exception as exc: logger.exception( - "Unexpected ReMe recording failure for session %s; the writer will continue: %s", - payload.get("session_id", ""), + "Unexpected ReMe recording failure; writer continues: %s", exc, ) finally: write_queue.task_done() finally: - current_thread = threading.current_thread() + current = threading.current_thread() with self._write_thread_lock: - if self._write_thread is current_thread: + if self._write_thread is current: self._write_thread = None - if not self._accept_writes: - self._client = None def _record_payload(self, payload: dict[str, Any]) -> None: if time.monotonic() < self._next_write_attempt: logger.warning( - "ReMe did not record completed turn for session %s because writes are cooling down", + "ReMe write for session %s skipped during cooldown", payload["session_id"], ) return - if not self._ensure_service(): - logger.warning( - "ReMe did not record completed turn for session %s because the service is unavailable", - payload["session_id"], - ) - return - assert self._client is not None - try: - self._client.call("auto_memory", payload) - except ReMeServiceError as exc: - self._next_write_attempt = time.monotonic() + self._health_retry_seconds - logger.warning("ReMe recording failed at %s: %s", self._endpoint, exc) - logger.warning( - "ReMe did not record completed turn for session %s", - payload["session_id"], - ) + with self._backend_lock: + if not self._ensure_backend(allow_shutdown=True): + logger.warning( + "ReMe write for session %s skipped because backend is unavailable", + payload["session_id"], + ) + return + assert self._backend is not None + try: + self._backend.auto_memory( + payload["session_id"], + payload["messages"], + timeout=self._request_timeout, + ) + except ReMeBackendError as exc: + self._next_write_attempt = time.monotonic() + self._health_retry_seconds + logger.warning( + "ReMe recording failed at %s: %s", + self._backend_label, + exc, + ) + finally: + self._close_backend_if_shutdown_locked() - def _ensure_service(self, *, force: bool = False) -> bool: - if self._client is None: - return False - if self._service_available and not force: + # pylint: disable-next=too-many-return-statements + def _ensure_backend( + self, + *, + force: bool = False, + allow_shutdown: bool = False, + deadline: float | None = None, + ) -> bool: + with self._backend_lock: + if self._backend_available and self._backend is not None and not force: + return True + if self._config is None or (self._shutdown_started and not allow_shutdown): + return False + now = time.monotonic() + if not force and now < self._next_health_probe: + return False + if self._backend is None: + try: + backend_config = self._config + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + backend_config = replace( + backend_config, + request_timeout=min(backend_config.request_timeout, remaining), + ) + self._backend = _backend_for(backend_config) + if deadline is None: + self._backend.start() + else: + self._backend.start(deadline=deadline) + except (ReMeBackendError, TypeError, ValueError, OSError) as exc: + failed, self._backend = self._backend, None + if failed is not None: + try: + cleanup_timeout = self._shutdown_timeout + if deadline is not None: + cleanup_timeout = max(0.0, deadline - time.monotonic()) + if deadline is None or cleanup_timeout > 0: + failed.close(timeout=cleanup_timeout) + except ReMeBackendError as close_exc: + logger.warning( + "Failed to clean up ReMe after startup error: %s", + close_exc, + ) + self._mark_unavailable("startup", exc) + return False + try: + health_timeout = self._health_timeout + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + health_timeout = min(health_timeout, remaining) + self._backend.health(timeout=health_timeout) + except ReMeBackendError as exc: + self._mark_unavailable("health check", exc) + return False + self._backend_available = True + self._next_health_probe = 0.0 return True - now = time.monotonic() - if not force and now < self._next_health_probe: - return False - try: - self._client.health(timeout=self._health_timeout) - except ReMeServiceError as exc: - self._mark_unavailable("health check", exc) - return False - self._service_available = True - self._next_health_probe = 0.0 - return True + + def _close_backend_if_shutdown_locked(self) -> None: + if self._deferred_backend_close: + self._close_backend_locked() + self._deferred_backend_close = False + + def _close_backend_locked(self, *, timeout: float | None = None) -> None: + backend, self._backend = self._backend, None + if backend is not None: + try: + backend.close( + timeout=self._shutdown_timeout if timeout is None else timeout, + ) + except ReMeBackendError as exc: + logger.warning( + "ReMe backend shutdown failed at %s: %s", + self._backend_label, + exc, + ) + self._backend_available = False def _mark_unavailable(self, operation: str, error: Exception) -> None: - self._service_available = False + self._backend_available = False self._next_health_probe = time.monotonic() + self._health_retry_seconds - logger.warning("ReMe %s failed at %s: %s", operation, self._endpoint, error) + logger.warning( + "ReMe %s failed at %s: %s", + operation, + self._backend_label, + error, + ) def register(ctx: Any) -> None: diff --git a/integrations/hermes_agent/backend.py b/integrations/hermes_agent/backend.py new file mode 100644 index 00000000..9922dffc --- /dev/null +++ b/integrations/hermes_agent/backend.py @@ -0,0 +1,45 @@ +"""Transport-independent backend contract for the Hermes provider.""" + +from __future__ import annotations + +from typing import Any, Protocol + + +class ReMeBackendError(RuntimeError): + """Raised when a ReMe backend cannot complete an operation.""" + + +class ReMeBackend(Protocol): + """Synchronous interface matching Hermes' memory-provider lifecycle.""" + + label: str + + def start(self, *, deadline: float | None = None) -> None: + """Start owned resources before an optional absolute deadline.""" + + def health(self, *, timeout: float) -> dict[str, Any]: + """Return a semantically healthy ReMe response.""" + + def search(self, query: str, *, limit: int, timeout: float) -> dict[str, Any]: + """Search the configured ReMe workspace.""" + + def auto_memory( + self, + session_id: str, + messages: list[dict[str, Any]], + *, + timeout: float, + ) -> dict[str, Any]: + """Record and extract memory from one completed turn.""" + + def close(self, *, timeout: float) -> None: + """Release resources within a bounded interval.""" + + +def require_healthy(response: dict[str, Any]) -> dict[str, Any]: + """Validate the semantic health flag in a successful ReMe response.""" + metadata = response.get("metadata") + health = metadata.get("health") if isinstance(metadata, dict) else None + if not isinstance(health, dict) or health.get("healthy") is not True: + raise ReMeBackendError("ReMe did not report a healthy component snapshot") + return response diff --git a/integrations/hermes_agent/client.py b/integrations/hermes_agent/client.py index 3521a217..cef5d9d8 100644 --- a/integrations/hermes_agent/client.py +++ b/integrations/hermes_agent/client.py @@ -15,15 +15,28 @@ class ReMeServiceError(RuntimeError): """Raised when a ReMe action cannot be completed successfully.""" +def normalize_http_endpoint(endpoint: str) -> str: + """Validate an HTTP service base URL without accepting embedded secrets.""" + normalized = str(endpoint or "").strip().rstrip("/") + try: + parsed = urlsplit(normalized) + _ = parsed.port + except ValueError as exc: + raise ValueError("ReMe endpoint must be an absolute http(s) URL") from exc + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError("ReMe endpoint must be an absolute http(s) URL") + if parsed.username is not None or parsed.password is not None: + raise ValueError("ReMe endpoint must be an absolute http(s) URL") + if parsed.query or parsed.fragment: + raise ValueError("ReMe endpoint must be an absolute http(s) URL") + return normalized + + class ReMeHttpClient: """Call ReMe JSON actions without adding a runtime dependency to Hermes.""" def __init__(self, endpoint: str, *, timeout: float) -> None: - endpoint = str(endpoint or "").strip().rstrip("/") - parsed = urlsplit(endpoint) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: - raise ValueError("ReMe endpoint must be an absolute http(s) URL") - self.endpoint = endpoint + self.endpoint = normalize_http_endpoint(endpoint) self.timeout = max(0.1, float(timeout)) def call( @@ -63,7 +76,9 @@ class ReMeHttpClient: if not isinstance(result, dict): raise ReMeServiceError("ReMe returned a non-object response") if result.get("success") is not True: - raise ReMeServiceError(str(result.get("answer") or "ReMe action did not report success")) + raise ReMeServiceError( + str(result.get("answer") or "ReMe action did not report success"), + ) return result def health(self, *, timeout: float) -> dict[str, Any]: diff --git a/integrations/hermes_agent/config.py b/integrations/hermes_agent/config.py new file mode 100644 index 00000000..78a5ff02 --- /dev/null +++ b/integrations/hermes_agent/config.py @@ -0,0 +1,208 @@ +"""Profile-local configuration for the Hermes ReMe memory provider.""" + +from __future__ import annotations + +import json +import math +import os +import tempfile + +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from .client import normalize_http_endpoint + +CONFIG_DIRECTORY = "reme" +CONFIG_FILENAME = "config.json" +LEGACY_CONFIG_FILENAME = "reme.json" +VALID_MODES = {"http", "embedded"} + + +class ReMeConfigError(ValueError): + """Raised when provider configuration is invalid.""" + + +@dataclass(frozen=True) +class ReMeConfig: + """Validated settings shared by the provider and both backends.""" + + mode: str = "http" + endpoint: str = "http://127.0.0.1:2333" + workspace_dir: str = "" + reme_config: str = "default" + request_timeout: float = 600.0 + recall_timeout: float = 5.0 + health_timeout: float = 2.0 + health_retry_seconds: float = 30.0 + shutdown_timeout: float = 30.0 + recall_limit: int = 5 + + +def config_path(hermes_home: str | Path) -> Path: + """Return the path used by Hermes' generic provider configuration UI.""" + return Path(hermes_home).expanduser() / CONFIG_DIRECTORY / CONFIG_FILENAME + + +def legacy_config_path(hermes_home: str | Path) -> Path: + """Return the pre-dashboard configuration path retained for compatibility.""" + return Path(hermes_home).expanduser() / LEGACY_CONFIG_FILENAME + + +def _default_hermes_home() -> Path: + from hermes_constants import get_hermes_home + + return Path(get_hermes_home()) + + +def _read_json_object(path: Path) -> dict[str, Any]: + if not path.is_file(): + return {} + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ReMeConfigError( + f"Unable to read ReMe provider config {path}: {exc}", + ) from exc + if not isinstance(loaded, dict): + raise ReMeConfigError( + f"ReMe provider config must contain a JSON object: {path}", + ) + return loaded + + +def _read_config_values(home: Path) -> dict[str, Any]: + """Merge legacy values under the current Dashboard-managed config.""" + values = _read_json_object(legacy_config_path(home)) + values.update(_read_json_object(config_path(home))) + return values + + +def _positive_float(value: Any, key: str, default: float) -> float: + if value in (None, ""): + return default + try: + number = float(value) + except (TypeError, ValueError) as exc: + raise ReMeConfigError(f"'{key}' must be a positive number") from exc + if not math.isfinite(number) or number <= 0: + raise ReMeConfigError(f"'{key}' must be a positive number") + return number + + +def _positive_int(value: Any, key: str, default: int) -> int: + if value in (None, ""): + return default + if isinstance(value, bool): + raise ReMeConfigError(f"'{key}' must be a positive integer") + try: + number = int(value) + except (TypeError, ValueError) as exc: + raise ReMeConfigError(f"'{key}' must be a positive integer") from exc + if number <= 0 or (isinstance(value, float) and not value.is_integer()): + raise ReMeConfigError(f"'{key}' must be a positive integer") + return number + + +def parse_config(values: dict[str, Any], *, hermes_home: str | Path) -> ReMeConfig: + """Validate and normalize an already loaded configuration mapping.""" + del hermes_home # Reserved for future profile-relative settings. + defaults = ReMeConfig() + mode = str(values.get("mode", defaults.mode) or defaults.mode).strip().lower() + if mode not in VALID_MODES: + raise ReMeConfigError("'mode' must be either 'http' or 'embedded'") + + endpoint = str(values.get("endpoint", defaults.endpoint) or "").strip().rstrip("/") + if mode == "http": + try: + endpoint = normalize_http_endpoint(endpoint) + except ValueError as exc: + raise ReMeConfigError( + "ReMe endpoint must be an absolute http(s) URL", + ) from exc + + raw_workspace = str( + values.get("workspace_dir", defaults.workspace_dir) or "", + ).strip() + if mode == "embedded" and not raw_workspace: + raise ReMeConfigError("'workspace_dir' is required in embedded mode") + workspace_dir = str(Path(raw_workspace).expanduser().absolute()) if raw_workspace else "" + + reme_config = str(values.get("reme_config", defaults.reme_config) or "").strip() + if mode == "embedded" and not reme_config: + raise ReMeConfigError("'reme_config' cannot be empty in embedded mode") + + return ReMeConfig( + mode=mode, + endpoint=endpoint or defaults.endpoint, + workspace_dir=workspace_dir, + reme_config=reme_config or defaults.reme_config, + request_timeout=_positive_float( + values.get("request_timeout"), + "request_timeout", + defaults.request_timeout, + ), + recall_timeout=_positive_float( + values.get("recall_timeout"), + "recall_timeout", + defaults.recall_timeout, + ), + health_timeout=_positive_float( + values.get("health_timeout"), + "health_timeout", + defaults.health_timeout, + ), + health_retry_seconds=_positive_float( + values.get("health_retry_seconds"), + "health_retry_seconds", + defaults.health_retry_seconds, + ), + shutdown_timeout=_positive_float( + values.get("shutdown_timeout"), + "shutdown_timeout", + defaults.shutdown_timeout, + ), + recall_limit=_positive_int( + values.get("recall_limit"), + "recall_limit", + defaults.recall_limit, + ), + ) + + +def load_config(hermes_home: str | Path | None = None) -> ReMeConfig: + """Load current config, inheriting omitted fields from the legacy file.""" + home = Path(hermes_home).expanduser() if hermes_home is not None else _default_hermes_home() + return parse_config(_read_config_values(home), hermes_home=home) + + +def save_config(values: dict[str, Any], hermes_home: str | Path) -> ReMeConfig: + """Validate and atomically write config in the current Hermes layout.""" + home = Path(hermes_home).expanduser() + current = config_path(home) + merged = _read_config_values(home) + merged.update( + {key: value for key, value in dict(values or {}).items() if value is not None}, + ) + validated = parse_config(merged, hermes_home=home) + + current.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f".{current.name}.", dir=current.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump( + asdict(validated), + handle, + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.chmod(tmp_name, 0o600) + os.replace(tmp_name, current) + finally: + if os.path.exists(tmp_name): + os.unlink(tmp_name) + return validated diff --git a/integrations/hermes_agent/config_schema.py b/integrations/hermes_agent/config_schema.py new file mode 100644 index 00000000..a73e4746 --- /dev/null +++ b/integrations/hermes_agent/config_schema.py @@ -0,0 +1,109 @@ +"""ReMe's configuration surface for Hermes' generic memory settings UI.""" + +# pylint: disable=no-name-in-module + +from plugins.memory.config_schema import ( + KIND_NUMBER, + KIND_SELECT, + KIND_TEXT, + ProviderConfigSchema, + ProviderField, + ProviderFieldOption, +) + +CONFIG_SCHEMA = ProviderConfigSchema( + name="reme", + label="ReMe", + docs_url="https://github.com/agentscope-ai/ReMe/tree/main/integrations/hermes_agent", + fields=( + ProviderField( + key="mode", + label="Mode", + kind=KIND_SELECT, + default="http", + description="Choose a ReMe service or an in-process ReMe SDK.", + options=( + ProviderFieldOption( + "http", + "HTTP", + "Connect to an independently managed ReMe service", + ), + ProviderFieldOption( + "embedded", + "Embedded", + "Run ReMe inside the Hermes Python process", + ), + ), + inline=True, + ), + ProviderField( + key="endpoint", + label="HTTP endpoint", + kind=KIND_TEXT, + default="http://127.0.0.1:2333", + description="Used only in HTTP mode.", + placeholder="http://127.0.0.1:2333", + inline=True, + ), + ProviderField( + key="workspace_dir", + label="Workspace directory", + kind=KIND_TEXT, + default="", + description="Required only in embedded mode. Use a separate workspace for each Hermes profile.", + placeholder="~/.reme-hermes-default", + inline=True, + ), + ProviderField( + key="reme_config", + label="ReMe configuration", + kind=KIND_TEXT, + default="default", + description="Built-in config name or a YAML/JSON path used by embedded mode.", + group="Embedded", + ), + ProviderField( + key="recall_limit", + label="Recall limit", + kind=KIND_NUMBER, + default="5", + description="Maximum number of search results included before a model call.", + group="Recall", + ), + ProviderField( + key="recall_timeout", + label="Recall timeout (seconds)", + kind=KIND_NUMBER, + default="5", + group="Timeouts", + ), + ProviderField( + key="request_timeout", + label="Write/start timeout (seconds)", + kind=KIND_NUMBER, + default="600", + group="Timeouts", + ), + ProviderField( + key="health_timeout", + label="Health timeout (seconds)", + kind=KIND_NUMBER, + default="2", + group="Timeouts", + ), + ProviderField( + key="health_retry_seconds", + label="Health retry delay (seconds)", + kind=KIND_NUMBER, + default="30", + group="Timeouts", + ), + ProviderField( + key="shutdown_timeout", + label="Shutdown timeout (seconds)", + kind=KIND_NUMBER, + default="30", + group="Timeouts", + ), + ), +) diff --git a/integrations/hermes_agent/embedded_backend.py b/integrations/hermes_agent/embedded_backend.py new file mode 100644 index 00000000..5c805147 --- /dev/null +++ b/integrations/hermes_agent/embedded_backend.py @@ -0,0 +1,323 @@ +"""In-process ReMe backend with one dedicated asyncio loop thread.""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import threading +import time + +from enum import Enum +from typing import Any, Coroutine + +from .backend import ReMeBackendError, require_healthy + + +class _State(Enum): + NEW = "new" + STARTING = "starting" + RUNNING = "running" + CLOSING = "closing" + CLOSED = "closed" + FAILED = "failed" + + +class EmbeddedReMeBackend: + """Own a ReMe Application and all of its async resources on one event loop.""" + + def __init__( + self, + workspace_dir: str, + *, + reme_config: str = "default", + start_timeout: float = 600.0, + ) -> None: + self.workspace_dir = workspace_dir + self.reme_config = reme_config + self.start_timeout = start_timeout + self.label = f"embedded:{workspace_dir}" + self._state = _State.NEW + self._state_lock = threading.RLock() + self._operation_lock = threading.Lock() + self._loop_ready = threading.Event() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._app: Any = None + self._app_close_future: concurrent.futures.Future[Any] | None = None + self._failure: BaseException | None = None + + @property + def state(self) -> str: + """Expose lifecycle state for diagnostics and focused tests.""" + with self._state_lock: + return self._state.value + + def start(self, *, deadline: float | None = None) -> None: + """Start the loop thread and construct the Application on that loop.""" + start_deadline = time.monotonic() + self.start_timeout + if deadline is not None: + start_deadline = min(start_deadline, deadline) + with self._state_lock: + if self._state is _State.RUNNING: + return + if self._state is not _State.NEW: + detail = f": {self._failure}" if self._failure else "" + raise ReMeBackendError( + f"Embedded ReMe cannot start from state {self._state.value}{detail}", + ) + self._state = _State.STARTING + self._thread = threading.Thread( + target=self._run_loop, + daemon=True, + name="reme-embedded-loop", + ) + self._thread.start() + + if not self._loop_ready.wait(timeout=max(0.0, start_deadline - time.monotonic())): + error = TimeoutError("Timed out while starting the embedded ReMe event loop") + self._fail(error) + self._close_after_failed_start(start_deadline) + raise ReMeBackendError( + "Timed out while starting the embedded ReMe event loop", + ) from error + + try: + self._submit( + self._start_application(), + timeout=max(0.0, start_deadline - time.monotonic()), + allow_starting=True, + ) + except ReMeBackendError as exc: + self._fail(exc) + self._close_after_failed_start(start_deadline) + raise + with self._state_lock: + self._state = _State.RUNNING + + def _close_after_failed_start(self, deadline: float) -> None: + """Begin cleanup without extending the startup caller's time budget.""" + try: + self.close(timeout=max(0.0, deadline - time.monotonic())) + except ReMeBackendError: + # close() still requests loop shutdown before reporting a timeout. + pass + + def _run_loop(self) -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + with self._state_lock: + self._loop = loop + should_run = self._state in {_State.STARTING, _State.RUNNING} + self._loop_ready.set() + try: + if should_run: + loop.run_forever() + finally: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete( + asyncio.gather(*pending, return_exceptions=True), + ) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.close() + + async def _start_application(self) -> None: + try: + from reme import Application + from reme.config import resolve_app_config + except ImportError as exc: + raise ReMeBackendError( + 'Embedded ReMe mode requires the SDK. Install it with: pip install "reme-ai[core]"', + ) from exc + + app_config = resolve_app_config( + config=self.reme_config, + log_config=False, + workspace_dir=self.workspace_dir, + enable_logo=False, + log_to_console=False, + log_to_file=False, + ) + app = Application(**app_config) + self._app = app + try: + await app.start() + except BaseException: + await app.close() + self._app = None + raise + + async def _close_application(self, app: Any) -> None: + """Close one Application and release it only after cleanup completes.""" + try: + await app.close() + finally: + with self._state_lock: + if self._app is app: + self._app = None + + def _stop_loop_after_app_close(self, future: concurrent.futures.Future[Any]) -> None: + """Finish deferred shutdown after Application cleanup leaves the foreground budget.""" + try: + future.result() + except BaseException as exc: # pragma: no cover - retained for diagnostics + with self._state_lock: + self._failure = self._failure or exc + with self._state_lock: + loop = self._loop + if loop is not None and loop.is_running(): + loop.call_soon_threadsafe(loop.stop) + + def _fail(self, error: BaseException) -> None: + with self._state_lock: + self._failure = self._failure or error + if self._state not in {_State.CLOSING, _State.CLOSED}: + self._state = _State.FAILED + + def _submit( + self, + coroutine: Coroutine[Any, Any, Any], + *, + timeout: float, + allow_starting: bool = False, + ) -> Any: + with self._state_lock: + allowed = {_State.RUNNING} + if allow_starting: + allowed.add(_State.STARTING) + if self._state not in allowed or self._loop is None or not self._loop.is_running(): + coroutine.close() + raise ReMeBackendError( + f"Embedded ReMe is not running (state: {self._state.value})", + ) + loop = self._loop + future = asyncio.run_coroutine_threadsafe(coroutine, loop) + try: + return future.result(timeout=timeout) + except concurrent.futures.TimeoutError as exc: + future.cancel() + raise ReMeBackendError( + f"Embedded ReMe operation timed out after {timeout:.1f}s", + ) from exc + except ReMeBackendError: + raise + except BaseException as exc: + raise ReMeBackendError(str(exc) or type(exc).__name__) from exc + + @staticmethod + def _response_dict(response: Any) -> dict[str, Any]: + if hasattr(response, "model_dump"): + result = response.model_dump() + elif isinstance(response, dict): + result = dict(response) + else: + raise ReMeBackendError("ReMe returned an unsupported response object") + if result.get("success") is not True: + raise ReMeBackendError( + str(result.get("answer") or "ReMe action did not report success"), + ) + return result + + async def _run_job(self, name: str, **kwargs: Any) -> dict[str, Any]: + if self._app is None: + raise ReMeBackendError("Embedded ReMe Application is unavailable") + response = await self._app.run_job(name, **kwargs) + return self._response_dict(response) + + def health(self, *, timeout: float) -> dict[str, Any]: + """Run ReMe's health-check job on the owned loop.""" + with self._operation_lock: + return require_healthy( + self._submit(self._run_job("health_check"), timeout=timeout), + ) + + def search(self, query: str, *, limit: int, timeout: float) -> dict[str, Any]: + """Run ReMe's search job on the owned loop.""" + with self._operation_lock: + return self._submit( + self._run_job("search", query=query, limit=limit), + timeout=timeout, + ) + + def auto_memory( + self, + session_id: str, + messages: list[dict[str, Any]], + *, + timeout: float, + ) -> dict[str, Any]: + """Run ReMe's automatic-memory job on the owned loop.""" + with self._operation_lock: + return self._submit( + self._run_job("auto_memory", session_id=session_id, messages=messages), + timeout=timeout, + ) + + def close(self, *, timeout: float) -> None: + """Close the Application, stop its loop, and join its thread.""" + deadline = time.monotonic() + max(0.0, timeout) + # A context manager cannot express the bounded wait required by shutdown. + acquired = self._operation_lock.acquire( # pylint: disable=consider-using-with + timeout=max(0.0, deadline - time.monotonic()), + ) + if not acquired: + raise ReMeBackendError( + "Timed out waiting for an embedded ReMe operation to finish", + ) + try: + self._close_locked(deadline) + finally: + self._operation_lock.release() + + def _close_locked(self, deadline: float) -> None: + """Close while holding the operation lock so jobs cannot overlap shutdown.""" + close_error: BaseException | None = None + defer_loop_stop = False + with self._state_lock: + if self._state is _State.CLOSED: + return + if self._state is _State.NEW: + self._state = _State.CLOSED + return + self._state = _State.CLOSING + loop = self._loop + thread = self._thread + app = self._app + + if loop is not None and loop.is_running() and app is not None: + remaining = max(0.0, deadline - time.monotonic()) + with self._state_lock: + future = self._app_close_future + if future is None: + future = asyncio.run_coroutine_threadsafe( + self._close_application(app), + loop, + ) + self._app_close_future = future + try: + future.result(timeout=remaining) + except concurrent.futures.TimeoutError: + defer_loop_stop = True + close_error = TimeoutError( + "Timed out while closing the embedded ReMe Application", + ) + except BaseException as exc: + close_error = exc + if defer_loop_stop: + future.add_done_callback(self._stop_loop_after_app_close) + if not defer_loop_stop and loop is not None and loop.is_running(): + loop.call_soon_threadsafe(loop.stop) + if thread is not None and thread is not threading.current_thread(): + thread.join(timeout=max(0.0, deadline - time.monotonic())) + + with self._state_lock: + self._state = _State.CLOSED if thread is None or not thread.is_alive() else _State.FAILED + if self._state is _State.FAILED and self._failure is None: + self._failure = TimeoutError( + "Timed out while stopping the embedded ReMe event loop", + ) + failure = self._failure if self._state is _State.FAILED else close_error + if failure is not None: + raise ReMeBackendError(str(failure)) from failure diff --git a/integrations/hermes_agent/figures/hermes-embedded-recall.png b/integrations/hermes_agent/figures/hermes-embedded-recall.png new file mode 100644 index 00000000..070656f0 Binary files /dev/null and b/integrations/hermes_agent/figures/hermes-embedded-recall.png differ diff --git a/integrations/hermes_agent/figures/hermes-http-recall.png b/integrations/hermes_agent/figures/hermes-http-recall.png new file mode 100644 index 00000000..69b29ef1 Binary files /dev/null and b/integrations/hermes_agent/figures/hermes-http-recall.png differ diff --git a/integrations/hermes_agent/figures/hermes-http-sessions.png b/integrations/hermes_agent/figures/hermes-http-sessions.png new file mode 100644 index 00000000..ce79c592 Binary files /dev/null and b/integrations/hermes_agent/figures/hermes-http-sessions.png differ diff --git a/integrations/hermes_agent/figures/hermes-provider-settings.png b/integrations/hermes_agent/figures/hermes-provider-settings.png new file mode 100644 index 00000000..8c0ab24f Binary files /dev/null and b/integrations/hermes_agent/figures/hermes-provider-settings.png differ diff --git a/integrations/hermes_agent/figures/hermes-reme-daily-note.png b/integrations/hermes_agent/figures/hermes-reme-daily-note.png new file mode 100644 index 00000000..67736409 Binary files /dev/null and b/integrations/hermes_agent/figures/hermes-reme-daily-note.png differ diff --git a/integrations/hermes_agent/http_backend.py b/integrations/hermes_agent/http_backend.py new file mode 100644 index 00000000..9e0b68d6 --- /dev/null +++ b/integrations/hermes_agent/http_backend.py @@ -0,0 +1,58 @@ +"""HTTP implementation of the ReMe backend contract.""" + +from __future__ import annotations + +from typing import Any + +from .backend import ReMeBackendError, require_healthy +from .client import ReMeHttpClient, ReMeServiceError + + +class HttpReMeBackend: + """Call an independently managed ReMe action service.""" + + def __init__(self, endpoint: str, *, request_timeout: float) -> None: + self._client = ReMeHttpClient(endpoint, timeout=request_timeout) + self.label = self._client.endpoint + + def start(self, *, deadline: float | None = None) -> None: + """HTTP service lifecycle is managed outside Hermes.""" + del deadline + + def _call( + self, + action: str, + payload: dict[str, Any] | None, + *, + timeout: float, + ) -> dict[str, Any]: + try: + return self._client.call(action, payload, timeout=timeout) + except (ReMeServiceError, ValueError) as exc: + raise ReMeBackendError(str(exc)) from exc + + def health(self, *, timeout: float) -> dict[str, Any]: + """Call the action service health check.""" + return require_healthy(self._call("health_check", None, timeout=timeout)) + + def search(self, query: str, *, limit: int, timeout: float) -> dict[str, Any]: + """Search through the HTTP action service.""" + return self._call("search", {"query": query, "limit": limit}, timeout=timeout) + + def auto_memory( + self, + session_id: str, + messages: list[dict[str, Any]], + *, + timeout: float, + ) -> dict[str, Any]: + """Submit a completed turn through the HTTP action service.""" + return self._call( + "auto_memory", + {"session_id": session_id, "messages": messages}, + timeout=timeout, + ) + + def close(self, *, timeout: float) -> None: + """Release no resources because urllib calls are request-scoped.""" + del timeout diff --git a/integrations/hermes_agent/plugin.yaml b/integrations/hermes_agent/plugin.yaml index 74e68ec2..5ac57b75 100644 --- a/integrations/hermes_agent/plugin.yaml +++ b/integrations/hermes_agent/plugin.yaml @@ -1,4 +1,11 @@ name: reme -version: 0.1.0 -description: "ReMe file-native long-term memory for Hermes Agent." +version: 0.2.0 +description: "ReMe local-first, file-native long-term memory for Hermes Agent (HTTP or embedded)." +kind: exclusive +manifest_version: 2 +requires_hermes: ">=0.21" +author: agentscope-ai +homepage: https://github.com/agentscope-ai/ReMe +license: Apache-2.0 +tags: [memory-provider, local-first, long-term-memory] pip_dependencies: [] diff --git a/tests/unit/test_hermes_agent_integration.py b/tests/unit/test_hermes_agent_integration.py new file mode 100644 index 00000000..6c0d37a5 --- /dev/null +++ b/tests/unit/test_hermes_agent_integration.py @@ -0,0 +1,569 @@ +"""Focused tests for the external Hermes memory-provider plugin.""" + +# pylint: disable=missing-class-docstring,missing-function-docstring +# pylint: disable=protected-access,wrong-import-position,unused-import + +from __future__ import annotations + +import sys +import types +import asyncio +import importlib +import threading +import time + +from dataclasses import dataclass +from pathlib import Path + +import pytest + +_PLUGIN_PARENT = Path(__file__).resolve().parents[2] / "integrations" +if str(_PLUGIN_PARENT) not in sys.path: + sys.path.insert(0, str(_PLUGIN_PARENT)) + + +try: + import agent.memory_provider # type: ignore[import-not-found] # noqa: F401 +except ImportError: + agent_module = types.ModuleType("agent") + memory_provider_module = types.ModuleType("agent.memory_provider") + + class MemoryProvider: + """Minimal Hermes contract used when hermes-agent is not installed.""" + + @dataclass(frozen=True) + class RecallStatus: + provider_label: str + count: int + glyph: str = "🧠" + + memory_provider_module.MemoryProvider = MemoryProvider + memory_provider_module.RecallStatus = RecallStatus + agent_module.memory_provider = memory_provider_module + sys.modules["agent"] = agent_module + sys.modules["agent.memory_provider"] = memory_provider_module + +PLUGIN_MODULE = importlib.import_module("hermes_agent") +BACKEND_MODULE = importlib.import_module("hermes_agent.backend") +CONFIG_MODULE = importlib.import_module("hermes_agent.config") +EMBEDDED_MODULE = importlib.import_module("hermes_agent.embedded_backend") +CLIENT_MODULE = importlib.import_module("hermes_agent.client") + +ReMeMemoryProvider = PLUGIN_MODULE.ReMeMemoryProvider +ReMeBackendError = BACKEND_MODULE.ReMeBackendError +ReMeConfig = CONFIG_MODULE.ReMeConfig +ReMeConfigError = CONFIG_MODULE.ReMeConfigError +config_path = CONFIG_MODULE.config_path +load_config = CONFIG_MODULE.load_config +parse_config = CONFIG_MODULE.parse_config +save_config = CONFIG_MODULE.save_config +EmbeddedReMeBackend = EMBEDDED_MODULE.EmbeddedReMeBackend +ReMeHttpClient = CLIENT_MODULE.ReMeHttpClient +scoped_session_id = PLUGIN_MODULE._scoped_session_id + + +def test_config_defaults_to_http(tmp_path): + config = load_config(tmp_path) + + assert config.mode == "http" + assert config.endpoint == "http://127.0.0.1:2333" + + +def test_provider_schema_exposes_mode_specific_and_advanced_fields(): + fields = {field["key"]: field for field in ReMeMemoryProvider().get_config_schema()} + + assert set(fields) == { + "mode", + "endpoint", + "workspace_dir", + "reme_config", + "recall_limit", + "recall_timeout", + "request_timeout", + "health_timeout", + "health_retry_seconds", + "shutdown_timeout", + } + assert fields["endpoint"]["when"] == {"mode": "http"} + assert fields["workspace_dir"]["when"] == {"mode": "embedded"} + + +def test_current_config_precedes_legacy(tmp_path): + (tmp_path / "reme.json").write_text( + '{"endpoint": "http://legacy:1"}', + encoding="utf-8", + ) + current = config_path(tmp_path) + current.parent.mkdir() + current.write_text('{"endpoint": "http://current:2"}', encoding="utf-8") + + assert load_config(tmp_path).endpoint == "http://current:2" + + +def test_sparse_dashboard_config_inherits_legacy_values(tmp_path): + (tmp_path / "reme.json").write_text( + '{"endpoint": "http://legacy:2444", "recall_limit": 3}', + encoding="utf-8", + ) + current = config_path(tmp_path) + current.parent.mkdir() + current.write_text('{"recall_limit": 7}', encoding="utf-8") + + config = load_config(tmp_path) + + assert config.endpoint == "http://legacy:2444" + assert config.recall_limit == 7 + + +def test_embedded_config_normalizes_workspace(tmp_path): + config = parse_config( + {"mode": " EMBEDDED ", "workspace_dir": str(tmp_path / "workspace")}, + hermes_home=tmp_path, + ) + + assert config.mode == "embedded" + assert config.workspace_dir == str((tmp_path / "workspace").absolute()) + + +def test_embedded_config_requires_workspace(tmp_path): + with pytest.raises(ReMeConfigError, match="workspace_dir"): + parse_config({"mode": "embedded"}, hermes_home=tmp_path) + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")]) +def test_config_rejects_non_finite_timeouts(tmp_path, value): + with pytest.raises(ReMeConfigError, match="positive number"): + parse_config({"request_timeout": value}, hermes_home=tmp_path) + + +@pytest.mark.parametrize( + "endpoint", + [ + "http://user:secret@127.0.0.1:2333", + "http://127.0.0.1:2333?token=secret", + "http://127.0.0.1:2333#fragment", + "http://127.0.0.1:invalid", + ], +) +def test_config_and_client_reject_unsafe_endpoint_shapes(tmp_path, endpoint): + with pytest.raises(ReMeConfigError, match="absolute http"): + parse_config({"endpoint": endpoint}, hermes_home=tmp_path) + with pytest.raises(ValueError, match="absolute http"): + ReMeHttpClient(endpoint, timeout=1) + + +def test_save_config_uses_dashboard_layout_and_private_permissions(tmp_path): + saved = save_config({"mode": "http", "recall_limit": 7}, tmp_path) + path = config_path(tmp_path) + + assert saved.recall_limit == 7 + assert path.is_file() + assert path.stat().st_mode & 0o777 == 0o600 + + +class _FakeResponse: + def __init__(self, answer="", metadata=None, success=True): + self.answer = answer + self.metadata = metadata or {} + self.success = success + + def model_dump(self): + return { + "answer": self.answer, + "metadata": self.metadata, + "success": self.success, + } + + +class _FakeApplication: + instances = [] + + def __init__(self, **config): + self.config = config + self.started = False + self.closed = False + self.calls = [] + self.__class__.instances.append(self) + + async def start(self): + self.started = True + + async def close(self): + self.closed = True + + async def run_job(self, name, **kwargs): + self.calls.append((name, kwargs)) + if name == "health_check": + return _FakeResponse(metadata={"health": {"healthy": True}}) + return _FakeResponse(answer=name, metadata={"counts": {"returned": 1}}) + + +def test_embedded_backend_owns_application_lifecycle(monkeypatch, tmp_path): + import reme + import reme.config + + _FakeApplication.instances.clear() + monkeypatch.setattr(reme, "Application", _FakeApplication) + monkeypatch.setattr( + reme.config, + "resolve_app_config", + lambda **kwargs: { + "workspace_dir": kwargs["workspace_dir"], + "marker": kwargs["config"], + }, + ) + backend = EmbeddedReMeBackend(str(tmp_path), start_timeout=2) + + backend.start() + assert backend.health(timeout=2)["success"] is True + assert backend.search("needle", limit=3, timeout=2)["answer"] == "search" + backend.auto_memory("session", [{"role": "user", "content": "hello"}], timeout=2) + backend.close(timeout=2) + + app = _FakeApplication.instances[0] + assert app.started is True + assert app.closed is True + assert app.config == { + "workspace_dir": str(tmp_path), + "marker": "default", + } + assert [name for name, _ in app.calls] == ["health_check", "search", "auto_memory"] + assert backend.state == "closed" + + +class _FakeBackend: + label = "fake" + + def __init__(self): + self.writes = [] + self.closed = False + + def start(self): + return None + + def health(self, *, timeout): + del timeout + return {"success": True, "metadata": {"health": {"healthy": True}}} + + def search(self, query, *, limit, timeout): + del query, limit, timeout + return { + "success": True, + "answer": " remembered ", + "metadata": {"counts": {"returned": 2}}, + } + + def auto_memory(self, session_id, messages, *, timeout): + del timeout + self.writes.append((session_id, messages)) + return {"success": True} + + def close(self, *, timeout): + del timeout + self.closed = True + + +def test_provider_selects_backend_recalls_and_writes(monkeypatch, tmp_path): + backend = _FakeBackend() + monkeypatch.setattr(PLUGIN_MODULE, "load_config", lambda home=None: ReMeConfig()) + monkeypatch.setattr(PLUGIN_MODULE, "_backend_for", lambda config: backend) + provider = ReMeMemoryProvider() + provider.initialize("session/one", hermes_home=tmp_path, agent_identity="work") + + assert provider.prefetch("project decision") == "remembered" + assert provider.recall_status().count == 2 + provider.sync_turn("hello", "hi") + provider.shutdown() + + assert len(backend.writes) == 1 + assert backend.writes[0][0].startswith("hermes-work-session-one-") + assert backend.closed is True + + +def test_session_scope_distinguishes_profiles_and_ambiguous_names(): + assert scoped_session_id("profile/a", "session") != scoped_session_id( + "profile-a", + "session", + ) + assert scoped_session_id("profile", "session/a") != scoped_session_id( + "profile", + "session-a", + ) + + +def test_reinitialize_closes_previous_backend(monkeypatch, tmp_path): + backends = [_FakeBackend(), _FakeBackend()] + monkeypatch.setattr(PLUGIN_MODULE, "load_config", lambda home=None: ReMeConfig()) + monkeypatch.setattr(PLUGIN_MODULE, "_backend_for", lambda config: backends.pop(0)) + provider = ReMeMemoryProvider() + + provider.initialize("first", hermes_home=tmp_path) + first = provider._backend + provider.initialize("second", hermes_home=tmp_path) + + assert first is not None and first.closed is True + assert provider._backend is not first + provider.shutdown() + + +def test_provider_failure_does_not_escape_model_path(monkeypatch, tmp_path): + backend = _FakeBackend() + backend.search = lambda *args, **kwargs: (_ for _ in ()).throw( + ReMeBackendError("offline"), + ) + monkeypatch.setattr(PLUGIN_MODULE, "load_config", lambda home=None: ReMeConfig()) + monkeypatch.setattr(PLUGIN_MODULE, "_backend_for", lambda config: backend) + provider = ReMeMemoryProvider() + provider.initialize("session", hermes_home=tmp_path) + + assert provider.prefetch("query") == "" + assert provider.recall_status() is None + provider.shutdown() + + +def test_provider_reports_embedded_workspace_for_hermes_backup(monkeypatch, tmp_path): + workspace = tmp_path / "workspace" + monkeypatch.setattr( + PLUGIN_MODULE, + "load_config", + lambda home=None: ReMeConfig(mode="embedded", workspace_dir=str(workspace)), + ) + + assert ReMeMemoryProvider().backup_paths() == [str(workspace)] + + +def test_backend_creation_is_serialized(monkeypatch): + provider = ReMeMemoryProvider() + provider._config = ReMeConfig() + created = [] + + def factory(config): + del config + time.sleep(0.05) + backend = _FakeBackend() + created.append(backend) + return backend + + monkeypatch.setattr(PLUGIN_MODULE, "_backend_for", factory) + gate = threading.Barrier(3) + results = [] + + def ensure(): + gate.wait() + results.append(provider._ensure_backend()) + + threads = [threading.Thread(target=ensure) for _ in range(2)] + for thread in threads: + thread.start() + gate.wait() + for thread in threads: + thread.join(timeout=2) + + assert results == [True, True] + assert len(created) == 1 + provider.shutdown() + + +def test_shutdown_defers_close_until_inflight_recall_finishes(): + entered = threading.Event() + release = threading.Event() + + class BlockingBackend(_FakeBackend): + def search(self, query, *, limit, timeout): + del query, limit, timeout + entered.set() + release.wait(timeout=2) + return {"success": True, "answer": "remembered", "metadata": {}} + + backend = BlockingBackend() + provider = ReMeMemoryProvider() + provider._config = ReMeConfig() + provider._backend = backend + provider._backend_available = True + provider._shutdown_timeout = 0.05 + recall = threading.Thread(target=provider.prefetch, args=("query",)) + recall.start() + assert entered.wait(timeout=1) + + provider.shutdown() + assert backend.closed is False + release.set() + recall.join(timeout=2) + + assert recall.is_alive() is False + assert backend.closed is True + + +def test_shutdown_drains_all_accepted_writes_before_closing_backend(): + entered = threading.Event() + release = threading.Event() + + class BlockingBackend(_FakeBackend): + def auto_memory(self, session_id, messages, *, timeout): + del timeout + self.writes.append((session_id, messages)) + if len(self.writes) == 1: + entered.set() + release.wait(timeout=2) + + backend = BlockingBackend() + provider = ReMeMemoryProvider() + provider._config = ReMeConfig() + provider._backend = backend + provider._backend_available = True + provider._shutdown_timeout = 1 + for index in range(3): + provider.sync_turn(f"user {index}", f"assistant {index}", session_id=f"session-{index}") + assert entered.wait(timeout=1) + + shutdown = threading.Thread(target=provider.shutdown) + shutdown.start() + deadline = time.monotonic() + 1 + while not provider._shutdown_started and time.monotonic() < deadline: + time.sleep(0.001) + assert provider._shutdown_started is True + release.set() + shutdown.join(timeout=2) + + assert shutdown.is_alive() is False + assert len(backend.writes) == 3 + assert backend.closed is True + + +def test_recall_timeout_includes_waiting_for_background_write(): + write_entered = threading.Event() + write_release = threading.Event() + search_entered = threading.Event() + + class BlockingBackend(_FakeBackend): + def auto_memory(self, session_id, messages, *, timeout): + del session_id, messages, timeout + write_entered.set() + write_release.wait(timeout=2) + + def search(self, query, *, limit, timeout): + del query, limit, timeout + search_entered.set() + return {"success": True, "answer": "remembered", "metadata": {}} + + backend = BlockingBackend() + provider = ReMeMemoryProvider() + provider._config = ReMeConfig() + provider._backend = backend + provider._backend_available = True + provider._recall_timeout = 0.05 + provider.sync_turn("user", "assistant", session_id="session") + assert write_entered.wait(timeout=1) + + started = time.monotonic() + assert provider.prefetch("query") == "" + elapsed = time.monotonic() - started + + assert elapsed < 0.2 + assert search_entered.is_set() is False + write_release.set() + provider.shutdown() + + +def test_recall_timeout_includes_embedded_startup_failure_cleanup(monkeypatch, tmp_path): + import reme + import reme.config + + class SlowApplication: + instances = [] + + def __init__(self, **config): + del config + self.closed = threading.Event() + self.thread_pool = object() + self.__class__.instances.append(self) + + async def start(self): + time.sleep(0.25) + + async def close(self): + await asyncio.sleep(0.4) + self.thread_pool = None + self.closed.set() + + monkeypatch.setattr(reme, "Application", SlowApplication) + monkeypatch.setattr(reme.config, "resolve_app_config", lambda **kwargs: kwargs) + backend = EmbeddedReMeBackend(str(tmp_path), start_timeout=1) + monkeypatch.setattr(PLUGIN_MODULE, "_backend_for", lambda config: backend) + provider = ReMeMemoryProvider() + provider._config = ReMeConfig( + mode="embedded", + workspace_dir=str(tmp_path), + request_timeout=1, + recall_timeout=0.15, + shutdown_timeout=0.6, + ) + provider._recall_timeout = 0.15 + provider._shutdown_timeout = 0.6 + + started = time.monotonic() + assert provider.prefetch("query") == "" + elapsed = time.monotonic() - started + + assert elapsed < 0.25 + assert backend._thread is not None + backend._thread.join(timeout=1) + assert backend._thread.is_alive() is False + app = SlowApplication.instances[0] + assert app.closed.is_set() + assert app.thread_pool is None + + +def test_embedded_start_timeout_closes_real_application_resources(monkeypatch, tmp_path): + import reme + import reme.config + + from reme.application import Application + from reme.components import BaseComponent + + component_closed = threading.Event() + + class SlowComponent(BaseComponent): + component_type = "slow_test" + + async def _start(self): + time.sleep(0.25) + + async def _close(self): + component_closed.set() + + app = Application( + workspace_dir=str(tmp_path), + service={"backend": "cli"}, + thread_pool_max_workers=1, + enable_logo=False, + log_to_console=False, + log_to_file=False, + ) + component = SlowComponent(name="slow", app_context=app.context) + app.context.components["slow_test"] = {"slow": component} + monkeypatch.setattr(reme, "Application", lambda **config: app) + monkeypatch.setattr(reme.config, "resolve_app_config", lambda **kwargs: kwargs) + backend = EmbeddedReMeBackend(str(tmp_path), start_timeout=1) + + started = time.monotonic() + with pytest.raises(ReMeBackendError, match="timed out"): + backend.start(deadline=started + 0.15) + elapsed = time.monotonic() - started + + assert elapsed < 0.25 + assert backend._thread is not None + backend._thread.join(timeout=1) + assert backend._thread.is_alive() is False + assert component_closed.is_set() + assert component.is_started is False + assert app.context.thread_pool is None + + +def test_shutdown_discard_keeps_sentinel_for_inflight_writer(): + provider = ReMeMemoryProvider() + provider._write_queue.put({"session_id": "queued"}) + provider._write_queue.put(None) + + assert provider._discard_queued_writes() == 1 + assert provider._write_queue.get_nowait() is None