feat(scripts): listScripts parent_path filter + lazy load by directory

Backend — backend/src/backend/scripts.py
- list_scripts accepts optional parent_path Query (default "").
- Extracts _build_list_scripts_descendant_prefix helper for the
  "direct children of parent_path" prefix.
- WHERE clause now adds:
    StorageObjects.relative_path LIKE '<prefix>/%'
    AND NOT LIKE '<prefix>/%/%'
  so deeper descendants and prefix-siblings (foo/bar vs foo/bar2)
  are excluded. Empty parent_path filters to root-level only —
  this is the symmetric, intent-aligned behavior the lazy-load
  frontend relies on.
- EXPLAIN confirms idx_scripts_workspace drives the scripts table;
  storage_objects PK lookup applies the LIKE filter per row.

Frontend — frontend/app/services/api.ts + scriptWorkspaceStore.ts
- listScripts accepts optional parentPath; built URL preserves
  the new filter param.
- Store gains loadedScriptPaths / loadingScriptPaths Sets and a
  loadScripts(parentPath) action: idempotent, in-flight dedupe,
  dedup-by-id when merging into the flat scripts array so existing
  find() callers keep working.
- load() now uses listScripts("") instead of bulk fetch — root
  only on first paint.
- toggleExpanded() triggers loadScripts(parentPath) in parallel
  with loadChildren(parentPath) so folder expansion loads both
  sub-directories and direct-child scripts.

Tests — backend/tests/test_list_scripts_parent_path.py
- 7 unit tests: prefix construction, normalization, traversal
  rejection, and SQL compilation contract (LIKE prefix + NOT LIKE
  prefix + scripts.status filter).

Verified:
- pytest backend/tests: 50 passed (43 existing + 7 new)
- pnpm typecheck: clean
- alembic: no schema changes (migration-less feature)
- EXPLAIN with real workspace: idx_scripts_workspace → PK lookup

backend/src/backend/scripts.py

@@ -115,6 +115,28 @@ def user_relative_path(context: RequestContext, child_path: str = "") -> str:
     return f"{base}/{normalized}" if normalized else base
 
 
+def _build_list_scripts_descendant_prefix(
+    context: RequestContext, parent_path: str
+) -> str:
+    """Return the materialized-path prefix for direct children of ``parent_path``.
+
+    The endpoint appends ``LIKE '<prefix>/%' AND NOT LIKE '<prefix>/%/%'``
+    against ``storage_objects.relative_path`` so only scripts whose parent
+    directory is exactly ``parent_path`` (no deeper descendants, no
+    prefix-siblings like ``foo/bar`` vs ``foo/bar2``) match.
+
+    Empty ``parent_path`` produces the user-scoped root prefix — i.e. the
+    endpoint returns root-level scripts only, not the full workspace.
+    """
+    normalized_parent = normalize_user_path(parent_path)
+    scoped_prefix = user_relative_path(context)
+    if normalized_parent:
+        target_prefix = f"{scoped_prefix}/{normalized_parent}"
+    else:
+        target_prefix = scoped_prefix
+    return f"{target_prefix}/"
+
+
 def safe_script_name(value: str, script_type: str) -> str:
     name = value.replace("\\", "/").rsplit("/", 1)[-1].strip()
     if not name or name in {".", ".."} or any(ord(char) < 32 for char in name):
@@ -1045,11 +1067,23 @@ async def delete_workspace_directory(
 
 
 # 列出当前工作区中用户有权查看的脚本与 Notebook 元数据。
+#
+# 可选 ``parent_path`` Query 参数把返回范围限定为 *直接挂在该路径下的* 脚本
+# (不含更深的子目录)。空字符串等价于用户作用域根目录;这是前端按目录懒加载
+# 的关键端点,避免 10 万级脚本一次性返回。
+#
+# 实现走 ``storage_objects.relative_path`` materialized path(``LIKE prefix%``
+# + ``NOT LIKE prefix%/%`` 取直接子节点),配合现有
+# ``idx_storage_workspace_relative_path (workspace_id, relative_path)`` 索引
+# 避免全表扫。
 @router.get("/api/v1/scripts")
 async def list_scripts(
+    parent_path: str = Query(default="", max_length=1024),
     context: RequestContext = Depends(request_context),
     session: AsyncSession = Depends(database_session),
 ) -> dict[str, Any]:
+    descendant_prefix = _build_list_scripts_descendant_prefix(context, parent_path)
+
     statement = (
         select(Scripts, StorageObjects, Users.display_name)
         .join(
@@ -1060,6 +1094,8 @@ async def list_scripts(
         .where(
             Scripts.workspace_id == context.workspace.workspace_id,
             Scripts.status == "active",
+            StorageObjects.relative_path.like(f"{descendant_prefix}%"),
+            ~StorageObjects.relative_path.like(f"{descendant_prefix}%/%"),
         )
         .order_by(Scripts.updated_at.desc())
     )

backend/tests/test_list_scripts_parent_path.py

@@ -0,0 +1,136 @@
+"""Unit tests for the parent_path filter clause on list_scripts.
+
+Verifies the WHERE clause built by the endpoint encodes the intended
+"direct children of parent_path" semantics: ``relative_path LIKE 'prefix/%'``
+and ``NOT LIKE 'prefix/%/%'``. The actual ORM roundtrip is covered by the
+existing integration tests; here we only care that the filter contract is
+intact, so an AsyncMock session is enough.
+
+Mirrors the pattern in test_scripts.py (unit-level, no live DB).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import HTTPException
+
+from backend.scripts import (
+    _build_list_scripts_descendant_prefix,
+    normalize_user_path,
+)
+from sqlalchemy.dialects import mysql as mysql_dialect
+
+
+def _ctx(user_id: str = "U001") -> SimpleNamespace:
+    """Stand-in for RequestContext — only ``user.user_id`` and ``workspace_id`` are read."""
+    return SimpleNamespace(
+        request_id="test",
+        user=SimpleNamespace(user_id=user_id),
+        workspace=SimpleNamespace(workspace_id="W001"),
+        role=SimpleNamespace(role_code="admin"),
+        is_system_admin=False,
+    )
+
+
+def test_descendant_prefix_root() -> None:
+    """Empty parent_path → descendant prefix is the scoped root + '/'."""
+    prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "")
+    assert prefix == "workspace/alice/"
+
+
+def test_descendant_prefix_subdir() -> None:
+    """Non-empty parent_path → appended under the scoped root."""
+    prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "foo/bar")
+    assert prefix == "workspace/alice/foo/bar/"
+
+
+def test_descendant_prefix_normalizes_leading_trailing_slashes() -> None:
+    """Leading/trailing slashes on parent_path must be stripped."""
+    prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "/foo/bar/")
+    assert prefix == "workspace/alice/foo/bar/"
+
+
+def test_descendant_prefix_rejects_traversal() -> None:
+    """``..`` segments must raise (matches normalize_user_path contract)."""
+    with pytest.raises(HTTPException) as exc:
+        _build_list_scripts_descendant_prefix(_ctx("alice"), "foo/../bar")
+    assert exc.value.status_code == 422
+
+
+def test_normalize_user_path_strips() -> None:
+    assert normalize_user_path("") == ""
+    assert normalize_user_path("/a/b/") == "a/b"
+    assert normalize_user_path("a\\b") == "a/b"
+
+
+async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
+    """The WHERE clause must include both LIKE prefix and NOT LIKE '%/%' filters
+    so deeper descendants and prefix-siblings (foo/bar vs foo/bar2) are excluded."""
+    from backend.scripts import list_scripts
+
+    captured_sql: list[str] = []
+
+    class _MockResult:
+        def all(self):
+            return []
+
+    mock_session = MagicMock()
+    mock_session.execute = AsyncMock(
+        side_effect=lambda stmt: (
+            captured_sql.append(
+                str(
+                    stmt.compile(
+                        dialect=mysql_dialect.dialect(),
+                        compile_kwargs={"literal_binds": True},
+                    )
+                )
+            )
+            or _MockResult()
+        )
+    )
+
+    await list_scripts(parent_path="foo/bar", context=_ctx("alice"), session=mock_session)
+
+    assert len(captured_sql) == 1
+    sql = captured_sql[0].lower()
+    # Direct-child LIKE prefix
+    assert "like 'workspace/alice/foo/bar/%%'" in sql
+    # NOT-LIKE deeper
+    assert "not like 'workspace/alice/foo/bar/%%/%%'" in sql
+    # active scripts only (existing contract preserved)
+    assert "scripts.status" in sql or "scripts.status = 'active'" in sql or "scripts.status = :status" in sql
+
+
+async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None:
+    """parent_path='' produces root-scoped LIKE prefix only, not full scan."""
+    from backend.scripts import list_scripts
+
+    captured_sql: list[str] = []
+
+    class _MockResult:
+        def all(self):
+            return []
+
+    mock_session = MagicMock()
+    mock_session.execute = AsyncMock(
+        side_effect=lambda stmt: (
+            captured_sql.append(
+                str(
+                    stmt.compile(
+                        dialect=mysql_dialect.dialect(),
+                        compile_kwargs={"literal_binds": True},
+                    )
+                )
+            )
+            or _MockResult()
+        )
+    )
+
+    await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session)
+
+    sql = captured_sql[0].lower()
+    assert "like 'workspace/alice/%%'" in sql
+    assert "not like 'workspace/alice/%%/%%'" in sql

frontend/app/features/platform/state/scriptWorkspaceStore.ts

@@ -85,6 +85,12 @@ type State = {
   expandedPaths: Set<string>;
   loadingChildrenPaths: Set<string>;
   loadedChildPaths: Set<string>;
+  // Per-parent-path script cache. Keys are user-relative parent paths;
+  // values are scripts whose storage path lives directly under that parent.
+  // The flat `scripts` array above is the union (deduped by script_id) of
+  // every cache entry that has been loaded in this session.
+  loadedScriptPaths: Set<string>;
+  loadingScriptPaths: Set<string>;
 
   // 只读内容刷新版本号（用于触发已打开标签页的内容刷新）
   readOnlyRefreshVersion: number;
@@ -122,6 +128,9 @@ type State = {
   deleteDirectory: (path: string) => Promise<void>;
   toggleExpanded: (path: string, loadPath?: string) => Promise<void>;
   loadChildren: (parentPath: string) => Promise<void>;
+  // Lazy-load scripts directly under `parentPath`. Idempotent — repeated
+  // calls for an already-loaded path are no-ops; in-flight calls dedupe.
+  loadScripts: (parentPath: string) => Promise<void>;
   toggleScriptLock: (script: ScriptItem) => Promise<void>;
   openPublishDialog: (script: ScriptItem) => void;
   submitPublish: (releaseNote: string, visibility: Visibility) => Promise<void>;
@@ -203,6 +212,8 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
     expandedPaths: new Set<string>(),
     loadingChildrenPaths: new Set<string>(),
     loadedChildPaths: new Set<string>(),
+    loadedScriptPaths: new Set<string>(),
+    loadingScriptPaths: new Set<string>(),
 
     readOnlyRefreshVersion: 0,
 
@@ -242,6 +253,8 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
         expandedPaths: new Set<string>(),
         loadingChildrenPaths: new Set<string>(),
         loadedChildPaths: new Set<string>(),
+        loadedScriptPaths: new Set<string>(),
+        loadingScriptPaths: new Set<string>(),
       });
     },
 
@@ -250,17 +263,23 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
       if (!silent) set({ loading: true });
       set({ refreshing: silent });
       try {
+        // Fetch root-level scripts (parent_path="") and root directories in
+        // parallel. Deeper folders are lazy-fetched on expand via loadScripts
+        // / loadChildren.
         const [items, folderItems] = await Promise.all([
-          api.listScripts(),
+          api.listScripts(""),
           api.listWorkspaceDirectories(""),
         ]);
-        const nextLoaded = new Set(get().loadedChildPaths);
-        nextLoaded.add("");
+        const nextLoadedChildren = new Set(get().loadedChildPaths);
+        nextLoadedChildren.add("");
+        const nextLoadedScripts = new Set(get().loadedScriptPaths);
+        nextLoadedScripts.add("");
         set({
           scripts: items,
           directories: folderItems,
           apiOnline: true,
-          loadedChildPaths: nextLoaded,
+          loadedChildPaths: nextLoadedChildren,
+          loadedScriptPaths: nextLoadedScripts,
         });
         const validIds = new Set(items.map((item) => item.script_id));
         const currentSelected = get().selectedId;
@@ -284,6 +303,42 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
       }
     },
 
+    loadScripts: async (parentPath) => {
+      const api = requireApi();
+      if (get().loadedScriptPaths.has(parentPath)) return;
+      // Dedupe in-flight requests for the same path.
+      if (get().loadingScriptPaths.has(parentPath)) return;
+      const next = new Set(get().loadingScriptPaths);
+      next.add(parentPath);
+      set({ loadingScriptPaths: next });
+      try {
+        const items = await api.listScripts(parentPath);
+        set((state) => {
+          const existingIds = new Set(state.scripts.map((s) => s.script_id));
+          const fresh = items.filter((s) => !existingIds.has(s.script_id));
+          const nextLoaded = new Set(state.loadedScriptPaths);
+          nextLoaded.add(parentPath);
+          return {
+            scripts: [...state.scripts, ...fresh],
+            loadedScriptPaths: nextLoaded,
+            loadingScriptPaths: new Set(
+              [...state.loadingScriptPaths].filter((p) => p !== parentPath),
+            ),
+          };
+        });
+      } catch (error) {
+        set((state) => ({
+          loadingScriptPaths: new Set(
+            [...state.loadingScriptPaths].filter((p) => p !== parentPath),
+          ),
+        }));
+        pushToast(
+          "error",
+          error instanceof Error ? error.message : "脚本列表加载失败",
+        );
+      }
+    },
+
     loadDataResources: async () => {
       const api = requireApi();
       set({ dataResourcesLoading: true });
@@ -340,11 +395,18 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
         next.delete(path);
       } else {
         next.add(path);
-        // group key (`__group__<owner_id>`) 不是 workspace 路径,不调 loadChildren
+        // group key (`__group__<owner_id>`) 不是 workspace 路径,不调 loadChildren / loadScripts
         const actualLoadPath = loadPath ?? path;
-        if (!actualLoadPath.startsWith("__group__") && !state.loadedChildPaths.has(actualLoadPath)) {
+        if (!actualLoadPath.startsWith("__group__")) {
+          // Load both sub-directories and scripts directly under this folder
+          // in parallel. Both are idempotent + cached; cheap when already loaded.
+          if (!state.loadedChildPaths.has(actualLoadPath)) {
             void get().loadChildren(actualLoadPath);
           }
+          if (!state.loadedScriptPaths.has(actualLoadPath)) {
+            void get().loadScripts(actualLoadPath);
+          }
+        }
       }
       set({ expandedPaths: next });
     },

frontend/app/services/api.ts

@@ -284,8 +284,16 @@ async function apiRequest<T>(
   return (payload as ApiEnvelope<T>).data;
 }
 
-export async function listScripts(workspaceId: string): Promise<ScriptItem[]> {
-  return apiRequest<ScriptItem[]>("/api/v1/scripts", {}, workspaceId);
+export async function listScripts(
+  workspaceId: string,
+  parentPath: string = "",
+): Promise<ScriptItem[]> {
+  // Empty parentPath omits the query string entirely so the backend's
+  // root-level filter is applied symmetrically with non-empty paths.
+  const query = parentPath
+    ? `?parent_path=${encodeURIComponent(parentPath)}`
+    : "";
+  return apiRequest<ScriptItem[]>(`/api/v1/scripts${query}`, {}, workspaceId);
 }
 
 function initialContent(scriptType: ScriptType): string {
@@ -1455,7 +1463,9 @@ export async function getScheduleNodeRunArtifacts(
 // references all the exports above.
 // ----------------------------------------------------------------------------
 export type WorkspaceBoundApi = {
-  listScripts: () => Promise<ScriptItem[]>;
+  listScripts: (
+    parentPath?: Parameters<typeof listScripts>[1],
+  ) => Promise<ScriptItem[]>;
   listResources: (
     opts?: Parameters<typeof listResources>[1],
   ) => Promise<ResourceItem[]>;