diff --git a/active-runtime/app/aios_capability_contract.py b/active-runtime/app/aios_capability_contract.py new file mode 100644 index 0000000..036fab4 --- /dev/null +++ b/active-runtime/app/aios_capability_contract.py @@ -0,0 +1,116 @@ +"""Canonical fail-closed capability semantics for formal AgentTask eligibility.""" + +from __future__ import annotations + +from collections.abc import Iterable + + +# These aliases are the verified C4 contract captured in +# ``qa/capability-contract.txt``. Unknown task names become exact requirements; +# unknown worker names are never granted. +_TASK_ALIASES: dict[str, frozenset[str]] = { + "release:candidate": frozenset({"release:candidate"}), + "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 = { + key: value for key, value in _TASK_ALIASES.items() + if key not in {"release:candidate", "governed_git_write", "governed.git.write"} +} +_WORKER_ALIASES.update({ + "governed.git.write.branch": frozenset({"governed.git.write.branch"}), + "governed.git.write.file": frozenset({"governed.git.write.file"}), +}) +_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]: + 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]: + 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: + 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: + 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]: + 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), + } diff --git a/capability-evidence/C4_SOURCE_PROVENANCE.md b/capability-evidence/C4_SOURCE_PROVENANCE.md new file mode 100644 index 0000000..a616ce9 --- /dev/null +++ b/capability-evidence/C4_SOURCE_PROVENANCE.md @@ -0,0 +1,14 @@ +# C4 capability contract provenance + +The C4 checkout has no `aios_capability_contract.py`, although +`active-runtime/app/aios_agent_task_protocol.py` imports it at module scope. +The read-only QA capture used for the missing source closure is: + +- source: `C:\\Users\\ImRuY\\.codex-restored\\work\\v2r2c5-release-20260926\\qa\\capability-contract.txt` +- SHA256: `c97c9f8a3cf199a9c0348da1212aa26d161aad8ddaed84db534e46f490f2422f` +- 140 lines + +The contract preserves separate task and worker alias maps, maps unknown task +names to `exact:`, ignores unknown worker names, and returns false for +unknown or mismatched eligibility. It is used by formal AgentTask poll and +claim paths only. Generic `task_plane` matching remains outside this lane. diff --git a/tests/test_v2r2c5_capability_contract.py b/tests/test_v2r2c5_capability_contract.py new file mode 100644 index 0000000..a52aa06 --- /dev/null +++ b/tests/test_v2r2c5_capability_contract.py @@ -0,0 +1,20 @@ +from aios_capability_contract import ( + has_capabilities, + normalize_required_capabilities, + normalize_worker_capabilities, +) + + +def test_n01_unknown_required_token_is_explicit_and_denied(): + assert normalize_required_capabilities(["unknown-token"]) == frozenset({"exact:unknown-token"}) + assert has_capabilities(["testing"], ["unknown-token"]) is False + + +def test_n02_unknown_worker_token_is_never_granted(): + assert "exact:unknown-token" not in normalize_worker_capabilities(["unknown-token"]) + assert has_capabilities(["unknown-token"], ["testing"]) is False + + +def test_n03_admitted_capability_and_claim_match_use_same_contract(): + assert has_capabilities(["git.read"], ["repository_read"]) is True + assert has_capabilities(["git.read"], ["testing"]) is False