codenib.agent.skills

Declarative skill definitions for CodeNib agents.

Built-in skills are loaded from sub-packages via SkillLoader. User code can still define additional skills with the @skill decorator.

Modules:

Name	Description
bm25_search
code_to_query
codenib_context
context	Typed skill-context layer.
core	Core data structures for skill definitions.
crossencoder_rerank
embedding_search
find_callees
find_callers
hybrid_search
llm_rerank
loader	Skill package loader.
registry	Skill registry and @skill decorator.
trace
typecheck	Type parsing, runtime arg coercion, and skill-dataflow type checking.

Classes:

Name	Description
Cost	Relative cost hint for the policy engine.
SkillInputSpec	Type specification for a single skill input.
SkillMetadata	Pure-data descriptor for a registered skill.
SkillOutputSpec	Type specification for skill output.
SkillType	Execution characteristic of a skill.
SkillLoader	Load skill packages from a directory into the SkillRegistry.
SkillRegistry	Global catalogue of skill metadata.

Functions:

Name	Description
skill	Register a function as a skill.

Cost

Bases: str, Enum

Relative cost hint for the policy engine.

SkillInputSpec dataclass

SkillInputSpec(
    name: str,
    type_hint: str,
    required: bool = True,
    default: Any = None,
    description: str = "",
    enum: list[str] | None = None,
    is_line_number: bool = False,
)

Type specification for a single skill input.

SkillMetadata dataclass

SkillMetadata(
    skill_id: str,
    skill_type: SkillType,
    inputs: list[SkillInputSpec] = list(),
    outputs: SkillOutputSpec = (lambda: SkillOutputSpec(type_hint="Any"))(),
    executor_fn: Callable[..., Any] | None = None,
    async_capable: bool = False,
    cacheable: bool = True,
    cost: Cost = MEDIUM,
    dependencies: list[str] = list(),
    resources: list[str] = list(),
    operator: str | None = None,
    defaults: dict[str, Any] = dict(),
    index_requirements: list[Any] = list(),
    skill_doc: str = "",
    description: str = "",
)

Pure-data descriptor for a registered skill.

The executor_fn is the only non-serializable field — the Rust compiler will never touch it; the Python scheduler calls it at execution time.

Methods:

Name	Description
to_dict	JSON-friendly dict (excludes executor_fn).

to_dict

to_dict() -> dict[str, Any]

JSON-friendly dict (excludes executor_fn).

Source code in codenib/agent/skills/core.py

def to_dict(self) -> Dict[str, Any]:
    """JSON-friendly dict (excludes ``executor_fn``)."""
    return {
        "skill_id": self.skill_id,
        "skill_type": self.skill_type.value,
        "inputs": [
            {
                "name": i.name,
                "type_hint": i.type_hint,
                "required": i.required,
                "default": i.default,
                "description": i.description,
                "is_line_number": i.is_line_number,
            }
            for i in self.inputs
        ],
        "outputs": {
            "type_hint": self.outputs.type_hint,
            "description": self.outputs.description,
        },
        "async_capable": self.async_capable,
        "cacheable": self.cacheable,
        "cost": self.cost.value,
        "dependencies": list(self.dependencies),
        "resources": list(self.resources),
        "operator": self.operator,
        "defaults": dict(self.defaults),
        "index_requirements": [
            r.to_dict() if hasattr(r, "to_dict") else r
            for r in self.index_requirements
        ],
        "skill_doc": self.skill_doc,
        "description": self.description,
    }

SkillOutputSpec dataclass

SkillOutputSpec(type_hint: str, description: str = '')

Type specification for skill output.

SkillType

Bases: str, Enum

Execution characteristic of a skill.

SkillLoader

Load skill packages from a directory into the SkillRegistry.

Each skill package is a subdirectory containing:

• config.yaml — skill metadata and default parameters
• skill.md — agent-readable description (optional)
• executor.py — create_executor(context) -> Callable factory

Methods:

