fix: keep connector tools available to Octo

src/octopal/runtime/octo/prompts/octo_system.md

@@ -92,6 +92,16 @@ Octopal skills are internal tools, not MCP servers.
 - Do not prefer exec_run for skill bundle scripts when run_skill_script is available.
 - A skill can be available even if there is no MCP entry for it.
 
+## Tool Catalog
+
+The visible tool list may be a curated active subset of the full catalog.
+
+- If the tools you can currently see do not cover the task, call tool_catalog_search before saying the capability is unavailable.
+- Use tool_catalog_search to search the full tool catalog by task intent, tool family, category, or capability.
+- Treat tool_catalog_search as a discovery step, not the final action.
+- After tool_catalog_search, the system may activate matching tools for the current turn. If that happens, call the newly activated tool directly on the next step.
+- Do not tell the user a tool is missing or inaccessible until you have checked the catalog when appropriate.
+
 ## Canonical Memory Management
 
 You are responsible for maintaining the long-term knowledge base in `memory/canon/`.

src/octopal/runtime/octo/router.py

@@ -43,6 +43,7 @@
 _MAX_VERIFY_CONTEXT_CHARS = 20000
 _DEFAULT_MAX_TOOL_COUNT = 64
 _MIN_TOOL_COUNT_ON_OVERFLOW = 12
