Merge 2.0.24 into main

apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py

@@ -65,13 +65,11 @@
 if str(_PLUGIN_DIR) not in sys.path:
     sys.path.insert(0, str(_PLUGIN_DIR))
 
-from bridge_client import BridgeError, MemosBridgeClient, MemosHttpClient  # noqa: E402
+from bridge_client import BridgeError, MemosBridgeClient  # noqa: E402
 from daemon_manager import (  # noqa: E402
     ensure_bridge_running,
     ensure_viewer_daemon,
     kill_zombie_bridges,
-    probe_viewer_status,
-    startup_lock_active,
 )
 
 
@@ -279,7 +277,7 @@ class MemTensorProvider(MemoryProvider):
     """
 
     def __init__(self) -> None:
-        self._bridge: MemosBridgeClient | MemosHttpClient | None = None
+        self._bridge: MemosBridgeClient | None = None
         self._reconnect_lock = threading.Lock()
         self._session_id: str = ""
         self._episode_id: str = ""
@@ -329,23 +327,6 @@ def is_available(self) -> bool:  # type: ignore[override]
 
     # ─── Lifecycle ────────────────────────────────────────────────────────
 
-    def _connect_http_bridge(self, session_id: str, *, timeout: float = 60.0) -> bool:
-        """Try to connect via HTTP bridge. Sets self._bridge on success."""
-        http_bridge: MemosHttpClient | None = None
-        try:
-            http_bridge = MemosHttpClient()
-            http_bridge.register_host_handler("host.llm.complete", self._handle_host_llm_complete)
-            self._bridge = http_bridge
-            self._open_session(session_id, timeout=timeout)
-            return True
-        except Exception as err:
-            logger.warning("MemOS: HTTP bridge failed, falling back to stdio — %s", err)
-            if http_bridge is not None:
-                with contextlib.suppress(Exception):
-                    http_bridge.close()
-            self._bridge = None
-            return False
-
     def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[override]
         """Called once at agent startup.
 