Name	Description
load_all	Scan skills_dir for subdirectories containing config.yaml
load_skill	Load a single skill package from skill_dir.

load_all

load_all(
    skills_dir: str,
    contexts: dict[str, Any] | None = None,
    *,
    registry: SkillRegistry | None = None
) -> list[SkillMetadata]

Scan skills_dir for subdirectories containing config.yaml and load each as a skill package.

Parameters:

Name	Type	Description	Default
skills_dir	str	Path to the skills directory.	required
contexts	dict[str, Any] | None	Mapping of context-type names to context objects (e.g. {"retrieve": RetrieveContext(...)}).	None
registry	SkillRegistry | None	Registry to populate; defaults to the singleton.	None

Returns:

Type	Description
list[SkillMetadata]	List of loaded SkillMetadata objects.

Source code in codenib/agent/skills/loader.py

def load_all(
    self,
    skills_dir: str,
    contexts: Optional[Dict[str, Any]] = None,
    *,
    registry: Optional[SkillRegistry] = None,
) -> List[SkillMetadata]:
    """
    Scan *skills_dir* for subdirectories containing ``config.yaml``
    and load each as a skill package.

    Args:
        skills_dir: Path to the skills directory.
        contexts: Mapping of context-type names to context objects
            (e.g. ``{"retrieve": RetrieveContext(...)}``).
        registry: Registry to populate; defaults to the singleton.

    Returns:
        List of loaded ``SkillMetadata`` objects.
    """
    contexts = contexts or {}
    reg = registry or SkillRegistry()
    loaded: List[SkillMetadata] = []

    skills_path = Path(skills_dir)
    if not skills_path.is_dir():
        return loaded

    for entry in sorted(skills_path.iterdir()):
        if not entry.is_dir():
            continue
        config_file = entry / "config.yaml"
        if not config_file.exists():
            continue

        metadata = self.load_skill(str(entry), contexts)
        if metadata is not None:
            if not reg.has(metadata.skill_id):
                reg.register(metadata)
            loaded.append(metadata)

    return loaded

load_skill

load_skill(
    skill_dir: str, contexts: dict[str, Any] | None = None
) -> SkillMetadata | None

Load a single skill package from skill_dir.

Returns None if the directory lacks a config.yaml.

Source code in codenib/agent/skills/loader.py

def load_skill(
    self,
    skill_dir: str,
    contexts: Optional[Dict[str, Any]] = None,
) -> Optional[SkillMetadata]:
    """
    Load a single skill package from *skill_dir*.

    Returns ``None`` if the directory lacks a ``config.yaml``.
    """
    contexts = contexts or {}
    skill_path = Path(skill_dir)

    config_file = skill_path / "config.yaml"
    if not config_file.exists():
        return None

    # --- 1. Parse config.yaml ---
    with open(config_file) as f:
        config = yaml.safe_load(f) or {}

    skill_id: str = config["skill_id"]
    skill_type = _SKILL_TYPE_MAP.get(
        config.get("skill_type", "custom"), SkillType.CUSTOM
    )
    cost = _COST_MAP.get(config.get("cost", "medium"), Cost.MEDIUM)
    operator = config.get("operator")
    defaults = config.get("defaults", {})
    resources = config.get("resources", [])
    dependencies = config.get("dependencies", [])

    inputs = _parse_inputs(config.get("inputs", []))
    outputs = _parse_outputs(config.get("outputs", {}))
    index_requirements = _parse_index_requirements(
        config.get("index_requirements", [])
    )

    # --- 2. Read skill.md ---
    skill_doc = ""
    skill_md_file = skill_path / "skill.md"
    if skill_md_file.exists():
        skill_doc = skill_md_file.read_text(encoding="utf-8")

    # --- 3. Load executor ---
    executor_fn = _load_executor(skill_path, skill_type, contexts)

    # --- 4. Build SkillMetadata ---
    return SkillMetadata(
        skill_id=skill_id,
        skill_type=skill_type,
        inputs=inputs,
        outputs=outputs,
        executor_fn=executor_fn,
        async_capable=False,
        cacheable=config.get("cacheable", True),
        cost=cost,
        dependencies=dependencies,
        resources=resources,
        operator=operator,
        defaults=defaults,
        index_requirements=index_requirements,
        skill_doc=skill_doc,
        description=skill_doc or config.get("description", ""),
    )