+_CATALOG_TOOL_EXPANSION_LIMIT = 12
 _MANDATORY_OCTO_TOOL_NAMES = {
     "fs_list",
     "fs_read",
@@ -53,6 +54,7 @@
 _PRIORITY_TOOL_NAMES = {
     "octo_context_reset",
     "octo_context_health",
+    "tool_catalog_search",
     "octo_experiment_log",
     "check_schedule",
     "start_worker",
@@ -69,6 +71,7 @@
     "octo_context_health",
     "check_schedule",
     "scheduler_status",
+    "tool_catalog_search",
     # Scheduler control loop
     "list_schedule",
     "schedule_task",
@@ -392,6 +395,25 @@ async def route_or_reply(
 
                     for call in tool_calls:
                         tool_result, tool_meta = await _handle_octo_tool_call(call, active_tool_specs, ctx)
+                        expanded_names: list[str] = []
+                        if str(call.get("function", {}).get("name") or "") == "tool_catalog_search":
+                            active_tool_specs, expanded_names = _expand_active_tool_specs_from_catalog_result(
+                                tool_result,
+                                active_tool_specs=active_tool_specs,
+                                ctx=ctx,
+                            )
+                            tools = [spec.to_openai_tool() for spec in active_tool_specs]
+                            if expanded_names:
+                                messages.append(
+                                    Message(
+                                        role="system",
+                                        content=(
+                                            "Tool catalog expansion complete. The following tools are now active for this turn:\n"
+                                            + "\n".join(f"- {name}" for name in expanded_names)
+                                            + "\nUse them directly if they fit the task."
+                                        ),
+                                    )
+                                )
                         tool_result_text = render_tool_result_for_llm(tool_result).text
                         loop_state = _record_octo_tool_call(
                             tool_call_history,
@@ -746,6 +768,7 @@ def _get_octo_tools(octo: Any, chat_id: int) -> tuple[list[ToolSpec], dict[str,
     )
     max_tools = _env_int("OCTOPAL_OCTO_MAX_TOOL_COUNT", _DEFAULT_MAX_TOOL_COUNT, minimum=8)
     tool_specs = _budget_tool_specs(tool_specs, max_count=max_tools)
+    ctx["active_tool_specs"] = tool_specs
     ctx["tool_resolution_report"] = resolution_report
     ctx["all_tool_specs"] = all_tools
     return tool_specs, ctx
@@ -825,7 +848,21 @@ def _is_transient_provider_error(exc: Exception) -> bool:
 
 def _tool_priority(spec: ToolSpec) -> tuple[int, str]:
     name = str(getattr(spec, "name", "") or "")
-    return (0 if name in _PRIORITY_TOOL_NAMES else 1, name)
+    if name in _PRIORITY_TOOL_NAMES:
+        return (0, name)
+    if _is_connector_tool(spec):
+        return (1, name)
+    return (2, name)
+
+
+def _is_connector_tool(spec: ToolSpec) -> bool:
+    metadata = getattr(spec, "metadata", None)
+    category = str(getattr(metadata, "category", "") or "").strip().lower()
+    if category == "connectors":
+        return True
+
+    name = str(getattr(spec, "name", "") or "").strip().lower()
+    return name.startswith(("gmail_", "calendar_", "drive_", "connector_"))
 
 
 def _budget_tool_specs(tool_specs: list[ToolSpec], *, max_count: int) -> list[ToolSpec]:
@@ -858,6 +895,52 @@ def _shrink_tool_specs_for_retry(tool_specs: list[ToolSpec]) -> list[ToolSpec]:
     return _budget_tool_specs(tool_specs, max_count=reduced)
 
 
+def _expand_active_tool_specs_from_catalog_result(
+    tool_result: Any,
+    *,
+    active_tool_specs: list[ToolSpec],
+    ctx: dict[str, object],
+) -> tuple[list[ToolSpec], list[str]]:
+    payload = tool_result if isinstance(tool_result, dict) else {}
+    if isinstance(tool_result, str):
+        try:
+            parsed = json.loads(tool_result)
+        except Exception:
+            parsed = None
+        if isinstance(parsed, dict):
+            payload = parsed
+    results = payload.get("results") if isinstance(payload, dict) else None
+    if not isinstance(results, list) or not results:
+        return active_tool_specs, []
+
+    all_specs = list(ctx.get("all_tool_specs") or [])
+    by_name = {str(getattr(spec, "name", "") or ""): spec for spec in all_specs}
+    selected = list(active_tool_specs)
+    selected_names = {str(getattr(spec, "name", "") or "") for spec in selected}
+
+    expanded_names: list[str] = []
+    for item in results:
+        if len(expanded_names) >= _CATALOG_TOOL_EXPANSION_LIMIT:
+            break
+        if not isinstance(item, dict):
+            continue
+        if bool(item.get("active_now")):
+            continue
+        name = str(item.get("name", "") or "").strip()
+        if not name or name in selected_names:
+            continue
+        spec = by_name.get(name)
+        if spec is None:
+            continue
+        selected.append(spec)
+        selected_names.add(name)
+        expanded_names.append(name)
+
+    if expanded_names:
+        ctx["active_tool_specs"] = selected
+    return selected, expanded_names
+
+
 async def _handle_octo_tool_call(
     call: dict,
     tools: list[ToolSpec],

src/octopal/tools/catalog.py

@@ -53,6 +53,147 @@
 logger = structlog.get_logger(__name__)
 
 
+def _tool_catalog_search(args, ctx) -> str:
+    query = str((args or {}).get("query", "") or "").strip().lower()
+    category_filter = str((args or {}).get("category", "") or "").strip().lower()
+    capability_filter = str((args or {}).get("capability", "") or "").strip().lower()
+    limit = max(1, min(int((args or {}).get("limit", 12) or 12), 50))
+
+    report = ctx.get("tool_resolution_report")
+    if report is not None and hasattr(report, "available_tools"):
+        candidates = list(report.available_tools)
+    else:
+        candidates = list(ctx.get("all_tool_specs") or [])
+
+    active_names = {
+        str(getattr(spec, "name", "") or "").strip().lower()
+        for spec in (ctx.get("active_tool_specs") or [])
+    }
+
+    scored: list[tuple[int, ToolSpec]] = []
+    for spec in candidates:
+        if str(spec.name).strip().lower() == "tool_catalog_search":
+            continue
+
+        metadata = getattr(spec, "metadata", None)
+        category = str(getattr(metadata, "category", "") or "").strip().lower()
+        capabilities = tuple(getattr(metadata, "capabilities", ()) or ())
+        profile_tags = tuple(getattr(metadata, "profile_tags", ()) or ())
+        if category_filter and category != category_filter:
+            continue
+        if capability_filter and capability_filter not in capabilities:
+            continue
+
+        score = _tool_catalog_search_score(
+            spec,
+            query=query,
+            category=category,
+            capabilities=capabilities,
+            profile_tags=profile_tags,
+        )
+        if query and score <= 0:
+            continue
+        scored.append((score, spec))
+
+    scored.sort(
+        key=lambda item: (
+            -item[0],
+            0 if str(getattr(item[1], "name", "") or "").strip().lower() not in active_names else 1,
+            str(getattr(item[1], "name", "") or ""),
+        )
+    )
+
+    items = []
+    for score, spec in scored[:limit]:
+        metadata = getattr(spec, "metadata", None)
+        params = spec.parameters if isinstance(spec.parameters, dict) else {}
+        properties = params.get("properties") if isinstance(params, dict) else {}
+        required = params.get("required") if isinstance(params, dict) else []
+        if not isinstance(properties, dict):
+            properties = {}
+        if not isinstance(required, list):
+            required = []
+
+        items.append(
+            {
+                "name": spec.name,
+                "description": spec.description,
+                "category": str(getattr(metadata, "category", "") or "misc"),
+                "risk": str(getattr(metadata, "risk", "") or "safe"),
+                "capabilities": list(getattr(metadata, "capabilities", ()) or ()),
+                "profile_tags": list(getattr(metadata, "profile_tags", ()) or ()),
+                "required_arguments": [str(item) for item in required if str(item).strip()],
+                "argument_names": sorted(str(key) for key in properties.keys()),
+                "active_now": str(spec.name).strip().lower() in active_names,
+                "score": score,
+            }
+        )
+
+    return json.dumps(
+        {
+            "status": "ok",
+            "query": query or None,
+            "category": category_filter or None,
+            "capability": capability_filter or None,
+            "count": len(items),
+            "results": items,
+            "hint": (
+                "If the needed tool is not active right now, use this catalog result to decide what tool family you need next."
+            ),
+        },
+        ensure_ascii=False,
+    )
+
+
+def _tool_catalog_search_score(
+    spec: ToolSpec,
+    *,
+    query: str,
+    category: str,
+    capabilities: tuple[str, ...],
+    profile_tags: tuple[str, ...],
+) -> int:
+    if not query:
+        return 1
+
+    name = str(getattr(spec, "name", "") or "").strip().lower()
+    description = str(getattr(spec, "description", "") or "").strip().lower()
+    query_terms = [term for term in query.replace("-", "_").split() if term]
+    if not query_terms:
+        query_terms = [query]
+
+    score = 0
+    for term in query_terms:
+        if name == term:
+            score += 120
+        elif name.startswith(term):
+            score += 80
+        elif term in name:
+            score += 55
+
+        if category == term:
+            score += 35
+        elif term and term in category:
+            score += 20
+
+        if term in description:
+            score += 12
+
+        for capability in capabilities:
+            if capability == term:
+                score += 30
+            elif term in capability:
+                score += 16
+
+        for tag in profile_tags:
+            if tag == term:
+                score += 18
+            elif term in tag:
+                score += 10
+
+    return score
+
+
 def get_tools(mcp_manager=None) -> list[ToolSpec]:
     tools = [
         ToolSpec(
@@ -146,6 +287,39 @@ def get_tools(mcp_manager=None) -> list[ToolSpec]:
             handler=_tool_octo_context_health,
             is_async=True,
         ),
+        ToolSpec(
+            name="tool_catalog_search",
+            description=(
+                "Search the full catalog of available Octo tools, including tools that may not be active in the current "
+                "tool budget. Use this when the visible toolset seems insufficient for the task."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Search text such as gmail, calendar, worker, drive, file, or schedule.",
+                    },
+                    "category": {
+                        "type": "string",
+                        "description": "Optional category filter such as connectors, browser, workers, or filesystem.",
+                    },
+                    "capability": {
+                        "type": "string",
+                        "description": "Optional capability filter such as gmail_read or connector_use.",
+                    },
+                    "limit": {
+                        "type": "integer",
+                        "minimum": 1,
+                        "maximum": 50,
+                        "description": "Maximum number of matches to return.",
+                    },
+                },
+                "additionalProperties": False,
+            },
+            permission="self_control",
+            handler=_tool_catalog_search,
+        ),
         ToolSpec(
             name="octo_opportunity_scan",
             description="Generate proactive opportunity cards (impact/effort/confidence/next_action) for the active chat.",

tests/test_connector_tools.py

@@ -5,11 +5,13 @@
 
 from octopal.infrastructure.config.models import ConnectorInstanceConfig, OctopalConfig
 from octopal.infrastructure.connectors.manager import ConnectorManager
+from octopal.runtime.octo.router import _budget_tool_specs
 from octopal.tools.catalog import get_tools
 from octopal.tools.connectors.calendar import get_calendar_connector_tools
 from octopal.tools.connectors.drive import get_drive_connector_tools
 from octopal.tools.connectors.gmail import get_gmail_connector_tools
 from octopal.tools.connectors.status import connector_status_read
+from octopal.tools.registry import ToolSpec
 
 
 def test_catalog_includes_read_only_connector_status_tool() -> None:
@@ -66,6 +68,30 @@ def get_all_tools(self):
     assert "gmail_get_message" in names
 
 
+def test_octo_budget_keeps_connector_alias_tools() -> None:
+    class _Manager:
+        def get_all_tools(self):
+            return [
+                ToolSpec(
+                    name=f"mcp_demo_tool_{index}",
+                    description="demo",
+                    parameters={"type": "object"},
+                    permission="mcp_exec",
+                    handler=lambda _args, _ctx: "ok",
+                    is_async=True,
+                )
+                for index in range(40)
+            ]
+
+    tools = get_tools(mcp_manager=_Manager())
+    active = _budget_tool_specs(tools, max_count=64)
+    names = {tool.name for tool in active}
+
+    assert "gmail_list_messages" in names
+    assert "gmail_search_messages" in names
+    assert "gmail_get_unread_count" in names
+
+
 def test_catalog_includes_first_class_calendar_tools_when_mcp_manager_is_present() -> None:
     class _Manager:
         def get_all_tools(self):