@@ -414,32 +395,13 @@ def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[ov
         except Exception:
             pass
 
-        # If the daemon is already running on the viewer port, connect
-        # to it over HTTP instead of spawning a new stdio bridge. This
-        # eliminates zombie bridge accumulation.
-        viewer_status = probe_viewer_status()
-        if viewer_status == "running_memos":
-            if self._connect_http_bridge(session_id):
-                logger.info(
-                    "MemOS: bridge ready (HTTP) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
-            else:
-                viewer_status = "free"  # force stdio fallback below
-        elif viewer_status == "free":
-            # Re-probe after a short wait only when another process may be
-            # mid-startup (startup lock is held). On a cold first-launch the
-            # lock doesn't exist, so we skip the delay entirely.
-            if startup_lock_active():
-                time.sleep(1.0)
-                viewer_status = probe_viewer_status()
-            if viewer_status == "running_memos" and self._connect_http_bridge(session_id):
-                logger.info(
-                    "MemOS: bridge ready (HTTP, late probe) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
+        # NOTE: An HTTP bridge path used to live here that connected to a
+        # running viewer daemon over HTTP instead of spawning a stdio
+        # subprocess. It depended on ``MemosHttpClient`` which was never
+        # committed — the class was referenced by name only. Issue #2096
+        # reverts the half-merged HTTP feature; the stdio path below is
+        # the sole connection mechanism until the HTTP client lands as a
+        # complete change.
 
         if self._bridge is None:
             try:
@@ -1958,13 +1920,9 @@ def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> N
                 logger.info("MemOS: old bridge closed (pid=%s)", old_pid)
 
             ensure_bridge_running()
-            # Try HTTP first if daemon is running
-            viewer_status = probe_viewer_status()
-            if viewer_status == "running_memos" and self._connect_http_bridge(
-                session_id, timeout=timeout
-            ):
-                logger.info("MemOS: reconnected via HTTP")
-                return
+            # NOTE: HTTP bridge reconnect path was removed alongside issue
+            # #2096. See ``initialize`` for the rationale. Reconnect always
+            # spawns a fresh stdio bridge.
 
             try:
                 ensure_viewer_daemon()

apps/memos-local-plugin/core/llm/client.ts

@@ -260,6 +260,21 @@ export function createLlmClientWithProvider(
     return [{ role: "system", content: systemInsert }, ...messages];
   }
 
+  function ensureJsonWordInUserMessage(messages: LlmMessage[]): LlmMessage[] {
+    const lastUserIdx = messages.map((m) => m.role).lastIndexOf("user");
+    if (lastUserIdx < 0) return [...messages, { role: "user", content: "Return valid json only." }];
+
+    const msg = messages[lastUserIdx];
+    if (/\bjson\b/i.test(msg.content)) return messages;
+
+    const out = messages.slice();
+    out[lastUserIdx] = {
+      ...msg,
+      content: `${msg.content}\n\nReturn valid json only.`,
+    };
+    return out;
+  }
+
   function buildCallInput(opts: LlmCallOptions | undefined, jsonMode: boolean): ProviderCallInput {
     return {
       temperature: opts?.temperature ?? config.temperature,
@@ -463,7 +478,7 @@ export function createLlmClientWithProvider(
   ): Promise<LlmCompletion> {
     const messages = normalizeMessages(input);
     const msgsWithJsonHint = opts?.jsonMode
-      ? inject(messages, buildJsonSystemHint())
+      ? ensureJsonWordInUserMessage(inject(messages, buildJsonSystemHint()))
       : messages;
     const call = buildCallInput(opts, opts?.jsonMode === true);
     const { completion } = await callWithFallback(msgsWithJsonHint, call, opts, opts?.op ?? "complete");
@@ -476,7 +491,7 @@ export function createLlmClientWithProvider(
   ): Promise<LlmJsonCompletion<T>> {
     const messages = normalizeMessages(input);
     const systemHint = buildJsonSystemHint(opts.schemaHint);
-    const msgs = inject(messages, systemHint);
+    const msgs = ensureJsonWordInUserMessage(inject(messages, systemHint));
     const call = buildCallInput(opts, true);
     const op = opts.op ?? "complete.json";
     const maxMalformedRetries = Math.max(0, opts.malformedRetries ?? 1);

apps/memos-local-plugin/tests/python/test_hermes_provider_pipeline.py

@@ -68,6 +68,51 @@ def request(self, method: str, params: dict | None = None, **_kwargs: object) ->
 
 
 class HermesProviderPipelineTests(unittest.TestCase):
+    def test_module_imports_cleanly(self) -> None:
+        """Regression guard for #2096: asserts that ``MemosHttpClient`` is
+        NOT present in ``memos_provider``, since the class was referenced
+        before it was ever committed (see issue #2096).
+
+        Note: the import itself is already validated at collection time —
+        the ``import memos_provider`` at the top of this file will raise
+        ``ImportError`` if a dangling reference is reintroduced, causing
+        the entire test file to fail to load. This test body only adds:
+
+        * the explicit negative guard on ``MemosHttpClient`` below (unique
+          to this test), which covers both the ``memos_provider``
+          re-export surface *and* ``bridge_client`` itself so a partial
+          re-add of the class only in ``bridge_client`` (with no
+          matching re-export) still fails the guard, and
+        * positive checks on ``MemTensorProvider`` (the class the Hermes
+          host actually instantiates) and on ``bridge_client``'s real
+          contract (``MemosBridgeClient`` / ``BridgeError``), rather than
+          on their incidental re-exports through ``memos_provider`` — the
+          latter only appear on the package namespace because
+          ``__init__.py`` uses a bare ``from bridge_client import ...``,
+          which is an implementation detail we don't want the test to
+          lock in.
+        """
+        import importlib
+
+        # MemTensorProvider is the class hermes-agent host instantiates.
+        self.assertTrue(hasattr(memos_provider, "MemTensorProvider"))
+
+        # Assert the actual contract on bridge_client directly rather
+        # than on its re-exports through memos_provider.
+        bc = importlib.import_module("bridge_client")
+        self.assertTrue(hasattr(bc, "MemosBridgeClient"))
+        self.assertTrue(hasattr(bc, "BridgeError"))
+
+        # ``MemosHttpClient`` was referenced by name in a half-merged HTTP
+        # bridge feature (see #2096). It must not reappear until the class
+        # itself is committed in ``bridge_client``. Guard both the
+        # ``memos_provider`` re-export (which is what the original
+        # ImportError travelled through) and ``bridge_client`` itself —
+        # otherwise a partial re-add of the class in ``bridge_client``
+        # without a matching re-export would slip past this test.
+        self.assertFalse(hasattr(memos_provider, "MemosHttpClient"))
+        self.assertFalse(hasattr(bc, "MemosHttpClient"))
+
     def test_lifecycle_persists_turn_and_closes_real_episode(self) -> None:
         bridge = FakeBridge()
         with (

apps/memos-local-plugin/tests/unit/llm/client.test.ts

@@ -96,12 +96,14 @@ describe("llm/client", () => {
     expect(fake.lastMessages).toEqual([{ role: "user", content: "hi there" }]);
   });
 
-  it("injects a json system hint when jsonMode=true", async () => {
+  it("injects json hints into system and user messages when jsonMode=true", async () => {
     const fake = new FakeProvider("openai_compatible", () => ({ text: '{"ok":1}', durationMs: 1 }));
     const client = createLlmClientWithProvider(cfg(), fake);
     await client.complete("do it", { jsonMode: true });
     expect(fake.lastMessages?.[0]?.role).toBe("system");
     expect(fake.lastMessages?.[0]?.content).toMatch(/single valid JSON value/i);
+    expect(fake.lastMessages?.at(-1)?.role).toBe("user");
+    expect(fake.lastMessages?.at(-1)?.content).toMatch(/valid json only/i);
     expect(fake.lastInput?.jsonMode).toBe(true);
   });
 
@@ -270,7 +272,9 @@ describe("llm/client", () => {
     expect(fake.lastMessages?.[0]?.role).toBe("system");
     expect(fake.lastMessages?.[0]?.content).toMatch(/You are strict\./);
     expect(fake.lastMessages?.[0]?.content).toMatch(/single valid JSON value/);
-    expect(fake.lastMessages?.[1]).toEqual({ role: "user", content: "go" });
+    expect(fake.lastMessages?.[1]?.role).toBe("user");
+    expect(fake.lastMessages?.[1]?.content).toMatch(/^go/);
+    expect(fake.lastMessages?.[1]?.content).toMatch(/valid json only/i);
   });
 
   it("rejects empty messages array", async () => {

docs/cn/open_source/open_source_api/scheduler/get_status.md

@@ -64,34 +64,48 @@ desc: 监控 MemOS 异步任务的生命周期，提供包括任务进度、队
 
 ## 4. 快速上手示例
 
-使用 SDK 轮询任务状态直至完成：
+这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。以下示例轮询任务状态直至完成：
 
 ```python
-from memos.api.client import MemOSClient
 import time
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头）
+base_url = "http://localhost:8000"
 
 # 1. 系统级概览：查看整个 MemOS 系统的运行健康度
-global_res = client.get_all_scheduler_status()
-if global_res:
-    print(f"系统运行概况: {global_res.data['scheduler_summary']}")
+resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10)
+resp.raise_for_status()
+global_res = resp.json()
+print(f"系统运行概况: {global_res['data']['scheduler_summary']}")
 
 # 2. 队列指标监控：检查特定用户的任务积压情况
-queue_res = client.get_task_queue_status(user_id="dev_user_01")
-if queue_res:
-    print(f"待处理任务数: {queue_res.data['remaining_tasks_count']}")
-    print(f"已下发未完成任务数: {queue_res.data['pending_tasks_count']}")
+resp = requests.get(
+    f"{base_url}/product/scheduler/task_queue_status",
+    params={"user_id": "dev_user_01"},
+    timeout=10,
+)
+resp.raise_for_status()
+queue_res = resp.json()
+print(f"排队中任务数: {queue_res['data']['remaining_tasks_count']}")
+print(f"已下发未确认任务数: {queue_res['data']['pending_tasks_count']}")
 
 # 3. 任务进度追踪：轮询特定任务直至结束
 task_id = "task_888999"
+active_states = {"waiting", "pending", "in_progress"}
 while True:
-    res = client.get_task_status(user_id="dev_user_01", task_id=task_id)
-    if res and res.code == 200:
-        current_status = res.data[0]['status'] # data 为状态列表
-        print(f"任务 {task_id} 当前状态: {current_status}")
-
-        if current_status in ['completed', 'failed', 'cancelled']:
-            break
+    resp = requests.get(
+        f"{base_url}/product/scheduler/status",
+        params={"user_id": "dev_user_01", "task_id": task_id},
+        timeout=10,
+    )
+    resp.raise_for_status()
+    items = resp.json().get("data", [])  # data 为状态列表：[{"task_id": ..., "status": ...}]
+    statuses = {item["status"] for item in items}
+    print(f"任务 {task_id} 当前状态: {statuses or '空'}")
+
+    if not statuses or statuses.isdisjoint(active_states):
+        break
     time.sleep(2)
 ```

docs/cn/open_source/open_source_api/scheduler/ wait.md → docs/cn/open_source/open_source_api/scheduler/wait.md

@@ -42,36 +42,47 @@ desc: 提供阻塞等待与流式进度观测能力，确保在执行后续操
 
 ## 4. 快速上手示例
 
-使用开源版 SDK 进行阻塞式等待：
+这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。注意：`user_name`、`timeout_seconds`、`poll_interval` 均为查询参数（Query），而非请求体（Body）。以下示例进行阻塞式等待：
 
 ```python
-from memos.api.client import MemOSClient
+import json
 
-client = MemOSClient(api_key="...", base_url="...")
+import requests
+
+# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头）
+base_url = "http://localhost:8000"
 user_name = "dev_user_01"
 
 # --- 场景 A：同步阻塞等待 (常用于 Python 自动化脚本) ---
 print(f"正在等待用户 {user_name} 的任务队列清空...")
-res = client.wait_until_idle(
-    user_name=user_name,
-    timeout_seconds=300,
-    poll_interval=2
+resp = requests.post(
+    f"{base_url}/product/scheduler/wait",
+    params={"user_name": user_name, "timeout_seconds": 300, "poll_interval": 2},
+    timeout=310,  # HTTP 超时应大于 timeout_seconds
 )
-if res and res.code == 200:
+resp.raise_for_status()
+result = resp.json()  # {"message": "idle" | "timeout", "data": {...}}
+if result["message"] == "idle":
     print("✅ 任务已全部完成。")
+else:
+    print(f"⚠️ 等待超时，仍有 {result['data']['running_tasks']} 个任务在执行。")
 
 # --- 场景 B：流式进度观测 (常用于前端进度条渲染) ---
 print("开始监听任务实时进度流...")
-# 注意：SSE 接口在 SDK 中通常返回一个生成器 (Generator)
-progress_stream = client.stream_scheduler_progress(
-    user_name=user_name,
-    timeout_seconds=300
-)
-
-for event in progress_stream:
-    # 实时打印剩余任务数
-    print(f"当前排队任务数: {event['remaining_tasks_count']}")
-    if event['status'] == 'idle':
-        print("🎉 调度器已空闲")
-        break
+with requests.get(
+    f"{base_url}/product/scheduler/wait/stream",
+    params={"user_name": user_name, "timeout_seconds": 300},
+    stream=True,
+    timeout=310,
+) as resp:
+    resp.raise_for_status()
+    for line in resp.iter_lines(decode_unicode=True):
+        if not line or not line.startswith("data:"):
+            continue
+        event = json.loads(line.removeprefix("data:").strip())
+        # 实时打印仍在执行的任务数
+        print(f"当前活跃任务数: {event['active_tasks']}，状态: {event['status']}")
+        if event["status"] in ("idle", "timeout"):
+            print("🎉 调度器已空闲" if event["status"] == "idle" else "⚠️ 监听超时")
+            break
 ```