Memoize Skill Capability Classification by Semantic Input

src/autoskillit/workspace/AGENTS.md

@@ -12,7 +12,7 @@ IL-1 workspace management — clone lifecycle, worktrees, skill resolution.
 | `_clone_detect.py` | `detect_*` helpers + `RUNS_DIR` + `classify_remote_url` |
 | `_clone_remote.py` | `CloneSourceResolution` + probe/isolate remotes |
 | `session_skills.py` | Per-session ephemeral skill dirs; subset filtering |
-| `skill_capabilities.py` | Semantic classification and validation of skill capability evidence |
+| `skill_capabilities.py` | Semantic classification, bounded process-local evidence memoization, and capability validation |
 | `skill_format.py` | SKILL.md frontmatter validation per agentskills.io spec |
 | `skill_projection.py` | Agent-safe projections of typed skill machine contracts |
 | `_projection_cache.py` | Projection asset inventory, cache-key record, and orphan sweep |
@@ -44,3 +44,7 @@ Clone paths live under `RUNS_DIR` (resolved by `_clone_detect.py`). `clone_regis
 coordinates deferred cleanup across concurrent pipeline sessions using file-based locking.
 `session_skills.py` builds per-session ephemeral copies of the bundled skill set so that
 headless sessions can use a filtered subset without polluting the installed package.
+
+`skill_capabilities.py` owns a process-local, weighted LRU keyed by exact canonical
+content and normalized logical skill name. The cache bounds resident entries and
+accounted payload bytes while coordinating concurrent scans outside its lock.

src/autoskillit/workspace/skill_capabilities.py

@@ -7,8 +7,10 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from collections import OrderedDict
+from dataclasses import dataclass, field
 from functools import cache
+from threading import Event, RLock
 from typing import TYPE_CHECKING, Literal
 
 import regex as re
@@ -21,6 +23,15 @@
 CapabilityActor = Literal["self", "parent", "external"]
 CapabilityDirection = Literal["outbound", "inbound", "descriptive"]
 CapabilitySourceClassification = Literal["executable", "artifact"]
+_SkillCapabilityEvidenceKey = tuple[str, str]
+
+# Accounted resident payload includes exact key strings, evidence source strings,
+# and a stable policy charge per immutable evidence record. Entry count bounds
+# the remaining fixed per-entry overhead.
+_SKILL_CAPABILITY_EVIDENCE_RECORD_WEIGHT_BYTES = 192
+_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_ENTRIES = 256
+_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_BYTES = 16 * 1024 * 1024
+_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_INPUT_BYTES = 512 * 1024
 
 
 @dataclass(frozen=True, slots=True)
@@ -67,6 +78,189 @@ def valid(self) -> bool:
         return not self.missing and not self.unsupported
 
 
+@dataclass(frozen=True, slots=True)
+class _SkillCapabilityEvidenceCacheEntry:
+    evidence: tuple[SkillCapabilityEvidence, ...]
+    weight_bytes: int
+
+
+@dataclass(frozen=True, slots=True)
+class _SkillCapabilityEvidenceCacheInfo:
+    max_entries: int
+    max_bytes: int
+    max_input_bytes: int
+    entry_count: int
+    weight_bytes: int
+    inflight_builds: int
+    inflight_waiters: int
+
+
+@dataclass(slots=True)
+class _SkillCapabilityEvidenceBuildState:
+    event: Event = field(default_factory=Event)
+    result: tuple[SkillCapabilityEvidence, ...] | None = None
+    error: BaseException | None = None
+
+
+class _SkillCapabilityEvidenceCache:
+    """Thread-safe weighted LRU with generation-scoped single-flight state."""
+
+    def __init__(
+        self,
+        *,
+        max_entries: int,
+        max_bytes: int,
+        max_input_bytes: int,
+    ) -> None:
+        for field_name, value in (
+            ("max_entries", max_entries),
+            ("max_bytes", max_bytes),
+            ("max_input_bytes", max_input_bytes),
+        ):
+            if value <= 0:
+                raise ValueError(f"{field_name} must be positive")
+
+        self._max_entries = max_entries
+        self._max_bytes = max_bytes
+        self._max_input_bytes = max_input_bytes
+        self._entries: OrderedDict[
+            _SkillCapabilityEvidenceKey,
+            _SkillCapabilityEvidenceCacheEntry,
+        ] = OrderedDict()
+        self._inflight: dict[
+            _SkillCapabilityEvidenceKey,
+            _SkillCapabilityEvidenceBuildState,
+        ] = {}
+        self._weight_bytes = 0
+        self._inflight_waiters = 0
+        self._lock = RLock()
+
+    @property
+    def max_input_bytes(self) -> int:
+        return self._max_input_bytes
+
+    def info(self) -> _SkillCapabilityEvidenceCacheInfo:
+        with self._lock:
+            return _SkillCapabilityEvidenceCacheInfo(
+                max_entries=self._max_entries,
+                max_bytes=self._max_bytes,
+                max_input_bytes=self._max_input_bytes,
+                entry_count=len(self._entries),
+                weight_bytes=self._weight_bytes,
+                inflight_builds=len(self._inflight),
+                inflight_waiters=self._inflight_waiters,
+            )
+
+    def _new_build_state(self) -> _SkillCapabilityEvidenceBuildState:
+        return _SkillCapabilityEvidenceBuildState()
+
+    def _lookup_or_register(
+        self,
+        key: _SkillCapabilityEvidenceKey,
+    ) -> tuple[
+        tuple[SkillCapabilityEvidence, ...] | None,
+        _SkillCapabilityEvidenceBuildState | None,
+        bool,
+    ]:
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is not None:
+                self._entries.move_to_end(key)
+                return entry.evidence, None, False
+
+            state = self._inflight.get(key)
+            if state is not None:
+                self._inflight_waiters += 1
+                return None, state, False
+
+            state = self._new_build_state()
+            self._inflight[key] = state
+            return None, state, True
+
+    def _wait_for_build(
+        self,
+        key: _SkillCapabilityEvidenceKey,
+        state: _SkillCapabilityEvidenceBuildState,
+    ) -> tuple[SkillCapabilityEvidence, ...]:
+        try:
+            state.event.wait()
+        except BaseException:
+            with self._lock:
+                self._inflight_waiters -= 1
+            raise
+
+        with self._lock:
+            self._inflight_waiters -= 1
+            if state.error is not None:
+                raise RuntimeError(
+                    "Capability evidence build failed in another thread"
+                ) from state.error
+            result = state.result
+            if result is None:
+                raise RuntimeError("Capability evidence build completed without a result")
+            entry = self._entries.get(key)
+            if entry is not None and entry.evidence is result:
+                self._entries.move_to_end(key)
+            return result
+
+    def _evict_if_needed_locked(self) -> None:
+        while len(self._entries) > self._max_entries or self._weight_bytes > self._max_bytes:
+            _, entry = self._entries.popitem(last=False)
+            self._weight_bytes -= entry.weight_bytes
+
+    def _publish_failure(
+        self,
+        key: _SkillCapabilityEvidenceKey,
+        state: _SkillCapabilityEvidenceBuildState,
+        error: BaseException,
+    ) -> None:
+        with self._lock:
+            state.result = None
+            state.error = error
+            if self._inflight.get(key) is state:
+                del self._inflight[key]
+            state.event.set()
+
+    def _complete_build(
+        self,
+        key: _SkillCapabilityEvidenceKey,
+        state: _SkillCapabilityEvidenceBuildState,
+        result: tuple[SkillCapabilityEvidence, ...],
+        weight_bytes: int,
+    ) -> tuple[SkillCapabilityEvidence, ...]:
+        with self._lock:
+            resident_mutated = False
+            try:
+                if weight_bytes <= self._max_bytes:
+                    resident_mutated = True
+                    previous = self._entries.pop(key, None)
+                    if previous is not None:
+                        self._weight_bytes -= previous.weight_bytes
+                    self._entries[key] = _SkillCapabilityEvidenceCacheEntry(
+                        evidence=result,
+                        weight_bytes=weight_bytes,
+                    )
+                    self._weight_bytes += weight_bytes
+                    self._evict_if_needed_locked()
+
+                state.result = result
+                state.error = None
+                if self._inflight.get(key) is state:
+                    del self._inflight[key]
+                state.event.set()
+            except BaseException as error:
+                if resident_mutated:
+                    self._entries.clear()
+                    self._weight_bytes = 0
+                state.result = None
+                state.error = error
+                if self._inflight.get(key) is state:
+                    del self._inflight[key]
+                state.event.set()
+                raise
+        return result
+
+
 @dataclass(frozen=True, slots=True)
 class _SourceLine:
     number: int
@@ -186,6 +380,41 @@ def _frontmatter_skill_name(content: str) -> str:
     return match.group(1).strip() if match else ""
 
 
+def _normalize_skill_capability_name(content: str, skill_name: str | None) -> str:
+    return skill_name or _frontmatter_skill_name(content)
+
+
+def _retained_string_weight_bytes(value: str) -> int:
+    return len(value.encode("utf-8", errors="surrogatepass"))
+
+
+def _skill_capability_evidence_input_weight_bytes(
+    content: str,
+    effective_skill_name: str,
+) -> int:
+    return _retained_string_weight_bytes(content) + _retained_string_weight_bytes(
+        effective_skill_name
+    )
+
+
+def _skill_capability_evidence_entry_weight_bytes(
+    input_weight_bytes: int,
+    evidence: tuple[SkillCapabilityEvidence, ...],
+) -> int:
+    return (
+        input_weight_bytes
+        + sum(_retained_string_weight_bytes(item.source) for item in evidence)
+        + len(evidence) * _SKILL_CAPABILITY_EVIDENCE_RECORD_WEIGHT_BYTES
+    )
+
+
+_SKILL_CAPABILITY_EVIDENCE_CACHE = _SkillCapabilityEvidenceCache(
+    max_entries=_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_ENTRIES,
+    max_bytes=_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_BYTES,
+    max_input_bytes=_SKILL_CAPABILITY_EVIDENCE_CACHE_MAX_INPUT_BYTES,
+)
+
+
 def _source_lines(body: str) -> tuple[_SourceLine, ...]:
     """Mark frontmatter, constraint blocks, and documentary fences non-executable."""
     result: list[_SourceLine] = []
@@ -411,16 +640,10 @@ def _is_cross_skill_ref(text: str, skill_name: str) -> bool:
     return "run_skill" in lower and "/autoskillit:" in lower
 
 
-def classify_skill_capability_evidence(
+def _scan_skill_capability_evidence_uncached(
     content: str,
-    skill_name: str | None = None,
+    effective_skill_name: str,
 ) -> tuple[SkillCapabilityEvidence, ...]:
-    """Classify all recognizable capability occurrences in ``content``.
-
-    Documentary occurrences are retained as ``artifact`` evidence so callers
-    can explain why a declaration was rejected without treating it as genuine.
-    """
-    effective_skill_name = skill_name or _frontmatter_skill_name(content)
     lines = _source_lines(content)
     found: list[SkillCapabilityEvidence] = []
     seen: set[tuple[str, tuple[int, int], str]] = set()
@@ -478,6 +701,56 @@ def add(capability: str, source_lines: tuple[_SourceLine, ...]) -> None:
     return tuple(sorted(found, key=lambda item: (item.source_span, item.capability)))
 
 
+def classify_skill_capability_evidence(
+    content: str,
+    skill_name: str | None = None,
+) -> tuple[SkillCapabilityEvidence, ...]:
+    """Classify all recognizable capability occurrences in ``content``.
+
+    Documentary occurrences are retained as ``artifact`` evidence so callers
+    can explain why a declaration was rejected without treating it as genuine.
+    """
+    effective_skill_name = _normalize_skill_capability_name(content, skill_name)
+    evidence_cache = _SKILL_CAPABILITY_EVIDENCE_CACHE
+    scanner = _scan_skill_capability_evidence_uncached
+    if len(content) + len(effective_skill_name) > evidence_cache.max_input_bytes:
+        return scanner(content, effective_skill_name)
+
+    input_weight_bytes = _skill_capability_evidence_input_weight_bytes(
+        content,
+        effective_skill_name,
+    )
+    if input_weight_bytes > evidence_cache.max_input_bytes:
+        return scanner(content, effective_skill_name)
+
+    hash(content)
+    hash(effective_skill_name)
+    key = (content, effective_skill_name)
+    resident, state, is_builder = evidence_cache._lookup_or_register(key)
+    if resident is not None:
+        return resident
+    if state is None:
+        raise RuntimeError("Capability evidence cache returned no build state")
+    if not is_builder:
+        return evidence_cache._wait_for_build(key, state)
+
+    try:
+        result = scanner(content, effective_skill_name)
+        completed_weight_bytes = _skill_capability_evidence_entry_weight_bytes(
+            input_weight_bytes,
+            result,
+        )
+    except BaseException as error:
+        evidence_cache._publish_failure(key, state, error)
+        raise
+    return evidence_cache._complete_build(
+        key,
+        state,
+        result,
+        completed_weight_bytes,
+    )
+
+
 def detect_skill_capabilities(
     content: str,
     skill_name: str | None = None,

tests/_test_filter.py

@@ -723,6 +723,7 @@ class ImportContext(enum.StrEnum):
             "recipe/test_recipe_backend_composition_matrix.py",
             "recipe/test_recipe_composition_vacuous_gate.py",
             "recipe/test_rules_backend_compat.py",
+            "recipe/test_skill_capability_cache_integration.py",
             "recipe/test_rules_skill_content.py",
             "recipe/test_rules_stamp_ownership.py",
             # recipe transitive entries (exercise workspace via deferred imports in

tests/arch/test_layer_enforcement.py

@@ -1664,6 +1664,9 @@ def test_default_classes_only_instantiated_inside_factory_or_allowlist() -> None
     # recipe tests — recipe layer is IL-2 and may use workspace (IL-1 sibling) or config (IL-1)
     "tests/recipe/test_rules_inputs.py": frozenset({"autoskillit.config"}),
     "tests/recipe/test_contracts.py": frozenset({"autoskillit.workspace"}),
+    "tests/recipe/test_skill_capability_cache_integration.py": frozenset(
+        {"autoskillit.workspace"}
+    ),
     "tests/recipe/test_rules_skill_content.py": frozenset({"autoskillit.workspace"}),
     "tests/recipe/test_rules_backend_compat.py": frozenset(
         {"autoskillit.server", "autoskillit.workspace"}

tests/arch/test_subpackage_isolation.py

@@ -106,6 +106,7 @@ def _get_call_func_name(node: ast.Call) -> str | None:
         "_step_context",  # core/_step_context.py: current_step_name, current_order_id ContextVars
         "_api_cache",  # recipe/_api_cache.py: _LOAD_CACHE = LoadCache()
         "_contracts_manifest",  # recipe/_contracts_manifest.py: _MANIFEST_CACHE = YamlFileCache()
+        "skill_capabilities",  # workspace/skill_capabilities.py: bounded evidence cache
         "methodology_venue_appendix",  # recipe/methodology_venue_appendix.py: _ML_SUB_AREA_CACHE
         "rules_blocks",  # recipe/rules/rules_blocks.py: _BUDGETS_CACHE = YamlFileCache()
         "rules_phoropter_adjacency",  # recipe/rules/rules_phoropter_adjacency.py: _PREFIXES_CACHE