Restore formal AgentTask capability contract

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),
+    }

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:<token>`, 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.

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