SkillRegistry

Global catalogue of skill metadata.

Implemented as an explicit singleton so that the @skill decorator and the compiler always share the same instance.

Methods:

Name	Description
reset	Tear down the singleton — only for unit tests.

reset classmethod

reset() -> None

Tear down the singleton — only for unit tests.

Source code in codenib/agent/skills/registry.py

@classmethod
def reset(cls) -> None:
    """Tear down the singleton — only for unit tests."""
    if cls._instance is not None:
        cls._instance._skills.clear()
        cls._instance = None

skill

skill(
    skill_id: str,
    skill_type: SkillType,
    inputs: list[SkillInputSpec] | None = None,
    outputs: SkillOutputSpec | None = None,
    *,
    dependencies: list[str] | None = None,
    resources: list[str] | None = None,
    cost: Cost = MEDIUM,
    operator: str | None = None,
    async_capable: bool = False,
    cacheable: bool = True,
    defaults: dict[str, Any] | None = None,
    index_requirements: list[Any] | None = None,
    skill_doc: str = "",
    description: str = ""
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a function as a skill.

Usage::

@skill(
    skill_id="embedding_search",
    skill_type=SkillType.RETRIEVAL,
    inputs=[SkillInputSpec(name="query", type_hint="str")],
    outputs=SkillOutputSpec(type_hint="List[QueriedNode]"),
    operator="faiss.retrieve",
    cost=Cost.HIGH,
)
def embedding_search(query: str, top_k: int = 10) -> List[QueriedNode]:
    ...

The decorator only stores metadata — the function body is the execution logic invoked by the scheduler.

Source code in codenib/agent/skills/registry.py

def skill(
    skill_id: str,
    skill_type: SkillType,
    inputs: Optional[List[SkillInputSpec]] = None,
    outputs: Optional[SkillOutputSpec] = None,
    *,
    dependencies: Optional[List[str]] = None,
    resources: Optional[List[str]] = None,
    cost: Cost = Cost.MEDIUM,
    operator: Optional[str] = None,
    async_capable: bool = False,
    cacheable: bool = True,
    defaults: Optional[Dict[str, Any]] = None,
    index_requirements: Optional[List[Any]] = None,
    skill_doc: str = "",
    description: str = "",
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Register a function as a skill.

    Usage::

        @skill(
            skill_id="embedding_search",
            skill_type=SkillType.RETRIEVAL,
            inputs=[SkillInputSpec(name="query", type_hint="str")],
            outputs=SkillOutputSpec(type_hint="List[QueriedNode]"),
            operator="faiss.retrieve",
            cost=Cost.HIGH,
        )
        def embedding_search(query: str, top_k: int = 10) -> List[QueriedNode]:
            ...

    The decorator only stores metadata — the function body is the execution
    logic invoked by the scheduler.
    """

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        _async = async_capable or inspect.iscoroutinefunction(fn)

        metadata = SkillMetadata(
            skill_id=skill_id,
            skill_type=skill_type,
            inputs=inputs or [],
            outputs=outputs or SkillOutputSpec(type_hint="Any"),
            executor_fn=fn,
            async_capable=_async,
            cacheable=cacheable,
            cost=cost,
            dependencies=dependencies or [],
            resources=resources or [],
            operator=operator,
            defaults=defaults or {},
            index_requirements=index_requirements or [],
            skill_doc=skill_doc,
            description=description or fn.__doc__ or "",
        )

        SkillRegistry().register(metadata)

        # Attach metadata to the function for introspection.
        fn._skill_metadata = metadata  # type: ignore[attr-defined]
        return fn

    return decorator