aios/active-runtime/app/aios_capability_contract.py

"""Canonical fail-closed capability semantics for task eligibility."""

from __future__ import annotations

from collections.abc import Iterable


_TASK_ALIASES: dict[str, frozenset[str]] = {
    "repository_read": frozenset({"repo.read"}),
    "git.read": frozenset({"repo.read"}),
    "repo.read": frozenset({"repo.read"}),
    "test_execution": frozenset({"test.execute"}),
    "git.test": frozenset({"test.execute"}),
    "test.execute": frozenset({"test.execute"}),
    "governed_git_write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
    "governed.git.write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
    "governed.write.branch": frozenset({"governed.git.write.branch"}),
    "governed.write.file": frozenset({"governed.git.write.file"}),
    "shell": frozenset({"runtime.shell"}),
    "git": frozenset({"runtime.git"}),
    "python": frozenset({"runtime.python"}),
    "code.read": frozenset({"legacy.code.read"}),
    "code.review": frozenset({"legacy.code.review"}),
    "code.write": frozenset({"legacy.code.write"}),
    "code.write.patch": frozenset({"legacy.code.write.patch"}),
    "test.python": frozenset({"legacy.test.python"}),
    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
    "git.commit": frozenset({"legacy.git.commit"}),
    "git.diff.read": frozenset({"legacy.git.diff.read"}),
    "git.force_push": frozenset({"legacy.git.force_push"}),
    "deploy.production": frozenset({"legacy.deploy.production"}),
    "secret.transform": frozenset({"legacy.secret.transform"}),
    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
    "review.security": frozenset({"legacy.review.security"}),
    "review.api-contract": frozenset({"legacy.review.api-contract"}),
    "gpu.train": frozenset({"legacy.gpu.train"}),
    "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
    "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
}

_WORKER_ALIASES: dict[str, frozenset[str]] = {
    "git.read": frozenset({"repo.read"}),
    "repo.read": frozenset({"repo.read"}),
    "repository_read": frozenset({"repo.read"}),
    "git.test": frozenset({"test.execute"}),
    "test.execute": frozenset({"test.execute"}),
    "test_execution": frozenset({"test.execute"}),
    "governed.write.branch": frozenset({"governed.git.write.branch"}),
    "governed.git.write.branch": frozenset({"governed.git.write.branch"}),
    "governed.write.file": frozenset({"governed.git.write.file"}),
    "governed.git.write.file": frozenset({"governed.git.write.file"}),
    "shell": frozenset({"runtime.shell"}),
    "git": frozenset({"runtime.git"}),
    "python": frozenset({"runtime.python"}),
    "code.read": frozenset({"legacy.code.read"}),
    "code.review": frozenset({"legacy.code.review"}),
    "code.write": frozenset({"legacy.code.write"}),
    "code.write.patch": frozenset({"legacy.code.write.patch"}),
    "test.python": frozenset({"legacy.test.python"}),
    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
    "git.commit": frozenset({"legacy.git.commit"}),
    "git.diff.read": frozenset({"legacy.git.diff.read"}),
    "git.force_push": frozenset({"legacy.git.force_push"}),
    "deploy.production": frozenset({"legacy.deploy.production"}),
    "secret.transform": frozenset({"legacy.secret.transform"}),
    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
    "review.security": frozenset({"legacy.review.security"}),
    "review.api-contract": frozenset({"legacy.review.api-contract"}),
    "gpu.train": frozenset({"legacy.gpu.train"}),
    "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
    "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
}

_UNKNOWN_PREFIX = "exact:"


def _clean(value: object) -> str:
    if not isinstance(value, str):
        raise ValueError("capability must be a string")
    text = value.strip().lower()
    if not text:
        raise ValueError("capability must not be empty")
    return text


def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
    """Map task-facing names to one semantic vocabulary.

    Unknown names remain explicit exact requirements. They are never treated
    as wildcards or granted by a failed normalization.
    """

    result: set[str] = set()
    for value in values:
        text = _clean(value)
        if "*" in text:
            result.add(_UNKNOWN_PREFIX + text)
        else:
            result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
    return frozenset(result)


def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
    """Map registered worker names to effective semantic capabilities."""

    result: set[str] = set()
    for value in values:
        text = _clean(value)
        if "*" in text:
            continue
        mapped = _WORKER_ALIASES.get(text)
        if mapped is not None:
            result.update(mapped)
    if {
        "governed.git.write.branch",
        "governed.git.write.file",
    }.issubset(result):
        result.add("governed.git.write")
    return frozenset(result)


def has_capabilities(
    worker_capabilities: Iterable[object],
    required_capabilities: Iterable[object],
) -> bool:
    """Return the single eligibility decision used by poll and claim."""

    required = normalize_required_capabilities(required_capabilities)
    effective = normalize_worker_capabilities(worker_capabilities)
    if any(value.startswith(_UNKNOWN_PREFIX) for value in required):
        return False
    return required.issubset(effective)


def required_capabilities_subset(
    granted_requirements: Iterable[object],
    required_requirements: Iterable[object],
) -> bool:
    """Check semantic capability containment for delegation constraints."""
    granted = normalize_required_capabilities(granted_requirements)
    required = normalize_required_capabilities(required_requirements)
    if any(value.startswith(_UNKNOWN_PREFIX) for value in granted | required):
        return False
    return required.issubset(granted)


def normalized_capability_matrix(
    worker_capabilities: Iterable[object],
    required_capabilities: Iterable[object],
) -> dict[str, object]:
    """Expose non-secret decision evidence for tests and diagnostics."""

    worker_values = tuple(worker_capabilities)
    required_values = tuple(required_capabilities)
    worker = normalize_worker_capabilities(worker_values)
    required = normalize_required_capabilities(required_values)
    return {
        "worker_effective": sorted(worker),
        "task_required": sorted(required),
        "eligible": has_capabilities(worker_values, required_values),
    }