mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-11 22:51:10 +00:00
feat(integrations): add dual-mode Hermes memory provider (#533)
* feat(integrations): add dual-mode Hermes memory provider * style(integrations): apply repository formatting * fix(integrations): address Hermes provider review * fix(integrations): bound embedded recall startup cleanup * fix(integrations): finish embedded application cleanup * docs(integrations): expand Hermes verification guide
This commit is contained in:
jinliyl
GitHub
parent
7e25d4679b
commit
1be61b1e4c
25 changed files with 2349 additions and 299 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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. |
|
||||
|
||||
<p align="center"><b>Integration demos</b></p>
|
||||
|
|
|
|||
|
|
@ -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 搜索、读取和写入记忆；自动捕获需要显式接入宿主生命周期。 |
|
||||
|
||||
<p align="center"><b>集成演示</b></p>
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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。
|
||||
|
||||
## 生产接入建议
|
||||
|
||||
|
|
|
|||
|
|
@ -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)。
|
||||
|
|
|
|||
|
|
@ -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)。
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
@ -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.
|
||||
|
||||
|
||||
|
||||
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
|
||||
`<memory-context>` 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
|
||||
|
||||
|
||||
|
||||
### Two independent Hermes sessions
|
||||
|
||||
|
||||
|
||||
### HTTP mode recall
|
||||
|
||||
|
||||
|
||||
### File-native durable result
|
||||
|
||||
|
||||
|
||||
### Embedded mode recall
|
||||
|
||||
|
||||
|
||||
## 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.
|
||||
|
|
|
|||
263
integrations/hermes_agent/README_ZH.md
Normal file
263
integrations/hermes_agent/README_ZH.md
Normal file
|
|
@ -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_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 会话
|
||||
|
||||
|
||||
|
||||
### HTTP 模式召回
|
||||
|
||||
|
||||
|
||||
### 文件原生的持久化结果
|
||||
|
||||
|
||||
|
||||
### Embedded 模式召回
|
||||
|
||||
|
||||
|
||||
## 常见问题
|
||||
|
||||
### `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 会话。
|
||||
|
|
@ -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", "<unknown>"),
|
||||
"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:
|
||||
|
|
|
|||
45
integrations/hermes_agent/backend.py
Normal file
45
integrations/hermes_agent/backend.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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]:
|
||||
|
|
|
|||
208
integrations/hermes_agent/config.py
Normal file
208
integrations/hermes_agent/config.py
Normal file
|
|
@ -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
|
||||
109
integrations/hermes_agent/config_schema.py
Normal file
109
integrations/hermes_agent/config_schema.py
Normal file
|
|
@ -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",
|
||||
),
|
||||
),
|
||||
)
|
||||
323
integrations/hermes_agent/embedded_backend.py
Normal file
323
integrations/hermes_agent/embedded_backend.py
Normal file
|
|
@ -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
|
||||
BIN
integrations/hermes_agent/figures/hermes-embedded-recall.png
Normal file
BIN
integrations/hermes_agent/figures/hermes-embedded-recall.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 533 KiB |
BIN
integrations/hermes_agent/figures/hermes-http-recall.png
Normal file
BIN
integrations/hermes_agent/figures/hermes-http-recall.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 352 KiB |
BIN
integrations/hermes_agent/figures/hermes-http-sessions.png
Normal file
BIN
integrations/hermes_agent/figures/hermes-http-sessions.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 302 KiB |
BIN
integrations/hermes_agent/figures/hermes-provider-settings.png
Normal file
BIN
integrations/hermes_agent/figures/hermes-provider-settings.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 357 KiB |
BIN
integrations/hermes_agent/figures/hermes-reme-daily-note.png
Normal file
BIN
integrations/hermes_agent/figures/hermes-reme-daily-note.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 204 KiB |
58
integrations/hermes_agent/http_backend.py
Normal file
58
integrations/hermes_agent/http_backend.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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: []
|
||||
|
|
|
|||
569
tests/unit/test_hermes_agent_integration.py
Normal file
569
tests/unit/test_hermes_agent_integration.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue