From c8fb895febb33ac5095fc6e45c6224a5867179ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?d=20=F0=9F=94=B9?= <258577966+voidborne-d@users.noreply.github.com> Date: Sat, 28 Mar 2026 19:06:42 +0000 Subject: [PATCH 01/15] fix: register_skill_dir returns existing SkillMeta for already-registered skills Fixes #29. When a skill is already registered, register_skill_dir() returned None, which caused fix_skill() to incorrectly report a failure. Now returns the existing SkillMeta instead of None when the skill_id is already present in the registry, making register_skill_dir() truly idempotent as its callers (fix_skill, _auto_register_skill_dirs) expect. --- openspace/skill_engine/registry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openspace/skill_engine/registry.py b/openspace/skill_engine/registry.py index bb35dd8..a217458 100644 --- a/openspace/skill_engine/registry.py +++ b/openspace/skill_engine/registry.py @@ -321,7 +321,7 @@ class SkillRegistry: meta = self._parse_skill(skill_dir.name, skill_dir, skill_file, content) if meta.skill_id in self._skills: logger.debug(f"register_skill_dir: {meta.skill_id} already exists") - return None + return self._skills[meta.skill_id] self._skills[meta.skill_id] = meta self._content_cache[meta.skill_id] = content logger.info(f"Hot-registered skill: {meta.skill_id}") From 4f61cb2fa1e0d34acbcb2d8d20c7f8c2c53e883a Mon Sep 17 00:00:00 2001 From: Marc von Renteln Date: Sun, 29 Mar 2026 11:16:10 +0200 Subject: [PATCH 02/15] fix: pin litellm to <1.82.7 to avoid PYSEC-2026-2 supply-chain attack Versions 1.82.7 and 1.82.8 of litellm were published on March 24, 2026 and contained malicious code that exfiltrated credentials (SSH keys, cloud credentials, .env files, API keys) to an attacker-controlled domain. Pin the dependency to >=1.70.0,<1.82.7 in both pyproject.toml and requirements.txt as a stopgap until litellm can be replaced with direct provider SDK calls. See: https://github.com/HKUDS/OpenSpace/issues/31 Ref: PYSEC-2026-2, BerriAI/litellm#24521 --- pyproject.toml | 2 +- requirements.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 571ef8a..06c199c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ authors = [ ] dependencies = [ - "litellm>=1.70.0", + "litellm>=1.70.0,<1.82.7", # pinned to avoid PYSEC-2026-2 supply-chain compromise (1.82.7/1.82.8 were malicious) "python-dotenv>=1.0.0", "openai>=1.0.0", "jsonschema>=4.25.0", diff --git a/requirements.txt b/requirements.txt index 3f8c7c4..6371308 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ # OpenSpace core dependencies -litellm>=1.70.0 +litellm>=1.70.0,<1.82.7 # pinned to avoid PYSEC-2026-2 supply-chain compromise (1.82.7/1.82.8 were malicious) python-dotenv>=1.0.0 openai>=1.0.0 jsonschema>=4.25.0 From aa16419e4696ab339343b771c5c59bd9d22ada79 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 14:58:01 +0800 Subject: [PATCH 03/15] docs: update register_skill_dir docstring to reflect idempotent return --- openspace/skill_engine/registry.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/openspace/skill_engine/registry.py b/openspace/skill_engine/registry.py index a217458..114f6d8 100644 --- a/openspace/skill_engine/registry.py +++ b/openspace/skill_engine/registry.py @@ -299,8 +299,8 @@ class SkillRegistry: skill_dir: Path to a directory containing ``SKILL.md``. Returns: - :class:`SkillMeta` if newly registered, ``None`` if already - present, the directory is invalid, or the skill fails safety checks. + :class:`SkillMeta` if newly registered or already present, + ``None`` if the directory is invalid or the skill fails safety checks. """ skill_file = skill_dir / "SKILL.md" if not skill_file.exists(): From f4451aa0ac540663d679a1d4a849e89333d78ce0 Mon Sep 17 00:00:00 2001 From: who96 <825265100@qq.com> Date: Thu, 26 Mar 2026 08:52:20 +0800 Subject: [PATCH 04/15] mcp: keep local skill search lightweight --- openspace/mcp_server.py | 65 +++++++++++++++++++++++++++++++++++- tests/test_issue3_startup.py | 49 +++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 1 deletion(-) create mode 100644 tests/test_issue3_startup.py diff --git a/openspace/mcp_server.py b/openspace/mcp_server.py index 7e3428f..d3b8e9e 100644 --- a/openspace/mcp_server.py +++ b/openspace/mcp_server.py @@ -119,6 +119,7 @@ mcp = FastMCP("OpenSpace", **_fastmcp_kwargs) _openspace_instance = None _openspace_lock = asyncio.Lock() _standalone_store = None +_local_skill_registry = None # Internal state: tracks bot skill directories already registered this session. _registered_skill_dirs: set = set() @@ -198,6 +199,64 @@ def _get_store(): return _standalone_store +def _get_local_skill_registry(): + """Build a lightweight SkillRegistry for local-only skill search. + + This avoids initializing the full OpenSpace engine when callers only + want to inspect local skills. It mirrors the skill directory discovery + order used by the full engine, but skips LLM / provider startup. + """ + global _local_skill_registry + if _local_skill_registry is not None: + return _local_skill_registry + + from openspace.config import get_config + from openspace.skill_engine import SkillRegistry + + skill_paths: List[Path] = [] + + host_dirs_raw = os.environ.get("OPENSPACE_HOST_SKILL_DIRS", "") + if host_dirs_raw: + for d in host_dirs_raw.split(","): + d = d.strip() + if not d: + continue + p = Path(d) + if p.exists(): + skill_paths.append(p) + else: + logger.warning("Host skill dir does not exist: %s", d) + + try: + skill_cfg = get_config().skills + except Exception as e: + logger.warning("Failed to load local skill config: %s", e) + skill_cfg = None + + if skill_cfg and skill_cfg.skill_dirs: + for d in skill_cfg.skill_dirs: + p = Path(d) + if p in skill_paths: + continue + if p.exists(): + skill_paths.append(p) + else: + logger.warning("Configured skill dir does not exist: %s", d) + + builtin_skills = Path(__file__).resolve().parent / "skills" + if builtin_skills.exists(): + skill_paths.append(builtin_skills) + + if not skill_paths: + logger.debug("No local skill directories found") + return None + + registry = SkillRegistry(skill_dirs=skill_paths) + registry.discover() + _local_skill_registry = registry + return registry + + def _get_cloud_client(): """Get a OpenSpaceClient instance (raises CloudError if not configured).""" from openspace.cloud.auth import get_openspace_auth @@ -597,7 +656,11 @@ async def search_skills( # Re-scan host skill directories so newly created skills are searchable. local_skills = None store = None - if source in ("all", "local"): + if source == "local": + registry = _get_local_skill_registry() + if registry: + local_skills = registry.list_skills() + elif source == "all": openspace = await _get_openspace() host_skill_dirs_raw = os.environ.get("OPENSPACE_HOST_SKILL_DIRS", "") diff --git a/tests/test_issue3_startup.py b/tests/test_issue3_startup.py new file mode 100644 index 0000000..386a412 --- /dev/null +++ b/tests/test_issue3_startup.py @@ -0,0 +1,49 @@ +import json + +import pytest + +from openspace import mcp_server + + +@pytest.mark.asyncio +async def test_local_search_skills_does_not_initialize_openspace(monkeypatch, tmp_path): + skill_root = tmp_path / "skills" + skill_dir = skill_root / "demo-local-skill" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text( + """--- +name: Demo Local Skill +description: Local regression test skill +--- + +Find me when querying demo local skill. +""", + encoding="utf-8", + ) + + async def forbidden_get_openspace(): + pytest.fail("_get_openspace should not run for source='local'") + + monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root)) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) + monkeypatch.setattr( + "openspace.cloud.embedding.generate_embedding", + lambda text, api_key=None: None, + ) + + response = await mcp_server.search_skills( + query="demo local skill", + source="local", + limit=5, + auto_import=False, + ) + + payload = json.loads(response) + assert payload["count"] >= 1 + assert any( + item["name"] == "Demo Local Skill" + for item in payload["results"] + ) From fb02862d4451921c7961b48bb5e4091296f53b5e Mon Sep 17 00:00:00 2001 From: who96 <825265100@qq.com> Date: Thu, 26 Mar 2026 09:18:17 +0800 Subject: [PATCH 05/15] Refresh local registry for local skill search --- openspace/mcp_server.py | 8 ++--- tests/test_issue3_startup.py | 69 ++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 6 deletions(-) diff --git a/openspace/mcp_server.py b/openspace/mcp_server.py index d3b8e9e..e81ef12 100644 --- a/openspace/mcp_server.py +++ b/openspace/mcp_server.py @@ -119,7 +119,6 @@ mcp = FastMCP("OpenSpace", **_fastmcp_kwargs) _openspace_instance = None _openspace_lock = asyncio.Lock() _standalone_store = None -_local_skill_registry = None # Internal state: tracks bot skill directories already registered this session. _registered_skill_dirs: set = set() @@ -205,11 +204,9 @@ def _get_local_skill_registry(): This avoids initializing the full OpenSpace engine when callers only want to inspect local skills. It mirrors the skill directory discovery order used by the full engine, but skips LLM / provider startup. + The registry is rebuilt per call so later local searches can see + newly added skills without requiring a process restart. """ - global _local_skill_registry - if _local_skill_registry is not None: - return _local_skill_registry - from openspace.config import get_config from openspace.skill_engine import SkillRegistry @@ -253,7 +250,6 @@ def _get_local_skill_registry(): registry = SkillRegistry(skill_dirs=skill_paths) registry.discover() - _local_skill_registry = registry return registry diff --git a/tests/test_issue3_startup.py b/tests/test_issue3_startup.py index 386a412..7e1e044 100644 --- a/tests/test_issue3_startup.py +++ b/tests/test_issue3_startup.py @@ -47,3 +47,72 @@ Find me when querying demo local skill. item["name"] == "Demo Local Skill" for item in payload["results"] ) + + +@pytest.mark.asyncio +async def test_local_search_skills_refreshes_registry_between_calls(monkeypatch, tmp_path): + skill_root = tmp_path / "skills" + first_skill_dir = skill_root / "first-local-skill" + first_skill_dir.mkdir(parents=True) + (first_skill_dir / "SKILL.md").write_text( + """--- +name: First Local Skill +description: First local skill for cache regression coverage +--- + +First local skill content. +""", + encoding="utf-8", + ) + + async def forbidden_get_openspace(): + pytest.fail("_get_openspace should not run for source='local'") + + monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root)) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) + monkeypatch.setattr( + "openspace.cloud.embedding.generate_embedding", + lambda text, api_key=None: None, + ) + + first_response = await mcp_server.search_skills( + query="first local skill", + source="local", + limit=5, + auto_import=False, + ) + + first_payload = json.loads(first_response) + assert any( + item["name"] == "First Local Skill" + for item in first_payload["results"] + ) + + second_skill_dir = skill_root / "second-local-skill" + second_skill_dir.mkdir(parents=True) + (second_skill_dir / "SKILL.md").write_text( + """--- +name: Second Local Skill +description: Second local skill created after the first search +--- + +Second local skill content. +""", + encoding="utf-8", + ) + + second_response = await mcp_server.search_skills( + query="second local skill", + source="local", + limit=5, + auto_import=False, + ) + + second_payload = json.loads(second_response) + assert any( + item["name"] == "Second Local Skill" + for item in second_payload["results"] + ) From 1620261783ee8676bbef5a90f017aa755e95e505 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 15:25:08 +0800 Subject: [PATCH 06/15] test: add edge case coverage for local skill search --- tests/test_issue3_startup.py | 59 ++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/tests/test_issue3_startup.py b/tests/test_issue3_startup.py index 7e1e044..9a7b894 100644 --- a/tests/test_issue3_startup.py +++ b/tests/test_issue3_startup.py @@ -116,3 +116,62 @@ Second local skill content. item["name"] == "Second Local Skill" for item in second_payload["results"] ) + + +@pytest.mark.asyncio +async def test_local_search_returns_empty_when_no_registry(monkeypatch): + """source='local' with no discoverable skills should return empty, not crash.""" + + async def forbidden_get_openspace(): + pytest.fail("_get_openspace should not run for source='local'") + + monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) + monkeypatch.setattr(mcp_server, "_get_local_skill_registry", lambda: None) + monkeypatch.setattr( + "openspace.cloud.embedding.generate_embedding", + lambda text, api_key=None: None, + ) + + response = await mcp_server.search_skills( + query="anything", + source="local", + limit=5, + auto_import=False, + ) + + payload = json.loads(response) + assert payload["count"] == 0 + assert payload["results"] == [] + + +@pytest.mark.asyncio +async def test_source_all_still_calls_get_openspace(monkeypatch, tmp_path): + """source='all' must go through _get_openspace(), not the lightweight path.""" + called = {"openspace": False} + + class FakeRegistry: + def list_skills(self): + return [] + + class FakeOpenSpace: + _skill_registry = FakeRegistry() + + async def tracking_get_openspace(): + called["openspace"] = True + return FakeOpenSpace() + + monkeypatch.delenv("OPENSPACE_HOST_SKILL_DIRS", raising=False) + monkeypatch.setattr(mcp_server, "_get_openspace", tracking_get_openspace) + monkeypatch.setattr( + "openspace.cloud.embedding.generate_embedding", + lambda text, api_key=None: None, + ) + + await mcp_server.search_skills( + query="anything", + source="all", + limit=5, + auto_import=False, + ) + + assert called["openspace"], "source='all' should call _get_openspace()" From 63b01cfcef0ff7b4b00502773733d2c8c5e0051e Mon Sep 17 00:00:00 2001 From: xlrrrr Date: Tue, 31 Mar 2026 15:36:13 +0800 Subject: [PATCH 07/15] fix: CLI entry point now respects OPENSPACE_MODEL and OPENSPACE_LLM_* env vars --- openspace/__main__.py | 38 +++++++++++++++++++++++++++++++------- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/openspace/__main__.py b/openspace/__main__.py index 28ce3c2..4a1d6e0 100644 --- a/openspace/__main__.py +++ b/openspace/__main__.py @@ -321,16 +321,41 @@ async def refresh_mcp_cache(config_path: Optional[str] = None): def _load_config(args) -> OpenSpaceConfig: """Load configuration""" + import os + from openspace.host_detection import build_llm_kwargs, build_grounding_config_path + cli_overrides = {} - if args.model: - cli_overrides['llm_model'] = args.model if args.max_iterations is not None: cli_overrides['grounding_max_iterations'] = args.max_iterations if args.timeout is not None: cli_overrides['llm_timeout'] = args.timeout if args.log_level: cli_overrides['log_level'] = args.log_level - + + # Resolve LLM model & credentials + # CLI --model > OPENSPACE_MODEL env > host-agent auto-detect > default + env_model = args.model or os.environ.get("OPENSPACE_MODEL", "") + model, llm_kwargs = build_llm_kwargs(env_model) + cli_overrides['llm_model'] = model + cli_overrides['llm_kwargs'] = llm_kwargs + + max_iter = int(os.environ.get("OPENSPACE_MAX_ITERATIONS", "20")) + enable_rec = os.environ.get("OPENSPACE_ENABLE_RECORDING", "true").lower() in ("true", "1", "yes") + backend_scope_raw = os.environ.get("OPENSPACE_BACKEND_SCOPE") + backend_scope = ( + [b.strip() for b in backend_scope_raw.split(",") if b.strip()] + if backend_scope_raw else None + ) + config_path = build_grounding_config_path() + + if 'grounding_max_iterations' not in cli_overrides: + cli_overrides['grounding_max_iterations'] = max_iter + cli_overrides['enable_recording'] = enable_rec + if backend_scope is not None: + cli_overrides['backend_scope'] = backend_scope + if config_path: + cli_overrides['grounding_config_path'] = config_path + try: # Load from config file if provided if args.config: @@ -338,18 +363,17 @@ def _load_config(args) -> OpenSpaceConfig: with open(args.config, 'r', encoding='utf-8') as f: config_dict = json.load(f) - # Apply CLI overrides + # Apply CLI / env overrides config_dict.update(cli_overrides) config = OpenSpaceConfig(**config_dict) print(f"✓ Loaded from config file: {args.config}") else: - # Use default config + CLI overrides config = OpenSpaceConfig(**cli_overrides) print("✓ Using default configuration") - if cli_overrides: - print(f"✓ CLI overrides: {', '.join(cli_overrides.keys())}") + if args.model: + print(f"✓ CLI overrides: llm_model") if args.log_level: Logger.set_level(args.log_level) From f845c5f7fbf8a10b822dc41833a4a517810d3259 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 16:23:33 +0800 Subject: [PATCH 08/15] fix(security): harden zip extraction and import_skill against path traversal - Add resolve() + is_relative_to() check in _extract_zip() to block nested traversal entries like nested/../../escape.txt - Sanitize server-provided skill name in import_skill() to prevent directory escape via malicious record metadata - Add 6 regression tests covering both attack vectors Closes #17 Co-authored-by: LeftX --- openspace/cloud/client.py | 11 +++- tests/test_zip_path_traversal.py | 89 ++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+), 2 deletions(-) create mode 100644 tests/test_zip_path_traversal.py diff --git a/openspace/cloud/client.py b/openspace/cloud/client.py index fb0792f..7bcd3b1 100644 --- a/openspace/cloud/client.py +++ b/openspace/cloud/client.py @@ -340,7 +340,11 @@ class OpenSpaceClient: record_data = self.fetch_record(skill_id) skill_name = record_data.get("name", skill_id) - skill_dir = target_dir / skill_name + if "/" in skill_name or "\\" in skill_name or skill_name.startswith("."): + skill_name = skill_id + skill_dir = (target_dir / skill_name).resolve() + if not skill_dir.is_relative_to(target_dir.resolve()): + raise CloudError(f"Skill name {skill_name!r} escapes target directory") # Check if already exists locally if skill_dir.exists() and (skill_dir / SKILL_FILENAME).exists(): @@ -401,6 +405,7 @@ class OpenSpaceClient: def _extract_zip(zip_data: bytes, target_dir: Path) -> List[str]: """Extract zip bytes to target directory with path traversal protection.""" extracted: List[str] = [] + resolved_target = target_dir.resolve() try: with zipfile.ZipFile(io.BytesIO(zip_data)) as zf: for info in zf.infolist(): @@ -409,7 +414,9 @@ class OpenSpaceClient: clean_name = Path(info.filename).as_posix() if clean_name.startswith("..") or clean_name.startswith("/"): continue - target_path = target_dir / clean_name + target_path = (target_dir / clean_name).resolve() + if not target_path.is_relative_to(resolved_target): + continue target_path.parent.mkdir(parents=True, exist_ok=True) target_path.write_bytes(zf.read(info)) extracted.append(clean_name) diff --git a/tests/test_zip_path_traversal.py b/tests/test_zip_path_traversal.py new file mode 100644 index 0000000..275176b --- /dev/null +++ b/tests/test_zip_path_traversal.py @@ -0,0 +1,89 @@ +"""Regression tests for zip extraction and import_skill path traversal.""" + +import io +import zipfile +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + + +def _make_zip(entries: dict[str, bytes]) -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + for name, data in entries.items(): + zf.writestr(name, data) + return buf.getvalue() + + +def _get_extract_zip(): + import importlib + mod = importlib.import_module("openspace.cloud.client") + return mod.OpenSpaceClient._extract_zip + + +class TestExtractZip: + def test_normal_file_extracted(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"hello.txt": b"world"}) + result = extract(zip_data, tmp_path) + assert "hello.txt" in result + assert (tmp_path / "hello.txt").read_bytes() == b"world" + + def test_dotdot_prefix_blocked(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"../escape.txt": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + + def test_nested_traversal_blocked(self, tmp_path): + """The real bug: nested/../../escape.txt bypassed the old startswith check.""" + extract = _get_extract_zip() + zip_data = _make_zip({"nested/../../escape.txt": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + assert not (tmp_path.parent / "escape.txt").exists() + + def test_absolute_path_blocked(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"/etc/passwd": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + + +class TestImportSkillNameTraversal: + def test_malicious_record_name_sanitized(self, tmp_path): + from openspace.cloud.client import OpenSpaceClient + + client = OpenSpaceClient.__new__(OpenSpaceClient) + target_dir = tmp_path / "skills" + target_dir.mkdir() + + malicious_name = "../../escapedir" + skill_id = "safe_skill_id" + zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) + + with patch.object(client, "fetch_record", return_value={"name": malicious_name}), \ + patch.object(client, "download_artifact", return_value=zip_data): + result = client.import_skill(skill_id, target_dir) + + assert result["status"] == "success" + resolved = Path(result["local_path"]).resolve() + assert resolved.is_relative_to(target_dir.resolve()) + assert not (tmp_path.parent / "escapedir").exists() + + def test_normal_record_name_works(self, tmp_path): + from openspace.cloud.client import OpenSpaceClient + + client = OpenSpaceClient.__new__(OpenSpaceClient) + target_dir = tmp_path / "skills" + target_dir.mkdir() + + zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) + + with patch.object(client, "fetch_record", return_value={"name": "my-skill"}), \ + patch.object(client, "download_artifact", return_value=zip_data): + result = client.import_skill("some_id", target_dir) + + assert result["status"] == "success" + assert (target_dir / "my-skill" / "SKILL.md").exists() From 64a30760586d98f0621a2ec79ff87d769a2cf464 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 17:03:00 +0800 Subject: [PATCH 09/15] fix: pass max_iterations to grounding agent in no-skill execution path Without this, the no-skill path ignores the resolved max_iterations and uses whatever default the agent has, instead of the configured grounding_max_iterations value. Co-authored-by: wul48527-code --- openspace/tool_layer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/openspace/tool_layer.py b/openspace/tool_layer.py index 1ea419f..4cf4b7b 100644 --- a/openspace/tool_layer.py +++ b/openspace/tool_layer.py @@ -507,6 +507,7 @@ class OpenSpace: f"Executing with GroundingAgent " f"(max {max_iterations} iterations, no skills)..." ) + execution_context["max_iterations"] = max_iterations result = await self._grounding_agent.process(execution_context) execution_time = asyncio.get_event_loop().time() - start_time From 29a3869d49c61fad94193fb78cfd9b8fec54fed2 Mon Sep 17 00:00:00 2001 From: xlrrrr Date: Tue, 31 Mar 2026 17:18:13 +0800 Subject: [PATCH 10/15] fix: improve MiniMax compatibility --- openspace/agents/grounding_agent.py | 15 +++- openspace/host_detection/nanobot.py | 2 +- openspace/host_detection/resolver.py | 33 ++++++++ openspace/llm/client.py | 111 ++++++++++++++++++++++++++- 4 files changed, 153 insertions(+), 8 deletions(-) diff --git a/openspace/agents/grounding_agent.py b/openspace/agents/grounding_agent.py index 4e36cba..4d983d2 100644 --- a/openspace/agents/grounding_agent.py +++ b/openspace/agents/grounding_agent.py @@ -120,6 +120,7 @@ class GroundingAgent(BaseAgent): logger.info(f"Skill registry attached ({count} skill(s) available for mid-iteration retrieval)") _MAX_SINGLE_CONTENT_CHARS = 30_000 + _ITERATION_GUIDANCE_PREFIX = "[INTERNAL ORCHESTRATION NOTE]" @classmethod def _cap_message_content(cls, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: @@ -414,13 +415,19 @@ class GroundingAgent(BaseAgent): # Remove previous iteration guidance to avoid accumulation messages = [ - msg for msg in messages - if not (msg.get("role") == "system" and "Iteration" in msg.get("content", "") and "complete" in msg.get("content", "")) + msg for msg in messages + if not ( + isinstance(msg.get("content"), str) + and msg.get("content", "").startswith(self._ITERATION_GUIDANCE_PREFIX) + ) ] + # MiniMax rejects system messages injected mid-conversation, + # so runtime guidance is sent as an internal user note. guidance_msg = { - "role": "system", - "content": f"Iteration {current_iteration} complete. " + "role": "user", + "content": f"{self._ITERATION_GUIDANCE_PREFIX}\n" + f"Iteration {current_iteration} complete. " f"Check if task is finished - if yes, output {GroundingAgentPrompts.TASK_COMPLETE}. " f"If not, continue with next action." } diff --git a/openspace/host_detection/nanobot.py b/openspace/host_detection/nanobot.py index c06c743..8f787b1 100644 --- a/openspace/host_detection/nanobot.py +++ b/openspace/host_detection/nanobot.py @@ -31,7 +31,7 @@ PROVIDER_REGISTRY: List[tuple] = [ ("zhipu", ("zhipu", "glm", "zai"), ""), ("dashscope", ("qwen", "dashscope"), ""), ("moonshot", ("moonshot", "kimi"), "https://api.moonshot.ai/v1"), - ("minimax", ("minimax",), "https://api.minimax.io/v1"), + ("minimax", ("minimax",), "https://api.minimaxi.com/v1"), ("groq", ("groq",), ""), ] diff --git a/openspace/host_detection/resolver.py b/openspace/host_detection/resolver.py index 3fb3611..d3d9cb0 100644 --- a/openspace/host_detection/resolver.py +++ b/openspace/host_detection/resolver.py @@ -54,12 +54,20 @@ def build_llm_kwargs(model: str) -> tuple[str, Dict[str, Any]]: # and the model name doesn't already carry that prefix, prepend # it so that litellm uses the correct request format (OpenAI- # compatible for gateways vs native for direct providers). + # Skip when the user explicitly provided a model (via OPENSPACE_MODEL + # or --model) AND explicit OPENSPACE_LLM_* overrides — the user knows + # exactly which endpoint they want to hit. _GATEWAY_PROVIDERS = {"openrouter", "aihubmix", "siliconflow"} + _has_explicit_llm_override = bool( + os.environ.get("OPENSPACE_LLM_API_BASE") + or os.environ.get("OPENSPACE_LLM_API_KEY") + ) if ( forced_provider and forced_provider in _GATEWAY_PROVIDERS and resolved_model and not resolved_model.lower().startswith(f"{forced_provider}/") + and not (model and _has_explicit_llm_override) ): resolved_model = f"{forced_provider}/{resolved_model}" logger.info( @@ -102,6 +110,31 @@ def build_llm_kwargs(model: str) -> tuple[str, Dict[str, Any]]: if not resolved_model: resolved_model = "openrouter/anthropic/claude-sonnet-4.5" + # Provider-specific adjustments for litellm routing + if resolved_model and "minimax" in resolved_model.lower(): + final_key = kwargs.get("api_key") + final_base = kwargs.get("api_base", "") + + if final_key: + os.environ.setdefault("MINIMAX_API_KEY", final_key) + if final_base: + os.environ.setdefault("MINIMAX_API_BASE", final_base) + + # api.minimaxi.com (domestic) is OpenAI-compatible but does not fully + # support litellm's minimax-specific request transformations (e.g. tool + # calling format). Switch to the generic openai/ prefix so litellm + # sends standard OpenAI-format requests that minimaxi.com accepts. + if ( + resolved_model.lower().startswith("minimax/") + and "minimaxi.com" in final_base + ): + original = resolved_model + resolved_model = "openai/" + resolved_model.split("/", 1)[1] + logger.info( + "Switched model prefix for minimaxi.com compat: %s -> %s", + original, resolved_model, + ) + if kwargs: safe = { k: (v[:8] + "..." if k == "api_key" and isinstance(v, str) and len(v) > 8 else v) diff --git a/openspace/llm/client.py b/openspace/llm/client.py index 19a1664..cc14c64 100644 --- a/openspace/llm/client.py +++ b/openspace/llm/client.py @@ -406,6 +406,95 @@ class LLMClient: self._logger = Logger.get_logger(__name__) self._last_call_time = 0.0 + @staticmethod + def _merge_consecutive_system_messages(messages: List[Dict]) -> List[Dict]: + """Merge consecutive system messages into one. + + Providers like MiniMax reject requests that contain multiple consecutive + messages with the same role (error 2013 "invalid chat setting"). + Merging is safe for all providers — it simply concatenates the content. + """ + if not messages: + return messages + merged: List[Dict] = [] + for msg in messages: + if ( + merged + and msg.get("role") == "system" + and merged[-1].get("role") == "system" + ): + merged[-1] = { + "role": "system", + "content": merged[-1].get("content", "") + "\n\n" + msg.get("content", ""), + } + else: + merged.append(msg.copy()) + return merged + + @staticmethod + def _is_minimax_model(model: str) -> bool: + return isinstance(model, str) and "minimax" in model.lower() + + @classmethod + def _rewrite_nonleading_system_messages_for_minimax( + cls, + messages: List[Dict], + ) -> List[Dict]: + """Rewrite non-leading system messages into internal user notes for MiniMax.""" + rewritten: List[Dict] = [] + rewritten_count = 0 + + for msg in messages: + msg_copy = msg.copy() + if msg_copy.get("role") == "system" and rewritten: + content = msg_copy.get("content", "") + if isinstance(content, str): + msg_copy["content"] = ( + "[INTERNAL ORCHESTRATION NOTE]\n" + "This note was originally injected as a system message by the " + "agent runtime. Treat it as workflow guidance, not as a new " + "end-user request.\n\n" + f"{content}" + ) + msg_copy["role"] = "user" + rewritten_count += 1 + rewritten.append(msg_copy) + + if rewritten_count: + logger.info( + "Rewrote %d non-leading system message(s) for MiniMax compatibility", + rewritten_count, + ) + + return rewritten + + @classmethod + def _normalize_messages_for_model(cls, messages: List[Dict], model: str) -> List[Dict]: + """Normalize message history only when a provider requires it.""" + if not cls._is_minimax_model(model): + return messages + + minimized_system_history = cls._merge_consecutive_system_messages(messages) + return cls._rewrite_nonleading_system_messages_for_minimax( + minimized_system_history + ) + + @staticmethod + def _serialize_response_field(value): + """Convert provider response fields into plain Python containers.""" + if hasattr(value, "model_dump"): + return value.model_dump(exclude_none=True) + if isinstance(value, list): + return [LLMClient._serialize_response_field(item) for item in value] + if isinstance(value, tuple): + return [LLMClient._serialize_response_field(item) for item in value] + if isinstance(value, dict): + return { + key: LLMClient._serialize_response_field(item) + for key, item in value.items() + } + return value + async def _rate_limit(self): """Apply rate limiting by adding delay between API calls""" if self.rate_limit_delay > 0: @@ -539,6 +628,7 @@ class LLMClient: "model": kwargs.get("model", self.model), **self.litellm_kwargs, } + request_model = completion_kwargs["model"] # Add thinking/reasoning_effort only if explicitly enabled and not using tools enable_thinking = kwargs.get("enable_thinking", self.enable_thinking) @@ -561,10 +651,16 @@ class LLMClient: if enable_thinking: completion_kwargs["reasoning_effort"] = kwargs.get("reasoning_effort", "medium") - # 4. Apply rate limiting + # 4. Normalize messages for providers with stricter role constraints. + current_messages = self._normalize_messages_for_model( + current_messages, + request_model, + ) + + # 5. Apply rate limiting await self._rate_limit() - # 5. Call LLM with retry (single round) + # 6. Call LLM with retry (single round) completion_kwargs["messages"] = current_messages response = await self._call_with_retry(**completion_kwargs) @@ -578,6 +674,11 @@ class LLMClient: "role": "assistant", "content": response_message.content or "", } + + for field_name in ("reasoning_details", "reasoning_content", "name"): + field_value = getattr(response_message, field_name, None) + if field_value: + assistant_message[field_name] = self._serialize_response_field(field_value) tool_calls = getattr(response_message, 'tool_calls', None) if tool_calls: @@ -722,6 +823,10 @@ class LLMClient: "content": summary_prompt } current_messages.append(summary_message) + current_messages = self._normalize_messages_for_model( + current_messages, + request_model, + ) # Apply rate limiting before summary call await self._rate_limit() @@ -729,7 +834,7 @@ class LLMClient: # Call LLM to generate summary (without tools) summary_kwargs = { **self.litellm_kwargs, - "model": self.model, + "model": request_model, "messages": current_messages, "tools": [], "tool_choice": "none", From e48e22afc8ec5c9c223630817f8526d09fbb07b9 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 17:18:38 +0800 Subject: [PATCH 11/15] fix: use unique workflow ID to prevent collision across roots workflow_dir.name was used as the discovery key and API ID, so two different WORKFLOW_ROOTS containing a leaf directory with the same name would silently drop one. Use root name + relative path joined with __ as a stable unique ID instead. Co-authored-by: wul48527-code --- openspace/dashboard_server.py | 20 +++++++-- tests/test_workflow_id.py | 79 +++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 4 deletions(-) create mode 100644 tests/test_workflow_id.py diff --git a/openspace/dashboard_server.py b/openspace/dashboard_server.py index 96d3e50..4333aef 100644 --- a/openspace/dashboard_server.py +++ b/openspace/dashboard_server.py @@ -419,6 +419,18 @@ def _build_lineage_payload(skill_id: str, store: SkillStore) -> Dict[str, Any]: } +def _workflow_id(workflow_dir: Path) -> str: + """Stable short ID for a workflow directory, unique across roots.""" + resolved = workflow_dir.resolve() + for root in WORKFLOW_ROOTS: + try: + rel = resolved.relative_to(root.resolve()) + return f"{root.name}__{'__'.join(rel.parts)}" + except ValueError: + continue + return workflow_dir.name + + def _discover_workflow_dirs() -> List[Path]: discovered: Dict[str, Path] = {} for root in WORKFLOW_ROOTS: @@ -439,14 +451,14 @@ def _scan_workflow_tree(directory: Path, discovered: Dict[str, Path], *, _depth: if not child.is_dir(): continue if (child / "metadata.json").exists() or (child / "traj.jsonl").exists(): - discovered.setdefault(child.name, child) + discovered.setdefault(str(child.resolve()), child) else: _scan_workflow_tree(child, discovered, _depth=_depth + 1, _max_depth=_max_depth) def _get_workflow_dir(workflow_id: str) -> Optional[Path]: for path in _discover_workflow_dirs(): - if path.name == workflow_id: + if _workflow_id(path) == workflow_id: return path return None @@ -464,7 +476,7 @@ def _build_workflow_summary(workflow_dir: Path) -> Dict[str, Any]: for candidate in video_candidates: if candidate.exists(): rel = candidate.relative_to(workflow_dir).as_posix() - video_url = url_for("workflow_artifact", workflow_id=workflow_dir.name, artifact_path=rel) + video_url = url_for("workflow_artifact", workflow_id=_workflow_id(workflow_dir), artifact_path=rel) break outcome = metadata.get("execution_outcome") or {} @@ -514,7 +526,7 @@ def _build_workflow_summary(workflow_dir: Path) -> Dict[str, Any]: iterations = len(trajectory) return { - "id": workflow_dir.name, + "id": _workflow_id(workflow_dir), "path": str(workflow_dir), "task_id": metadata.get("task_id") or metadata.get("task_name") or workflow_dir.name, "task_name": metadata.get("task_name") or metadata.get("task_id") or workflow_dir.name, diff --git a/tests/test_workflow_id.py b/tests/test_workflow_id.py new file mode 100644 index 0000000..d4ca82f --- /dev/null +++ b/tests/test_workflow_id.py @@ -0,0 +1,79 @@ +"""Tests for workflow ID uniqueness and discovery.""" + +from pathlib import Path +from unittest.mock import patch + +import pytest + +from openspace import dashboard_server + + +def _make_workflow(path: Path): + """Create a minimal workflow directory with metadata.json.""" + path.mkdir(parents=True, exist_ok=True) + (path / "metadata.json").write_text("{}", encoding="utf-8") + + +class TestWorkflowId: + def test_same_leaf_name_different_roots_get_unique_ids(self, tmp_path): + root_a = tmp_path / "root_a" + root_b = tmp_path / "root_b" + wf_a = root_a / "task1" + wf_b = root_b / "task1" + _make_workflow(wf_a) + _make_workflow(wf_b) + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): + id_a = dashboard_server._workflow_id(wf_a) + id_b = dashboard_server._workflow_id(wf_b) + + assert id_a != id_b + assert "task1" in id_a + assert "task1" in id_b + + def test_nested_workflow_id_includes_path(self, tmp_path): + root = tmp_path / "root" + wf = root / "sub" / "deep" / "task1" + _make_workflow(wf) + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]): + wf_id = dashboard_server._workflow_id(wf) + + assert wf_id == "root__sub__deep__task1" + + def test_workflow_outside_roots_falls_back_to_name(self, tmp_path): + wf = tmp_path / "orphan_workflow" + _make_workflow(wf) + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", []): + wf_id = dashboard_server._workflow_id(wf) + + assert wf_id == "orphan_workflow" + + +class TestDiscoverWorkflowDirs: + def test_same_name_workflows_both_discovered(self, tmp_path): + root_a = tmp_path / "root_a" + root_b = tmp_path / "root_b" + _make_workflow(root_a / "task1") + _make_workflow(root_b / "task1") + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): + dirs = dashboard_server._discover_workflow_dirs() + + assert len(dirs) == 2 + + def test_get_workflow_dir_resolves_correct_path(self, tmp_path): + root_a = tmp_path / "root_a" + root_b = tmp_path / "root_b" + _make_workflow(root_a / "task1") + _make_workflow(root_b / "task1") + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): + id_a = dashboard_server._workflow_id(root_a / "task1") + id_b = dashboard_server._workflow_id(root_b / "task1") + found_a = dashboard_server._get_workflow_dir(id_a) + found_b = dashboard_server._get_workflow_dir(id_b) + + assert found_a.resolve() == (root_a / "task1").resolve() + assert found_b.resolve() == (root_b / "task1").resolve() From 4032fe682d903c96a8fdb40d418bf7db7aec89b8 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 17:24:12 +0800 Subject: [PATCH 12/15] test: add regression tests for max_iterations in no-skill path --- tests/test_max_iterations.py | 64 ++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 tests/test_max_iterations.py diff --git a/tests/test_max_iterations.py b/tests/test_max_iterations.py new file mode 100644 index 0000000..04e6863 --- /dev/null +++ b/tests/test_max_iterations.py @@ -0,0 +1,64 @@ +"""Test that max_iterations is passed to grounding agent in no-skill path.""" + +import asyncio +import types +from unittest.mock import AsyncMock, patch + +import pytest + +from openspace.tool_layer import OpenSpace, OpenSpaceConfig + + +def _make_openspace(configured_max: int = 50) -> OpenSpace: + config = OpenSpaceConfig() + config.grounding_max_iterations = configured_max + os_inst = OpenSpace(config) + os_inst._initialized = True + os_inst._running = False + os_inst._task_done = asyncio.Event() + os_inst._task_done.set() + os_inst._grounding_client = types.SimpleNamespace(_registry={}) + os_inst._recording_manager = None + os_inst._skill_registry = None + os_inst._execution_analyzer = None + os_inst._skill_evolver = None + return os_inst + + +@pytest.mark.asyncio +async def test_no_skill_path_passes_configured_max_iterations(): + """When no skills match, the configured grounding_max_iterations + must be forwarded to the agent, not silently dropped.""" + os_inst = _make_openspace(configured_max=50) + recorded = {} + + async def fake_process(context): + recorded.update(context) + return {"status": "success", "iterations": 1, "tool_executions": []} + + os_inst._grounding_agent = types.SimpleNamespace(process=fake_process) + os_inst._maybe_analyze_execution = AsyncMock() + os_inst._maybe_evolve_quality = AsyncMock() + + await os_inst.execute("do something") + + assert recorded.get("max_iterations") == 50 + + +@pytest.mark.asyncio +async def test_no_skill_path_uses_caller_override_when_larger(): + """Caller-provided max_iterations should win when larger than config.""" + os_inst = _make_openspace(configured_max=20) + recorded = {} + + async def fake_process(context): + recorded.update(context) + return {"status": "success", "iterations": 1, "tool_executions": []} + + os_inst._grounding_agent = types.SimpleNamespace(process=fake_process) + os_inst._maybe_analyze_execution = AsyncMock() + os_inst._maybe_evolve_quality = AsyncMock() + + await os_inst.execute("do something", max_iterations=100) + + assert recorded.get("max_iterations") == 100 From 028c5b01f9515dece7f4164c9564803b577de0b2 Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Tue, 31 Mar 2026 17:38:39 +0800 Subject: [PATCH 13/15] fix: use hash-based workflow ID to prevent separator collisions The previous __-joined scheme was not injective: a directory named a__b and a nested path a/b both mapped to the same ID. Use a sha256 hash suffix of the resolved path instead, which is collision-free and keeps the dir name as a human-readable prefix. Added regression test for separator collision case. --- openspace/dashboard_server.py | 18 +++++++++--------- tests/test_workflow_id.py | 29 ++++++++++++++++++++++------- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/openspace/dashboard_server.py b/openspace/dashboard_server.py index 4333aef..f6916b6 100644 --- a/openspace/dashboard_server.py +++ b/openspace/dashboard_server.py @@ -420,15 +420,15 @@ def _build_lineage_payload(skill_id: str, store: SkillStore) -> Dict[str, Any]: def _workflow_id(workflow_dir: Path) -> str: - """Stable short ID for a workflow directory, unique across roots.""" - resolved = workflow_dir.resolve() - for root in WORKFLOW_ROOTS: - try: - rel = resolved.relative_to(root.resolve()) - return f"{root.name}__{'__'.join(rel.parts)}" - except ValueError: - continue - return workflow_dir.name + """Stable short ID for a workflow directory, unique across roots. + + Uses a hash suffix derived from the resolved path to avoid collisions + when directory names contain the separator character. + """ + import hashlib + resolved = str(workflow_dir.resolve()) + path_hash = hashlib.sha256(resolved.encode()).hexdigest()[:8] + return f"{workflow_dir.name}_{path_hash}" def _discover_workflow_dirs() -> List[Path]: diff --git a/tests/test_workflow_id.py b/tests/test_workflow_id.py index d4ca82f..84bf3e1 100644 --- a/tests/test_workflow_id.py +++ b/tests/test_workflow_id.py @@ -28,27 +28,42 @@ class TestWorkflowId: id_b = dashboard_server._workflow_id(wf_b) assert id_a != id_b - assert "task1" in id_a - assert "task1" in id_b + assert id_a.startswith("task1_") + assert id_b.startswith("task1_") - def test_nested_workflow_id_includes_path(self, tmp_path): + def test_id_contains_dir_name_and_hash(self, tmp_path): root = tmp_path / "root" - wf = root / "sub" / "deep" / "task1" + wf = root / "my-task" _make_workflow(wf) with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]): wf_id = dashboard_server._workflow_id(wf) - assert wf_id == "root__sub__deep__task1" + assert wf_id.startswith("my-task_") + assert len(wf_id) == len("my-task_") + 8 # 8-char hex hash - def test_workflow_outside_roots_falls_back_to_name(self, tmp_path): + def test_separator_collision_produces_different_ids(self, tmp_path): + """Regression: a dir named 'a__b' must not collide with path a/b.""" + root = tmp_path / "root" + flat = root / "a__b" + nested = root / "a" / "b" + _make_workflow(flat) + _make_workflow(nested) + + with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]): + id_flat = dashboard_server._workflow_id(flat) + id_nested = dashboard_server._workflow_id(nested) + + assert id_flat != id_nested + + def test_workflow_outside_roots_still_works(self, tmp_path): wf = tmp_path / "orphan_workflow" _make_workflow(wf) with patch.object(dashboard_server, "WORKFLOW_ROOTS", []): wf_id = dashboard_server._workflow_id(wf) - assert wf_id == "orphan_workflow" + assert wf_id.startswith("orphan_workflow_") class TestDiscoverWorkflowDirs: From f05514845de0288b3a9032cfd7c1f360e6f43a8a Mon Sep 17 00:00:00 2001 From: spidercatfly Date: Thu, 2 Apr 2026 22:18:31 +0800 Subject: [PATCH 14/15] chore: remove outdated test files --- tests/test_issue3_startup.py | 177 ------------------------------- tests/test_max_iterations.py | 64 ----------- tests/test_workflow_id.py | 94 ---------------- tests/test_zip_path_traversal.py | 89 ---------------- 4 files changed, 424 deletions(-) delete mode 100644 tests/test_issue3_startup.py delete mode 100644 tests/test_max_iterations.py delete mode 100644 tests/test_workflow_id.py delete mode 100644 tests/test_zip_path_traversal.py diff --git a/tests/test_issue3_startup.py b/tests/test_issue3_startup.py deleted file mode 100644 index 9a7b894..0000000 --- a/tests/test_issue3_startup.py +++ /dev/null @@ -1,177 +0,0 @@ -import json - -import pytest - -from openspace import mcp_server - - -@pytest.mark.asyncio -async def test_local_search_skills_does_not_initialize_openspace(monkeypatch, tmp_path): - skill_root = tmp_path / "skills" - skill_dir = skill_root / "demo-local-skill" - skill_dir.mkdir(parents=True) - (skill_dir / "SKILL.md").write_text( - """--- -name: Demo Local Skill -description: Local regression test skill ---- - -Find me when querying demo local skill. -""", - encoding="utf-8", - ) - - async def forbidden_get_openspace(): - pytest.fail("_get_openspace should not run for source='local'") - - monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root)) - monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) - monkeypatch.delenv("OPENAI_API_KEY", raising=False) - monkeypatch.delenv("OPENAI_BASE_URL", raising=False) - monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) - monkeypatch.setattr( - "openspace.cloud.embedding.generate_embedding", - lambda text, api_key=None: None, - ) - - response = await mcp_server.search_skills( - query="demo local skill", - source="local", - limit=5, - auto_import=False, - ) - - payload = json.loads(response) - assert payload["count"] >= 1 - assert any( - item["name"] == "Demo Local Skill" - for item in payload["results"] - ) - - -@pytest.mark.asyncio -async def test_local_search_skills_refreshes_registry_between_calls(monkeypatch, tmp_path): - skill_root = tmp_path / "skills" - first_skill_dir = skill_root / "first-local-skill" - first_skill_dir.mkdir(parents=True) - (first_skill_dir / "SKILL.md").write_text( - """--- -name: First Local Skill -description: First local skill for cache regression coverage ---- - -First local skill content. -""", - encoding="utf-8", - ) - - async def forbidden_get_openspace(): - pytest.fail("_get_openspace should not run for source='local'") - - monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root)) - monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) - monkeypatch.delenv("OPENAI_API_KEY", raising=False) - monkeypatch.delenv("OPENAI_BASE_URL", raising=False) - monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) - monkeypatch.setattr( - "openspace.cloud.embedding.generate_embedding", - lambda text, api_key=None: None, - ) - - first_response = await mcp_server.search_skills( - query="first local skill", - source="local", - limit=5, - auto_import=False, - ) - - first_payload = json.loads(first_response) - assert any( - item["name"] == "First Local Skill" - for item in first_payload["results"] - ) - - second_skill_dir = skill_root / "second-local-skill" - second_skill_dir.mkdir(parents=True) - (second_skill_dir / "SKILL.md").write_text( - """--- -name: Second Local Skill -description: Second local skill created after the first search ---- - -Second local skill content. -""", - encoding="utf-8", - ) - - second_response = await mcp_server.search_skills( - query="second local skill", - source="local", - limit=5, - auto_import=False, - ) - - second_payload = json.loads(second_response) - assert any( - item["name"] == "Second Local Skill" - for item in second_payload["results"] - ) - - -@pytest.mark.asyncio -async def test_local_search_returns_empty_when_no_registry(monkeypatch): - """source='local' with no discoverable skills should return empty, not crash.""" - - async def forbidden_get_openspace(): - pytest.fail("_get_openspace should not run for source='local'") - - monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace) - monkeypatch.setattr(mcp_server, "_get_local_skill_registry", lambda: None) - monkeypatch.setattr( - "openspace.cloud.embedding.generate_embedding", - lambda text, api_key=None: None, - ) - - response = await mcp_server.search_skills( - query="anything", - source="local", - limit=5, - auto_import=False, - ) - - payload = json.loads(response) - assert payload["count"] == 0 - assert payload["results"] == [] - - -@pytest.mark.asyncio -async def test_source_all_still_calls_get_openspace(monkeypatch, tmp_path): - """source='all' must go through _get_openspace(), not the lightweight path.""" - called = {"openspace": False} - - class FakeRegistry: - def list_skills(self): - return [] - - class FakeOpenSpace: - _skill_registry = FakeRegistry() - - async def tracking_get_openspace(): - called["openspace"] = True - return FakeOpenSpace() - - monkeypatch.delenv("OPENSPACE_HOST_SKILL_DIRS", raising=False) - monkeypatch.setattr(mcp_server, "_get_openspace", tracking_get_openspace) - monkeypatch.setattr( - "openspace.cloud.embedding.generate_embedding", - lambda text, api_key=None: None, - ) - - await mcp_server.search_skills( - query="anything", - source="all", - limit=5, - auto_import=False, - ) - - assert called["openspace"], "source='all' should call _get_openspace()" diff --git a/tests/test_max_iterations.py b/tests/test_max_iterations.py deleted file mode 100644 index 04e6863..0000000 --- a/tests/test_max_iterations.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Test that max_iterations is passed to grounding agent in no-skill path.""" - -import asyncio -import types -from unittest.mock import AsyncMock, patch - -import pytest - -from openspace.tool_layer import OpenSpace, OpenSpaceConfig - - -def _make_openspace(configured_max: int = 50) -> OpenSpace: - config = OpenSpaceConfig() - config.grounding_max_iterations = configured_max - os_inst = OpenSpace(config) - os_inst._initialized = True - os_inst._running = False - os_inst._task_done = asyncio.Event() - os_inst._task_done.set() - os_inst._grounding_client = types.SimpleNamespace(_registry={}) - os_inst._recording_manager = None - os_inst._skill_registry = None - os_inst._execution_analyzer = None - os_inst._skill_evolver = None - return os_inst - - -@pytest.mark.asyncio -async def test_no_skill_path_passes_configured_max_iterations(): - """When no skills match, the configured grounding_max_iterations - must be forwarded to the agent, not silently dropped.""" - os_inst = _make_openspace(configured_max=50) - recorded = {} - - async def fake_process(context): - recorded.update(context) - return {"status": "success", "iterations": 1, "tool_executions": []} - - os_inst._grounding_agent = types.SimpleNamespace(process=fake_process) - os_inst._maybe_analyze_execution = AsyncMock() - os_inst._maybe_evolve_quality = AsyncMock() - - await os_inst.execute("do something") - - assert recorded.get("max_iterations") == 50 - - -@pytest.mark.asyncio -async def test_no_skill_path_uses_caller_override_when_larger(): - """Caller-provided max_iterations should win when larger than config.""" - os_inst = _make_openspace(configured_max=20) - recorded = {} - - async def fake_process(context): - recorded.update(context) - return {"status": "success", "iterations": 1, "tool_executions": []} - - os_inst._grounding_agent = types.SimpleNamespace(process=fake_process) - os_inst._maybe_analyze_execution = AsyncMock() - os_inst._maybe_evolve_quality = AsyncMock() - - await os_inst.execute("do something", max_iterations=100) - - assert recorded.get("max_iterations") == 100 diff --git a/tests/test_workflow_id.py b/tests/test_workflow_id.py deleted file mode 100644 index 84bf3e1..0000000 --- a/tests/test_workflow_id.py +++ /dev/null @@ -1,94 +0,0 @@ -"""Tests for workflow ID uniqueness and discovery.""" - -from pathlib import Path -from unittest.mock import patch - -import pytest - -from openspace import dashboard_server - - -def _make_workflow(path: Path): - """Create a minimal workflow directory with metadata.json.""" - path.mkdir(parents=True, exist_ok=True) - (path / "metadata.json").write_text("{}", encoding="utf-8") - - -class TestWorkflowId: - def test_same_leaf_name_different_roots_get_unique_ids(self, tmp_path): - root_a = tmp_path / "root_a" - root_b = tmp_path / "root_b" - wf_a = root_a / "task1" - wf_b = root_b / "task1" - _make_workflow(wf_a) - _make_workflow(wf_b) - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): - id_a = dashboard_server._workflow_id(wf_a) - id_b = dashboard_server._workflow_id(wf_b) - - assert id_a != id_b - assert id_a.startswith("task1_") - assert id_b.startswith("task1_") - - def test_id_contains_dir_name_and_hash(self, tmp_path): - root = tmp_path / "root" - wf = root / "my-task" - _make_workflow(wf) - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]): - wf_id = dashboard_server._workflow_id(wf) - - assert wf_id.startswith("my-task_") - assert len(wf_id) == len("my-task_") + 8 # 8-char hex hash - - def test_separator_collision_produces_different_ids(self, tmp_path): - """Regression: a dir named 'a__b' must not collide with path a/b.""" - root = tmp_path / "root" - flat = root / "a__b" - nested = root / "a" / "b" - _make_workflow(flat) - _make_workflow(nested) - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]): - id_flat = dashboard_server._workflow_id(flat) - id_nested = dashboard_server._workflow_id(nested) - - assert id_flat != id_nested - - def test_workflow_outside_roots_still_works(self, tmp_path): - wf = tmp_path / "orphan_workflow" - _make_workflow(wf) - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", []): - wf_id = dashboard_server._workflow_id(wf) - - assert wf_id.startswith("orphan_workflow_") - - -class TestDiscoverWorkflowDirs: - def test_same_name_workflows_both_discovered(self, tmp_path): - root_a = tmp_path / "root_a" - root_b = tmp_path / "root_b" - _make_workflow(root_a / "task1") - _make_workflow(root_b / "task1") - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): - dirs = dashboard_server._discover_workflow_dirs() - - assert len(dirs) == 2 - - def test_get_workflow_dir_resolves_correct_path(self, tmp_path): - root_a = tmp_path / "root_a" - root_b = tmp_path / "root_b" - _make_workflow(root_a / "task1") - _make_workflow(root_b / "task1") - - with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]): - id_a = dashboard_server._workflow_id(root_a / "task1") - id_b = dashboard_server._workflow_id(root_b / "task1") - found_a = dashboard_server._get_workflow_dir(id_a) - found_b = dashboard_server._get_workflow_dir(id_b) - - assert found_a.resolve() == (root_a / "task1").resolve() - assert found_b.resolve() == (root_b / "task1").resolve() diff --git a/tests/test_zip_path_traversal.py b/tests/test_zip_path_traversal.py deleted file mode 100644 index 275176b..0000000 --- a/tests/test_zip_path_traversal.py +++ /dev/null @@ -1,89 +0,0 @@ -"""Regression tests for zip extraction and import_skill path traversal.""" - -import io -import zipfile -from pathlib import Path -from unittest.mock import patch, MagicMock - -import pytest - - -def _make_zip(entries: dict[str, bytes]) -> bytes: - buf = io.BytesIO() - with zipfile.ZipFile(buf, "w") as zf: - for name, data in entries.items(): - zf.writestr(name, data) - return buf.getvalue() - - -def _get_extract_zip(): - import importlib - mod = importlib.import_module("openspace.cloud.client") - return mod.OpenSpaceClient._extract_zip - - -class TestExtractZip: - def test_normal_file_extracted(self, tmp_path): - extract = _get_extract_zip() - zip_data = _make_zip({"hello.txt": b"world"}) - result = extract(zip_data, tmp_path) - assert "hello.txt" in result - assert (tmp_path / "hello.txt").read_bytes() == b"world" - - def test_dotdot_prefix_blocked(self, tmp_path): - extract = _get_extract_zip() - zip_data = _make_zip({"../escape.txt": b"bad"}) - result = extract(zip_data, tmp_path) - assert result == [] - - def test_nested_traversal_blocked(self, tmp_path): - """The real bug: nested/../../escape.txt bypassed the old startswith check.""" - extract = _get_extract_zip() - zip_data = _make_zip({"nested/../../escape.txt": b"bad"}) - result = extract(zip_data, tmp_path) - assert result == [] - assert not (tmp_path.parent / "escape.txt").exists() - - def test_absolute_path_blocked(self, tmp_path): - extract = _get_extract_zip() - zip_data = _make_zip({"/etc/passwd": b"bad"}) - result = extract(zip_data, tmp_path) - assert result == [] - - -class TestImportSkillNameTraversal: - def test_malicious_record_name_sanitized(self, tmp_path): - from openspace.cloud.client import OpenSpaceClient - - client = OpenSpaceClient.__new__(OpenSpaceClient) - target_dir = tmp_path / "skills" - target_dir.mkdir() - - malicious_name = "../../escapedir" - skill_id = "safe_skill_id" - zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) - - with patch.object(client, "fetch_record", return_value={"name": malicious_name}), \ - patch.object(client, "download_artifact", return_value=zip_data): - result = client.import_skill(skill_id, target_dir) - - assert result["status"] == "success" - resolved = Path(result["local_path"]).resolve() - assert resolved.is_relative_to(target_dir.resolve()) - assert not (tmp_path.parent / "escapedir").exists() - - def test_normal_record_name_works(self, tmp_path): - from openspace.cloud.client import OpenSpaceClient - - client = OpenSpaceClient.__new__(OpenSpaceClient) - target_dir = tmp_path / "skills" - target_dir.mkdir() - - zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) - - with patch.object(client, "fetch_record", return_value={"name": "my-skill"}), \ - patch.object(client, "download_artifact", return_value=zip_data): - result = client.import_skill("some_id", target_dir) - - assert result["status"] == "success" - assert (target_dir / "my-skill" / "SKILL.md").exists() From e1a85244758149498d82d9b311689cbfb1dbced8 Mon Sep 17 00:00:00 2001 From: spidercatfly Date: Thu, 2 Apr 2026 23:17:44 +0800 Subject: [PATCH 15/15] feat: migrate cloud search to server-side embedding endpoint --- openspace/cloud/client.py | 29 +++++++ openspace/cloud/search.py | 157 ++++++++++++++++++++++++-------------- openspace/mcp_server.py | 59 ++++++-------- 3 files changed, 152 insertions(+), 93 deletions(-) diff --git a/openspace/cloud/client.py b/openspace/cloud/client.py index 7bcd3b1..f97b2ac 100644 --- a/openspace/cloud/client.py +++ b/openspace/cloud/client.py @@ -5,6 +5,7 @@ All methods are **synchronous** (use ``urllib``). In async contexts Provides both low-level HTTP operations and higher-level workflows: - ``fetch_record`` / ``download_artifact`` / ``fetch_metadata`` + - ``search_record_embeddings`` - ``stage_artifact`` / ``create_record`` - ``upload_skill`` (stage → diff → create — full workflow) - ``import_skill`` (fetch → download → extract — full workflow) @@ -29,6 +30,7 @@ logger = logging.getLogger("openspace.cloud") SKILL_FILENAME = "SKILL.md" SKILL_ID_FILENAME = ".skill_id" +RECORD_EMBEDDING_SEARCH_MAX_LIMIT = 300 _TEXT_EXTENSIONS = frozenset({ ".md", ".txt", ".yaml", ".yml", ".json", ".py", ".sh", ".toml", @@ -142,6 +144,33 @@ class OpenSpaceClient: return all_items + def search_record_embeddings( + self, + *, + query: str, + limit: int = RECORD_EMBEDDING_SEARCH_MAX_LIMIT, + level: Optional[str] = None, + tags: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + """POST /records/embeddings/search — fetch server-ranked embedding rows.""" + search_request_payload: Dict[str, Any] = { + "query": query, + "limit": limit, + } + if level: + search_request_payload["level"] = level + if tags: + search_request_payload["tags"] = tags + + _, response_body = self._request( + "POST", + "/records/embeddings/search", + body=json.dumps(search_request_payload).encode("utf-8"), + extra_headers={"Content-Type": "application/json"}, + timeout=30, + ) + return json.loads(response_body.decode("utf-8")) + def stage_artifact(self, skill_dir: Path) -> tuple[str, int]: """POST /artifacts/stage — upload skill files. diff --git a/openspace/cloud/search.py b/openspace/cloud/search.py index 678f71b..bd25436 100644 --- a/openspace/cloud/search.py +++ b/openspace/cloud/search.py @@ -18,6 +18,7 @@ import re from typing import Any, Dict, List, Optional logger = logging.getLogger("openspace.cloud") +CLOUD_EMBEDDING_SEARCH_MAX_LIMIT = 300 def _check_safety(text: str) -> list[str]: @@ -159,36 +160,43 @@ class SkillSearchEngine: from openspace.cloud.embedding import cosine_similarity scored = [] - for c in candidates: - name = c.get("name", "") - slug = c.get("skill_id", name).split("__")[0].replace(":", "-") + for candidate in candidates: + candidate_name = candidate.get("name", "") + candidate_slug = candidate.get("skill_id", candidate_name).split("__")[0].replace(":", "-") - # Vector score - vector_score = 0.0 + # Vector score. If client-side query embeddings are unavailable, + # reuse the server-side cloud rank so cloud results keep semantic signal. + vector_score: Optional[float] = None + ranking_signal_score = 0.0 if query_embedding: - skill_emb = c.get("_embedding") - if skill_emb and isinstance(skill_emb, list): - vector_score = cosine_similarity(query_embedding, skill_emb) + candidate_embedding = candidate.get("_embedding") + if candidate_embedding and isinstance(candidate_embedding, list): + vector_score = cosine_similarity(query_embedding, candidate_embedding) + ranking_signal_score = vector_score + elif isinstance(candidate.get("_search_rank"), (int, float)): + ranking_signal_score = float(candidate["_search_rank"]) # Lexical boost - lexical = _lexical_boost(query_tokens, name, slug) + lexical_boost = _lexical_boost(query_tokens, candidate_name, candidate_slug) - final_score = vector_score + lexical + final_score = ranking_signal_score + lexical_boost - entry: Dict[str, Any] = { - "skill_id": c.get("skill_id", ""), - "name": name, - "description": c.get("description", ""), - "source": c.get("source", ""), + result_entry: Dict[str, Any] = { + "skill_id": candidate.get("skill_id", ""), + "name": candidate_name, + "description": candidate.get("description", ""), + "source": candidate.get("source", ""), "score": round(final_score, 4), } - if vector_score > 0: - entry["vector_score"] = round(vector_score, 4) + if vector_score is not None and vector_score > 0: + result_entry["vector_score"] = round(vector_score, 4) + if isinstance(candidate.get("_search_rank"), (int, float)): + result_entry["server_search_rank"] = round(float(candidate["_search_rank"]), 4) # Include optional fields for key in ("path", "visibility", "created_by", "origin", "tags", "quality", "safety_flags"): - if c.get(key): - entry[key] = c[key] - scored.append(entry) + if candidate.get(key): + result_entry[key] = candidate[key] + scored.append(result_entry) scored.sort(key=lambda x: -x["score"]) return scored @@ -275,47 +283,85 @@ def build_local_candidates( def build_cloud_candidates( - items: List[Dict[str, Any]], + cloud_items: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: - """Build search candidate dicts from cloud metadata items. + """Build search candidate dicts from cloud metadata/search items. Args: - items: Items from ``OpenSpaceClient.fetch_metadata()``. + cloud_items: Items from cloud metadata or embedding search endpoints. Returns: List of candidate dicts (with safety filtering applied). """ candidates: List[Dict[str, Any]] = [] - for item in items: - name = item.get("name", "") - desc = item.get("description", "") - tags = item.get("tags", []) - safety_text = f"{name}\n{desc}\n{' '.join(tags)}" + for item in cloud_items: + candidate_name = item.get("name", "") + candidate_description = item.get("description", "") + candidate_tags = item.get("tags", []) + safety_text = f"{candidate_name}\n{candidate_description}\n{' '.join(candidate_tags)}" flags = _check_safety(safety_text) if not _is_safe(flags): continue - c_entry: Dict[str, Any] = { + candidate_entry: Dict[str, Any] = { "skill_id": item.get("record_id", ""), - "name": name, - "description": desc, + "name": candidate_name, + "description": candidate_description, "source": "cloud", "visibility": item.get("visibility", "public"), "is_local": False, "created_by": item.get("created_by", ""), "origin": item.get("origin", ""), - "tags": tags, + "tags": candidate_tags, "safety_flags": flags if flags else None, } # Carry pre-computed embedding - platform_emb = item.get("embedding") - if platform_emb and isinstance(platform_emb, list): - c_entry["_embedding"] = platform_emb - candidates.append(c_entry) + server_embedding = item.get("embedding") + if server_embedding and isinstance(server_embedding, list): + candidate_entry["_embedding"] = server_embedding + server_search_rank = item.get("search_rank") + if isinstance(server_search_rank, (int, float)): + candidate_entry["_search_rank"] = float(server_search_rank) + candidates.append(candidate_entry) return candidates +def build_cloud_results( + cloud_search_items: List[Dict[str, Any]], + *, + limit: int, +) -> List[Dict[str, Any]]: + """Map server-ranked cloud search rows to MCP search result shape.""" + results: List[Dict[str, Any]] = [] + seen_names: set[str] = set() + + for candidate in build_cloud_candidates(cloud_search_items): + candidate_name = candidate.get("name", "") + dedupe_name = candidate_name or candidate.get("skill_id", "") + if dedupe_name in seen_names: + continue + seen_names.add(dedupe_name) + + entry: Dict[str, Any] = { + "skill_id": candidate.get("skill_id", ""), + "name": candidate_name, + "description": candidate.get("description", ""), + "source": "cloud", + "score": round(float(candidate.get("_search_rank", 0.0)), 4), + } + if isinstance(candidate.get("_search_rank"), (int, float)): + entry["server_search_rank"] = round(float(candidate["_search_rank"]), 4) + for key in ("visibility", "created_by", "origin", "tags", "safety_flags"): + if candidate.get(key): + entry[key] = candidate[key] + results.append(entry) + if len(results) >= limit: + break + + return results + + async def hybrid_search_skills( query: str, local_skills: list = None, @@ -341,8 +387,8 @@ async def hybrid_search_skills( """ from openspace.cloud.embedding import generate_embedding - q = query.strip() - if not q: + normalized_query = query.strip() + if not normalized_query: return [] candidates: List[Dict[str, Any]] = [] @@ -357,16 +403,16 @@ async def hybrid_search_skills( auth_headers, api_base = get_openspace_auth() if auth_headers: - client = OpenSpaceClient(auth_headers, api_base) - try: - from openspace.cloud.embedding import resolve_embedding_api - has_emb = bool(resolve_embedding_api()[0]) - except Exception: - has_emb = False - items = await asyncio.to_thread( - client.fetch_metadata, include_embedding=has_emb, limit=200, + cloud_client = OpenSpaceClient(auth_headers, api_base) + cloud_result_limit = limit if source == "cloud" else CLOUD_EMBEDDING_SEARCH_MAX_LIMIT + cloud_search_items = await asyncio.to_thread( + cloud_client.search_record_embeddings, + query=normalized_query, + limit=cloud_result_limit, ) - candidates.extend(build_cloud_candidates(items)) + if source == "cloud": + return build_cloud_results(cloud_search_items, limit=limit) + candidates.extend(build_cloud_candidates(cloud_search_items)) except Exception as e: logger.warning(f"hybrid_search_skills: cloud unavailable: {e}") @@ -376,18 +422,17 @@ async def hybrid_search_skills( # query embedding (optional — key/URL resolved inside generate_embedding) query_embedding: Optional[List[float]] = None try: - query_embedding = await asyncio.to_thread(generate_embedding, q) + query_embedding = await asyncio.to_thread(generate_embedding, normalized_query) if query_embedding: - for c in candidates: - if not c.get("_embedding") and c.get("_embedding_text"): - emb = await asyncio.to_thread( - generate_embedding, c["_embedding_text"], + for candidate in candidates: + if not candidate.get("_embedding") and candidate.get("_embedding_text"): + candidate_embedding = await asyncio.to_thread( + generate_embedding, candidate["_embedding_text"], ) - if emb: - c["_embedding"] = emb + if candidate_embedding: + candidate["_embedding"] = candidate_embedding except Exception: pass engine = SkillSearchEngine() - return engine.search(q, candidates, query_embedding=query_embedding, limit=limit) - + return engine.search(normalized_query, candidates, query_embedding=query_embedding, limit=limit) diff --git a/openspace/mcp_server.py b/openspace/mcp_server.py index e81ef12..b010f2f 100644 --- a/openspace/mcp_server.py +++ b/openspace/mcp_server.py @@ -368,63 +368,48 @@ async def _cloud_search_and_import(task: str, limit: int = 8) -> List[Dict[str, """Search cloud for skills relevant to *task* and auto-import top hits. This is **stage 1** of a two-stage pipeline: - Stage 1 (here): cloud BM25+embedding → pick top-N to import locally. + Stage 1 (here): server-side embedding search → pick top-N to import locally. Stage 2 (tool_layer): local BM25 + LLM → select from ALL local skills (including ones just imported) for injection. Stage 1 intentionally imports more than will be used (default: 8) so - that stage 2 has a larger pool to choose from. The two BM25 passes - are NOT redundant — stage 1 filters thousands of cloud candidates down + that stage 2 has a larger pool to choose from. Stage 1 relies on the + server's embedding search to filter thousands of cloud candidates down to a manageable import set; stage 2 makes the final task-specific choice. """ try: - from openspace.cloud.search import ( - SkillSearchEngine, build_cloud_candidates, - ) - from openspace.cloud.embedding import generate_embedding, resolve_embedding_api - - client = _get_cloud_client() - embedding_api_key, _ = resolve_embedding_api() - has_embedding = bool(embedding_api_key) - - items = await asyncio.to_thread( - client.fetch_metadata, include_embedding=has_embedding, limit=200, - ) - if not items: + normalized_task_query = task.strip() + if not normalized_task_query: return [] - candidates = build_cloud_candidates(items) - if not candidates: + cloud_client = _get_cloud_client() + cloud_search_results = await asyncio.to_thread( + cloud_client.search_record_embeddings, + query=normalized_task_query, + limit=min(limit * 2, 300), + ) + if not cloud_search_results: return [] - query_embedding: Optional[List[float]] = None - if has_embedding: - query_embedding = await asyncio.to_thread( - generate_embedding, task, - ) - - engine = SkillSearchEngine() - results = engine.search(task, candidates, query_embedding=query_embedding, limit=limit * 2) - - cloud_hits = [ - r for r in results - if r.get("source") == "cloud" - and r.get("visibility", "public") == "public" - and r.get("skill_id") + public_cloud_hits = [ + cloud_result for cloud_result in cloud_search_results + if cloud_result.get("visibility", "public") == "public" + and cloud_result.get("record_id") ][:limit] import_results: List[Dict[str, Any]] = [] - for hit in cloud_hits: + for cloud_hit in public_cloud_hits: try: - imp = await _do_import_cloud_skill(skill_id=hit["skill_id"]) + skill_id = cloud_hit["record_id"] + imp = await _do_import_cloud_skill(skill_id=skill_id) import_results.append({ - "skill_id": hit["skill_id"], - "name": hit.get("name", ""), + "skill_id": skill_id, + "name": cloud_hit.get("name", ""), "import_status": imp.get("status", "error"), "local_path": imp.get("local_path", ""), }) except Exception as e: - logger.warning(f"Cloud import failed for {hit['skill_id']}: {e}") + logger.warning(f"Cloud import failed for {skill_id}: {e}") if import_results: logger.info(f"Cloud search imported {len(import_results)} skill(s)")