Fix #34: memos add 写入成功但 list/get 查不到：user_id 作用域在写入侧被静默丢弃

MemOS CLI.md

@@ -53,6 +53,8 @@ Commands:
 - `memos get "$MEMORY_ID" --format json --detail detail`
 - `memos delete "$MEMORY_ID" --format json`
 
+> 说明：`memos list` 是 `memos get` 的别名——两者都在当前 `user_id` 作用域下列出记忆，参数与输出格式完全一致。
+
 
 ```
 

src/memos_cli/backend/memory_api.py

@@ -55,13 +55,20 @@ def ping(self, timeout: float = 5.0) -> dict[str, Any]:
         raise APIError("Unable to reach MemOS API with the configured base URL")
 
     def add_memory(self, messages: list[dict[str, Any]], **kwargs: Any) -> dict[str, Any]:
-        """Add messages."""
+        """Add messages scoped to a user."""
+        # Reads (get_memories, chat) require user_id; writes must be scoped to the
+        # same resolver so a memory is never stored without a user scope. Otherwise a
+        # record written unscoped is silently invisible to the scoped read path.
+        user_id = kwargs.get("user_id")
+        if not user_id:
+            raise APIError("Add memory requires user_id")
+
         message_payload: dict[str, Any] = {
             "messages": messages,
+            "user_id": user_id,
         }
 
         common_fields = [
-            "user_id",
             "conversation_id",
             "agent_id",
             "app_id",
@@ -138,14 +145,17 @@ def search_memories(self, query: str, **kwargs: Any) -> dict[str, Any]:
             "memory_limit_number": limit,
             "include_preference": kwargs.get("include_preference", True),
         }
-        # Only include non-None values
-        if kwargs.get("user_id"):
+        # Skip both None and empty strings so the read contract matches the write
+        # contract in add_memory/get_memories (both raise on falsy user_id). Sending
+        # user_id="" to /search/memory would filter by an empty scope on the server,
+        # which never matches any memory written with a real user_id.
+        if kwargs.get("user_id") not in (None, ""):
             payload["user_id"] = kwargs["user_id"]
-        if kwargs.get("conversation_id"):
+        if kwargs.get("conversation_id") is not None:
             payload["conversation_id"] = kwargs["conversation_id"]
-        if kwargs.get("agent_id"):
+        if kwargs.get("agent_id") is not None:
             payload["agent_id"] = kwargs["agent_id"]
-        if kwargs.get("app_id"):
+        if kwargs.get("app_id") is not None:
             payload["app_id"] = kwargs["app_id"]
         if kwargs.get("filter") is not None:
             payload["filter"] = kwargs["filter"]

src/memos_cli/commands/memory.py

@@ -201,6 +201,30 @@ def get(
         include_tool_memory=include_tool_memory,
         output_format=output_format,
         detail=detail,
+        command_name="get",
+    )
+
+
+def list(
+    user_id_arg: str | None = typer.Argument(None, help="User ID"),
+    user_id: str | None = typer.Option(None, "--user-id", help="User ID"),
+    page: int | None = typer.Option(None, "--page", min=1, help="Page number"),
+    size: int | None = typer.Option(None, "--size", min=1, help="Page size"),
+    include_preference: str | None = typer.Option(None, "--include-preference", help="Include preference memory: true or false"),
+    include_tool_memory: str | None = typer.Option(None, "--include-tool-memory", help="Include tool memory: true or false"),
+    output_format: str | None = typer.Option(None, "--format", help=FORMAT_HELP),
+    detail: str | None = typer.Option(None, "--detail", help=DETAIL_HELP),
+):
+    """List memories for a user (alias for `get`)."""
+    cmd_get(
+        user_id=user_id_arg or user_id,
+        page=page,
+        size=size,
+        include_preference=include_preference,
+        include_tool_memory=include_tool_memory,
+        output_format=output_format,
+        detail=detail,
+        command_name="list",
     )
 
 

src/memos_cli/commands/memory_cmd.py

@@ -618,8 +618,9 @@ def cmd_get(
     include_tool_memory: str | None,
     output_format: str,
     detail: str,
+    command_name: str = "get",
 ) -> None:
-    """Execute get."""
+    """Execute get (also available as `list`, an alias for the same operation)."""
     start_time = time.time()
     final_output = resolve_output_format(output_format)
     final_detail = validate_detail(detail)
@@ -641,7 +642,7 @@ def cmd_get(
     if final_output == "agent":
         format_agent_envelope(
             console,
-            command="get",
+            command=command_name,
             data=memories,
             duration_ms=duration_ms,
             count=len(memories),

src/memos_cli/main.py

@@ -14,7 +14,18 @@
 from memos_cli.completion import register_completion_compat
 from memos_cli.commands.init import init_cmd, uninstall_cmd
 from memos_cli.commands.config_cmd import config_app
-from memos_cli.commands.memory import add, extract, feedback, rerank, search, chat, get, delete, origin
+from memos_cli.commands.memory import (
+    add,
+    extract,
+    feedback,
+    rerank,
+    search,
+    chat,
+    get,
+    list as list_cmd,
+    delete,
+    origin,
+)
 from memos_cli.commands.message import message, status
 from memos_cli.commands.kb import kb_app
 from memos_cli.state import set_runtime_options
@@ -34,6 +45,7 @@ class CommandFirstTyperGroup(TyperGroup):
         "add",
         "search",
         "get",
+        "list",
         "origin",
         "delete",
         "extract",
@@ -136,6 +148,7 @@ def _fire_telemetry(command_name: str, extra: dict | None = None):
 app.command(rich_help_panel="Memory Operations")(feedback)
 app.command(rich_help_panel="Memory Operations")(search)
 app.command(rich_help_panel="Memory Operations")(get)
+app.command(name="list", rich_help_panel="Memory Operations")(list_cmd)
 app.command(rich_help_panel="Memory Operations")(origin)
 app.command(rich_help_panel="Memory Operations")(delete)
 

tests/test_memory_scope.py

@@ -0,0 +1,165 @@
+"""Regression tests for user_id scope symmetry and the `list` alias.
+
+Issue #34: `memos add` could store a memory without a user scope (the resolved
+user_id was silently dropped when its value was None), while `memos get`/`search`
+always filtered by user_id — so a just-added memory could not be read back.
+"""
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+import typer
+
+from memos_cli.backend.memory_api import MemoryAPI
+from memos_cli.backend.transport import APIError
+from memos_cli.commands import memory, memory_cmd
+from memos_cli.config import MemOSConfig, PlatformConfig
+
+
+class RecordingTransport:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str, dict]] = []
+
+    def request_json(self, method: str, path: str, **kwargs):
+        self.calls.append((method, path, kwargs))
+        return {"code": 0, "data": {}}
+
+
+class AddMemoryScopeTests(unittest.TestCase):
+    def test_add_memory_always_includes_user_id_in_body(self) -> None:
+        transport = RecordingTransport()
+        api = MemoryAPI(transport)
+
+        api.add_memory(
+            [{"role": "user", "content": "loves green tea"}],
+            user_id="user_7",
+            conversation_id="conversation_1",
+        )
+
+        body = transport.calls[0][2]["json_body"]
+        self.assertEqual(body["user_id"], "user_7")
+        self.assertEqual(body["conversation_id"], "conversation_1")
+
+    def test_add_memory_rejects_missing_user_id(self) -> None:
+        api = MemoryAPI(RecordingTransport())
+
+        with self.assertRaises(APIError) as raised:
+            api.add_memory(
+                [{"role": "user", "content": "loves green tea"}],
+                user_id=None,
+            )
+
+        self.assertIn("Add memory requires user_id", str(raised.exception))
+
+    def test_search_memories_drops_empty_string_scope(self) -> None:
+        """search must treat "" the same way add/get do: reject as an unscoped read.
+
+        add_memory and get_memories both raise on falsy user_id. Sending user_id=""
+        to /search/memory would filter the server-side query by an empty scope and
+        never return memories written under a real user, silently reproducing the
+        invisible-scope bug #34 was meant to fix.
+        """
+        transport = RecordingTransport()
+        api = MemoryAPI(transport)
+
+        api.search_memories("greens", user_id="")
+
+        body = transport.calls[0][2]["json_body"]
+        self.assertNotIn("user_id", body)
+
+
+class ResolveScopeTests(unittest.TestCase):
+    def test_resolve_scope_uses_config_default_when_flag_absent(self) -> None:
+        config = MemOSConfig(
+            platform=PlatformConfig(api_key="test-key"),
+        )
+        config.defaults.user_id = "default_user"
+
+        scope = memory_cmd.resolve_scope(
+            config=config,
+            user_id=None,
+            agent_id=None,
+            app_id=None,
+            run_id=None,
+        )
+
+        self.assertEqual(scope["user_id"], "default_user")
+
+
+class ListAliasTests(unittest.TestCase):
+    def test_list_command_forwards_to_cmd_get_as_list(self) -> None:
+        with patch.object(memory, "cmd_get") as cmd_get:
+            memory.list(
+                None,
+                user_id="user_7",
+                page=None,
+                size=None,
+                include_preference=None,
+                include_tool_memory=None,
+                output_format="table",
+                detail="simple",
+            )
+
+        self.assertEqual(cmd_get.call_count, 1)
+        self.assertEqual(cmd_get.call_args.kwargs["user_id"], "user_7")
+        self.assertEqual(cmd_get.call_args.kwargs["command_name"], "list")
+
+    def test_get_command_forwards_to_cmd_get_as_get(self) -> None:
+        with patch.object(memory, "cmd_get") as cmd_get:
+            memory.get(
+                None,
+                user_id="user_7",
+                page=None,
+                size=None,
+                include_preference=None,
+                include_tool_memory=None,
+                output_format="table",
+                detail="simple",
+            )
+
+        self.assertEqual(cmd_get.call_count, 1)
+        self.assertEqual(cmd_get.call_args.kwargs["command_name"], "get")
+
+    def test_main_registers_list_command(self) -> None:
+        from memos_cli import main
+
+        click_group = typer.main.get_command(main.app)
+        self.assertIn("list", click_group.commands)
+
+    def test_cmd_get_outputs_list_command_name_in_agent_mode(self) -> None:
+        config = MemOSConfig(
+            platform=PlatformConfig(
+                api_key="test-key",
+                base_url="https://example.test/api",
+            )
+        )
+        config.defaults.user_id = "user_1"
+
+        class Backend:
+            def get_memories(self, **kwargs):
+                return {"data": {"memory_detail_list": []}}
+
+        captured: dict = {}
+
+        def fake_envelope(console, **kwargs):
+            captured.update(kwargs)
+
+        with patch.object(memory_cmd, "_load_backend", return_value=(config, Backend())):
+            with patch.object(memory_cmd, "format_agent_envelope", side_effect=fake_envelope):
+                memory_cmd.cmd_get(
+                    user_id=None,
+                    page=None,
+                    size=None,
+                    include_preference=None,
+                    include_tool_memory=None,
+                    output_format="agent",
+                    detail="simple",
+                    command_name="list",
+                )
+
+        self.assertEqual(captured["command"], "list")
+
+
+if __name__ == "__main__":
+    unittest.main()