Files
Cognitive_Agent/src/cognitive_agent/application/capability_registry.py
T

138 lines
5.1 KiB
Python
Executable File

"""Load model-facing capability context from domain and native-CAN catalogs."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import yaml
@dataclass(frozen=True)
class CapabilityDefinition:
"""One capability together with its authoritative catalog provenance."""
name: str
domain: str
availability: str
catalog_id: str
catalog_path: Path
executor_binding: str
spec: dict[str, Any]
@property
def target_devices(self) -> tuple[str, ...]:
return tuple(str(value) for value in self.spec.get("target_devices", ()))
@property
def target_scopes(self) -> tuple[str, ...]:
return tuple(str(value) for value in self.spec.get("target_scopes", ()))
@property
def parameters(self) -> tuple[dict[str, Any], ...]:
return tuple(dict(value) for value in self.spec.get("parameters", ()))
class CapabilityRegistry:
"""Expose domain catalogs without embedding domain semantics in prompts or code."""
def __init__(self, repository_root: Path, executable_domains: set[str]) -> None:
self.root = repository_root
self.executable_domains = set(executable_domains)
self._definitions: list[CapabilityDefinition] = []
self._snapshot = self._load()
def snapshot(self) -> dict[str, Any]:
return self._snapshot
def definitions(self, *, domain: str | None = None) -> tuple[CapabilityDefinition, ...]:
"""Return catalog entries without inventing aliases or execution support."""
return tuple(
definition
for definition in self._definitions
if domain is None or definition.domain == domain
)
def find(self, name: str) -> tuple[CapabilityDefinition, ...]:
"""Return every authoritative definition for a canonical capability id."""
return tuple(
definition for definition in self._definitions if definition.name == name
)
def _load(self) -> dict[str, Any]:
domains = []
catalog_paths = [
*(self.root / "domains").glob("*/config/capabilities.yaml"),
*(self.root / "config" / "native_can_capabilities").glob("*.yaml"),
]
for catalog_path in sorted(catalog_paths):
if catalog_path.parts[-3] == "_template":
continue
catalog = yaml.safe_load(catalog_path.read_text(encoding="utf-8"))
domain = catalog["spec"]["domain"]
native_can_catalog = catalog_path.parent.name == "native_can_capabilities"
availability = (
"native_can"
if native_can_catalog
else "executable"
if domain in self.executable_domains
else "catalog_only"
)
executor_binding = (
"NativeCanRuntime/NativeCanExecutor"
if native_can_catalog
else "SeatCompiler/SeatExecutor"
if domain == "seat"
else f"ConfiguredDomainRuntime[{domain}]"
)
catalog_id = str(catalog.get("metadata", {}).get("id", catalog_path.stem))
self._definitions.extend(
CapabilityDefinition(
name=str(capability["name"]),
domain=str(domain),
availability=availability,
catalog_id=catalog_id,
catalog_path=catalog_path,
executor_binding=executor_binding,
spec=dict(capability),
)
for capability in catalog["spec"]["capabilities"]
)
domains.append(
{
"domain": domain,
"availability": availability,
"capabilities": catalog["spec"]["capabilities"],
}
)
return {
"interaction_context": {
"surface": "driver_console",
"speaker_default_scope": "driver",
},
"execution_policy": {
"executable_domains": sorted(self.executable_domains),
"catalog_only_behavior": "return_as_deferred_request",
"native_can_behavior": (
"compile_for_body_can_when_hardware_mode_is_authorized"
),
"planning_environment": "dry_run_candidate_synthesis",
"named_preset_required": False,
"fuzzy_goal_behavior": (
"decompose a commonly understood user goal into executable atomic "
"capabilities and infer conservative absolute targets"
),
"capability_selection": (
"prefer_executable_when_equally_suitable; never replace a clearly better "
"capability solely because it is executable"
),
"executable_request_parameters": (
"resolve relative intent against runtime state and return absolute target values"
),
},
"domains": domains,
}