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) 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/cloud/client.py b/openspace/cloud/client.py index fb0792f..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. @@ -340,7 +369,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 +434,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 +443,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/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/dashboard_server.py b/openspace/dashboard_server.py index 96d3e50..f6916b6 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. + + 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]: 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/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", diff --git a/openspace/mcp_server.py b/openspace/mcp_server.py index 7e3428f..b010f2f 100644 --- a/openspace/mcp_server.py +++ b/openspace/mcp_server.py @@ -198,6 +198,61 @@ 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. + The registry is rebuilt per call so later local searches can see + newly added skills without requiring a process restart. + """ + 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() + return registry + + def _get_cloud_client(): """Get a OpenSpaceClient instance (raises CloudError if not configured).""" from openspace.cloud.auth import get_openspace_auth @@ -313,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)") @@ -597,7 +637,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/openspace/skill_engine/registry.py b/openspace/skill_engine/registry.py index bb35dd8..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(): @@ -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}") 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 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