refactor(orchestration): retire last bare AgentRegistry + canonical capability lookup (#6828)

autobot-backend/api/agent.py

@@ -1220,52 +1220,90 @@ async def analyze_development_task(
         await handle_ai_stack_error(e, "Development analysis")
 
 
+async def _ai_stack_available_agents() -> List[dict]:
+    """Return AI Stack agents as capability-info dicts; empty list when unreachable.
+
+    #6828: AI Stack unavailability degrades to the local registry instead of a
+    503 (same precedent as the #10511 PG fallback in api/agent_org.py).
+    """
+    try:
+        ai_client = await get_ai_stack_client()
+        agents_info = await ai_client.list_available_agents()
+    except AIStackError as e:
+        logger.warning("AI Stack unavailable for /agents/available — serving local registry only: %s", e)
+        return []
+
+    agents_with_caps = []
+    for agent in agents_info.get("agents", []):
+        # Issue #281: capabilities from module constant, with fallback for unknown agents
+        agent_info = AGENT_CAPABILITIES.get(
+            agent,
+            {
+                "description": f"AI agent: {agent}",
+                "capabilities": ["general_processing"],
+            },
+        ).copy()  # Copy to avoid modifying the constant
+        agent_info["name"] = agent
+        agent_info["status"] = "available"
+        agents_with_caps.append(agent_info)
+    return agents_with_caps
+
+
+def _local_registry_agents() -> List[dict]:
+    """Agents from the canonical in-process AgentCapabilityRegistry (#6828).
+
+    Exposes the orchestration profiles (capabilities + specializations) through
+    the same response shape as AI Stack agents so /agents/available is the one
+    unified "find an agent that can do X" entry point.
+    """
+    from orchestration.agent_registry import get_default_capability_registry
+
+    agents = []
+    for profile in get_default_capability_registry().get_all().values():
+        capabilities = sorted({c.value for c in profile.capabilities} | set(profile.specializations))
+        agents.append(
+            {
+                "name": profile.agent_id,
+                "description": f"Local orchestration agent ({profile.agent_type})",
+                "capabilities": capabilities,
+                "status": profile.availability_status,
+            }
+        )
+    return agents
+
+
 @router.get("/agents/available", response_model=DataResponse[AgentAvailableData])
 @with_error_handling(
     category=ErrorCategory.SERVER_ERROR,
     operation="list_available_agents",
     error_code_prefix="AGENT",
 )
-async def list_available_agents():
+async def list_available_agents(capability: str | None = None):
     """
-    List all available AI Stack agents with their capabilities.
+    List all available agents (AI Stack + local orchestration registry).
 
     Issue #281: Refactored to use module-level AGENT_CAPABILITIES constant.
     Issue #987: Removed auth requirement — read-only status, accessible to SLM admin.
+    Issue #6828: unified entry point — merges AI Stack agents with the canonical
+    AgentCapabilityRegistry profiles; optional ``capability`` query filters to
+    agents that can do X (case-insensitive substring on capability names).
     """
-    try:
-        ai_client = await get_ai_stack_client()
-        agents_info = await ai_client.list_available_agents()
+    agents_with_caps = await _ai_stack_available_agents()
+    known = {a["name"] for a in agents_with_caps}
+    agents_with_caps.extend(a for a in _local_registry_agents() if a["name"] not in known)
 
-        # Issue #281: Use module-level constant for agent capabilities
-        # Combine available agents with capability info
-        available_list = agents_info.get("agents", [])
-        agents_with_caps = []
-
-        for agent in available_list:
-            # Get capabilities from module constant, with fallback for unknown agents
-            agent_info = AGENT_CAPABILITIES.get(
-                agent,
-                {
-                    "description": f"AI agent: {agent}",
-                    "capabilities": ["general_processing"],
-                },
-            ).copy()  # Copy to avoid modifying the constant
-            agent_info["name"] = agent
-            agent_info["status"] = "available"
-            agents_with_caps.append(agent_info)
+    if capability:
+        needle = capability.lower()
+        agents_with_caps = [a for a in agents_with_caps if any(needle in c.lower() for c in a["capabilities"])]
 
-        return create_success_response(
-            {
-                "total_agents": len(agents_with_caps),
-                "agents": agents_with_caps,
-                "coordination_modes": ["parallel", "sequential", "intelligent"],
-                "multi_agent_support": True,
-            }
-        )
-
-    except AIStackError as e:
-        await handle_ai_stack_error(e, "List available agents")
+    return create_success_response(
+        {
+            "total_agents": len(agents_with_caps),
+            "agents": agents_with_caps,
+            "coordination_modes": ["parallel", "sequential", "intelligent"],
+            "multi_agent_support": True,
+        }
+    )
 
 
 @router.get("/agents/status", response_model=DataResponse[AgentStatusData])

autobot-backend/api/agent_available_unified_test.py

@@ -0,0 +1,82 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Tests for the unified GET /agent/agents/available entry point (#6828).
+
+The endpoint must merge AI Stack agents with the canonical local
+AgentCapabilityRegistry, support a ``capability`` filter ("find an agent that
+can do X"), and degrade to local-only instead of 503 when the AI Stack is
+unreachable (#10511 precedent).
+"""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+import api.agent as agent_api
+from services.ai_stack_client import AIStackError
+
+
+def _stub_ai_client(agents: list[str]) -> AsyncMock:
+    client = AsyncMock()
+    client.list_available_agents = AsyncMock(return_value={"agents": agents})
+    return client
+
+
+def _agents_from(result) -> dict:
+    # create_success_response returns a DataResponse pydantic model.
+    return {a["name"]: a for a in result.data["agents"]}
+
+
+async def _call(monkeypatch, ai_client, capability=None):
+    async def _get_client():
+        return ai_client
+
+    monkeypatch.setattr(agent_api, "get_ai_stack_client", _get_client)
+    return await agent_api.list_available_agents(capability=capability)
+
+
+async def test_merges_ai_stack_and_local_registry(monkeypatch) -> None:
+    result = await _call(monkeypatch, _stub_ai_client(["rag"]))
+    agents = _agents_from(result)
+
+    assert "rag" in agents  # AI Stack source preserved
+    assert "research_agent" in agents  # local canonical registry merged in
+    assert agents["rag"]["description"].startswith("Retrieval-Augmented")
+
+
+async def test_ai_stack_entry_wins_name_collisions(monkeypatch) -> None:
+    result = await _call(monkeypatch, _stub_ai_client(["research_agent"]))
+    agents = _agents_from(result)
+
+    # One entry only, and it is the AI Stack one (fallback description).
+    assert agents["research_agent"]["description"] == "AI agent: research_agent"
+
+
+async def test_capability_filter_finds_agent_that_can_do_x(monkeypatch) -> None:
+    result = await _call(monkeypatch, _stub_ai_client(["rag"]), capability="web_search")
+    agents = _agents_from(result)
+
+    assert "research_agent" in agents  # specialization: web_search
+    assert "rag" not in agents  # no web_search capability
+
+
+async def test_degrades_to_local_when_ai_stack_unavailable(monkeypatch) -> None:
+    failing = AsyncMock()
+    failing.list_available_agents = AsyncMock(side_effect=AIStackError("down"))
+
+    result = await _call(monkeypatch, failing)
+    agents = _agents_from(result)
+
+    assert "research_agent" in agents
+    assert "system_agent" in agents
+
+
+async def test_total_agents_matches_list_length(monkeypatch) -> None:
+    result = await _call(monkeypatch, _stub_ai_client(["rag", "chat"]))
+    assert result.data["total_agents"] == len(result.data["agents"])
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-v"]))

autobot-backend/api/agent_org.py

@@ -54,9 +54,9 @@ def _fallback_agents_from_registry() -> List[Dict[str, Any]]:
     caller can treat both sources uniformly.
     """
     try:
-        from orchestration.agent_registry import AgentCapabilityRegistry
+        from orchestration.agent_registry import get_default_capability_registry
 
-        registry = AgentCapabilityRegistry(initialize_defaults=True)
+        registry = get_default_capability_registry()  # canonical accessor (#6828)
         agents = []
         for profile in registry.get_all().values():
             agents.append(

autobot-backend/orchestration/agent_registry.py

@@ -214,7 +214,10 @@ class AgentCapabilityRegistry:
     Scope (#6828): holds in-memory AgentProfile + AgentCapability catalogue
     populated at orchestrator startup from DEFAULT_AGENT_CONFIGS.  This is
     the **what-can-each-agent-do** registry — it does not track live health or
-    database persistence.  See also:
+    database persistence.  It is the canonical implementer of the shared
+    ``AgentCapabilityLookup`` ("find an agent that can do X") and
+    ``AgentRegistryProtocol`` (specialization updates) protocols in
+    ``autobot_shared.agent_registry_protocol``.  See also:
     - agents.agent_client.AgentHealthRegistry — health-tracking runtime registry
     - services.agent_registry_service.AgentRegistryService — DB-backed CRUD
     - agents.agent_orchestration.distributed_management.DistributedAgentManager — dynamic/distributed
@@ -393,6 +396,29 @@ def release(self, agent_id: str) -> bool:
             agent.current_workload -= 1
         return True
 
+    async def update_specializations(
+        self,
+        agent_id: str,
+        top_types: List[str],
+        rates: Dict[str, float],
+    ) -> None:
+        """Persist discovered specializations onto an agent's profile (#6828).
+
+        Concrete implementation of the shared ``AgentRegistryProtocol`` — the
+        callback surface ``AgentEvolutionTracker`` uses to report emergent
+        specializations.  Discovered types are promoted to the front of the
+        profile's ``specializations`` (existing ones retained, deduplicated);
+        per-type success rates are recorded in ``performance_metrics`` under
+        ``specialization:<task_type>`` keys.
+        """
+        agent = self._agents.get(agent_id)
+        if agent is None:
+            logger.warning("update_specializations: unknown agent %s — ignored", agent_id)
+            return
+        agent.specializations = list(top_types) + [s for s in agent.specializations if s not in top_types]
+        agent.performance_metrics.update({f"specialization:{task_type}": rate for task_type, rate in rates.items()})
+        logger.info("Agent %s specializations updated: %s", agent_id, top_types)
+
     def update_performance(
         self,
         agent_id: str,
@@ -439,6 +465,20 @@ def __contains__(self, agent_id: str) -> bool:
 _default_registry: "AgentCapabilityRegistry | None" = None
 
 
+def get_default_capability_registry() -> AgentCapabilityRegistry:
+    """Return the process-wide default AgentCapabilityRegistry (#6828).
+
+    The canonical read entry point for "find an agent that can do X" against
+    the default profiles — API fallbacks (api/agent_org, api/agent) and the
+    tool-dispatch boundary (``resolve_forbidden_tools``) share this instance
+    instead of constructing ad-hoc registries per call.
+    """
+    global _default_registry  # noqa: PLW0603
+    if _default_registry is None:
+        _default_registry = AgentCapabilityRegistry(initialize_defaults=True)
+    return _default_registry
+
+
 def resolve_forbidden_tools(agent_id: "str | None") -> "frozenset[str]":
     """Resolve *agent_id*'s ``forbidden_work`` manifest from the default profiles.
 
@@ -448,7 +488,4 @@ def resolve_forbidden_tools(agent_id: "str | None") -> "frozenset[str]":
     """
     if not agent_id:
         return frozenset()
-    global _default_registry  # noqa: PLW0603
-    if _default_registry is None:
-        _default_registry = AgentCapabilityRegistry(initialize_defaults=True)
-    return _default_registry.forbidden_tools(agent_id)
+    return get_default_capability_registry().forbidden_tools(agent_id)

autobot-backend/services/mesh_brain/agent_evolution.py

@@ -37,9 +37,14 @@ async def get_agent_specializations(self, agent_id: str, min_tasks: int, days: i
     async def get_all_agent_ids(self) -> list[str]: ...
 
 
-# Issue #6828: AgentRegistry Protocol promoted to autobot_shared/agent_registry_protocol.py.
-# Re-exported here for backwards compatibility with any importer of this module.
-AgentRegistry = AgentRegistryProtocol
+# Issue #6828: the registry Protocol lives in autobot_shared/agent_registry_protocol.py
+# (AgentRegistryProtocol); its canonical concrete implementer is
+# orchestration.agent_registry.AgentCapabilityRegistry.update_specializations.
+# The legacy `AgentRegistry = AgentRegistryProtocol` compat alias was retired after
+# a repo-wide importer audit found zero consumers — the bare name was the last
+# remnant of the 4-way AgentRegistry naming collision this issue resolved.
+# NOTE (wire-in): AgentEvolutionTracker itself has no production scheduler/caller
+# yet — parked pending a wire-in issue (see #6828 closure report).
 
 
 class AgentEvolutionTracker: