feat: Trust Model v2 — 4-state unification (v0.2.0)

CHANGELOG.md

@@ -0,0 +1,42 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+## capacium-models v0.2.0 — Trust Model v2 (2026-05-24)
+
+### Breaking Changes
+
+- **Trust model unified to 4 states**: `TrustState` enum reduced from 6
+  states to 4: `discovered`, `audited`, `verified`, `signed`. Legacy states
+  `indexed` and `claimed` are mapped automatically via `LEGACY_STATE_MAP`.
+
+### New Features
+
+- **TrustMachine state machine**: New `TrustMachine` class with `transition()`
+  method enforcing valid trust state transitions. Supports `promote()` and
+  `demote()` with reason tracking and timestamp history.
+
+- **Legacy state normalization**: `normalize_legacy_state()` function
+  transparently maps `indexed` → `audited` and `claimed` → `verified` for
+  backward compatibility with existing database records.
+
+- **Trust ordering**: `TrustState.ordering()` returns states in promotion
+  order with comparison support (`>=`, `<=`).
+
+### Bug Fixes
+
+- **Search min_trust_state filter** (355cded): Fixed `min_trust_state` filter
+  to normalize legacy states before comparison. Legacy aliases (e.g.
+  `indexed`) are now included in the SQL `IN` clause so database rows with
+  old state values still match correctly.
+
+- **Trust model completeness** (e57b2cd): Fixed incomplete v2 trust model
+  implementation — added missing `LEGACY_STATE_MAP` export and
+  `normalize_legacy_state()` function.
+
+## capacium-models v0.1.0 — Initial Release (2026-05-11)
+
+- `Listing` dataclass with `from_dict()` / `to_dict()` roundtrip
+- `SearchQuery` with faceted filtering (text, type, category, trust, tags)
+- `ExchangeSearch` with dynamic SQL WHERE clause builder
+- `TrustState` enum (original 6-state model)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "capacium-models"
-version = "0.1.0"
+version = "0.2.0"
 description = "Shared domain models for the Capacium ecosystem — Listing, TrustState, TrustMachine, and more"
 authors = [{name = "Capacium"}]
 requires-python = ">=3.10"

src/capacium_models/models.py

@@ -136,6 +136,24 @@ class Listing:
     github_created_at: Optional[str] = None
     github_pushed_at: Optional[str] = None
     github_updated_at: Optional[str] = None
+
+    # -- Document Model v2 --
+    source_repo: Optional[str] = None      # e.g. "anthropics/skills"
+    source_path: str = "/"                  # e.g. "skills/ai-ml/3d-cv-labeling"
+    repo_type: str = "unknown"             # single_skill | multi_skill | collection | mcp_server | unknown
+    skill_path: Optional[str] = None       # path to SKILL.md within repo
+    skill_name: Optional[str] = None       # parsed from SKILL.md frontmatter
+    skill_compatibility: Optional[Dict[str, Any]] = None  # from SKILL.md
+    skill_metadata: Optional[Dict[str, Any]] = None       # from SKILL.md
+
+    # -- Manifest --
+    manifest_raw: Optional[str] = None     # raw capability.yaml content
+    manifest_version: Optional[str] = None
+
+    # -- Capability IR --
+    capability_ir: Optional[Dict[str, Any]] = None   # framework-agnostic IR (JSONB)
+    adaptation_targets: List[str] = field(default_factory=list)  # ["mcp-server", "a2a-agent"]
+
     # -- Capacium Trust --
     fingerprint: Optional[str] = None
     signature_value: Optional[str] = None
@@ -157,17 +175,18 @@ def to_dict(self) -> Dict[str, Any]:
     def from_dict(cls, data: Dict[str, Any]) -> "Listing":
         known = {f.name for f in cls.__dataclass_fields__.values()}
         filtered = {k: v for k, v in data.items() if k in known}
-        for list_field in ("source_urls", "tags", "target_frameworks", "trust_history", "github_topics"):
+        for list_field in ("source_urls", "tags", "target_frameworks", "trust_history", "github_topics", "adaptation_targets"):
             if list_field in filtered and isinstance(filtered[list_field], str):
                 try:
                     filtered[list_field] = json.loads(filtered[list_field])
                 except (json.JSONDecodeError, TypeError):
                     filtered[list_field] = []
-        if "mcp_metadata" in filtered and isinstance(filtered["mcp_metadata"], str):
-            try:
-                filtered["mcp_metadata"] = json.loads(filtered["mcp_metadata"])
-            except (json.JSONDecodeError, TypeError):
-                filtered["mcp_metadata"] = None
+        for dict_field in ("mcp_metadata", "skill_compatibility", "skill_metadata", "capability_ir"):
+            if dict_field in filtered and isinstance(filtered[dict_field], str):
+                try:
+                    filtered[dict_field] = json.loads(filtered[dict_field])
+                except (json.JSONDecodeError, TypeError):
+                    filtered[dict_field] = None
         return cls(**filtered)
 
     def get_mcp_metadata_obj(self) -> Optional[MCPMetadata]:

src/capacium_models/search.py

@@ -54,11 +54,17 @@ def search(self, query: SearchQuery) -> Tuple[List[Listing], int]:
 
         if query.min_trust_state:
             try:
-                min_state = TrustState(query.min_trust_state)
+                from .trust import normalize_legacy_state, LEGACY_STATE_MAP
+                normalized = normalize_legacy_state(query.min_trust_state)
+                min_state = TrustState(normalized)
                 valid_states = [
                     s.value for s in TrustState.ordering()
                     if s >= min_state
                 ]
+                # Include legacy aliases that map to valid states
+                for legacy, modern in LEGACY_STATE_MAP.items():
+                    if modern in valid_states and legacy not in valid_states:
+                        valid_states.append(legacy)
                 if valid_states:
                     placeholders = ", ".join("?" for _ in valid_states)
                     conditions.append(f"trust_state IN ({placeholders})")

src/capacium_models/trust.py

@@ -1,30 +1,31 @@
 """TrustState machine for Exchange listings.
 
 Enforces valid state transitions, records history, and validates entry criteria.
+
+v2: Unified 4-state model (discovered → audited → verified → signed).
+Legacy states (indexed, claimed) are mapped via normalize_legacy_state().
 """
 
 from datetime import datetime
 from enum import Enum
-from typing import List
+from typing import Dict, List
 
 from .models import Listing, TrustTransition
 
 
 class TrustState(str, Enum):
-    DISCOVERED = "discovered"
-    INDEXED = "indexed"
-    CLAIMED = "claimed"
-    VERIFIED = "verified"
-    AUDITED = "audited"
+    DISCOVERED = "discovered"    # Found by crawler, minimal validation
+    AUDITED = "audited"          # Schema validated, composite score computed
+    VERIFIED = "verified"        # Publisher claimed and ownership verified
+    SIGNED = "signed"            # Cryptographic signature present and valid
 
     @classmethod
     def ordering(cls) -> List["TrustState"]:
         return [
             cls.DISCOVERED,
-            cls.INDEXED,
-            cls.CLAIMED,
-            cls.VERIFIED,
             cls.AUDITED,
+            cls.VERIFIED,
+            cls.SIGNED,
         ]
 
     def __lt__(self, other):
@@ -52,21 +53,41 @@ def __le__(self, other):
         return NotImplemented
 
 
+# Legacy state mapping for backward compatibility
+LEGACY_STATE_MAP: Dict[str, str] = {
+    "indexed": "audited",
+    "claimed": "verified",
+}
+
+
+def normalize_legacy_state(state: str) -> str:
+    """Map old trust states to new ones.
+
+    indexed → audited, claimed → verified.
+    All other states pass through unchanged.
+    """
+    return LEGACY_STATE_MAP.get(state, state)
+
+
 class TrustTransitionError(Exception):
     pass
 
 
 VALID_TRANSITIONS = {
-    TrustState.DISCOVERED: {TrustState.INDEXED},
-    TrustState.INDEXED: {TrustState.CLAIMED},
-    TrustState.CLAIMED: {TrustState.VERIFIED},
-    TrustState.VERIFIED: {TrustState.AUDITED, TrustState.CLAIMED},
-    TrustState.AUDITED: {TrustState.VERIFIED},
+    TrustState.DISCOVERED: {TrustState.AUDITED},
+    TrustState.AUDITED: {TrustState.VERIFIED, TrustState.DISCOVERED},
+    TrustState.VERIFIED: {TrustState.SIGNED, TrustState.AUDITED},
+    TrustState.SIGNED: {TrustState.VERIFIED},
 }
 
 
 class TrustMachine:
 
+    @staticmethod
+    def normalize_legacy_state(state: str) -> str:
+        """Map old trust states to new ones (convenience wrapper)."""
+        return normalize_legacy_state(state)
+
     @staticmethod
     def can_transition(from_state: TrustState, to_state: TrustState) -> bool:
         allowed = VALID_TRANSITIONS.get(from_state, set())
@@ -76,27 +97,23 @@ def can_transition(from_state: TrustState, to_state: TrustState) -> bool:
     def validate_entry_criteria(listing: Listing, target_state: TrustState) -> list:
         errors = []
 
-        if target_state == TrustState.INDEXED:
+        if target_state == TrustState.AUDITED:
             if not listing.canonical_name:
-                errors.append("Indexed state requires canonical_name")
+                errors.append("Audited state requires canonical_name")
             if not listing.short_description:
-                errors.append("Indexed state requires short_description")
+                errors.append("Audited state requires short_description")
             if not listing.canonical_source_url:
-                errors.append("Indexed state requires canonical_source_url")
-            if not listing.primary_category:
-                errors.append("Indexed state requires primary_category")
-
-        elif target_state == TrustState.CLAIMED:
-            if not listing.publisher_id:
-                errors.append("Claimed state requires publisher_id")
+                errors.append("Audited state requires canonical_source_url")
 
         elif target_state == TrustState.VERIFIED:
             if not listing.publisher_id:
                 errors.append("Verified state requires publisher_id")
 
-        elif target_state == TrustState.AUDITED:
-            if not listing.publisher_id:
-                errors.append("Audited state requires a verified publisher")
+        elif target_state == TrustState.SIGNED:
+            if not listing.fingerprint:
+                errors.append("Signed state requires fingerprint")
+            if not listing.signature_value:
+                errors.append("Signed state requires signature_value")
 
         return errors
 
@@ -108,7 +125,9 @@ def transition(
         reason: str = "",
         triggered_by: str = "system",
     ) -> Listing:
-        from_state = TrustState(listing.trust_state)
+        # Normalize legacy states before processing
+        current_state_str = normalize_legacy_state(listing.trust_state)
+        from_state = TrustState(current_state_str)
 
         if from_state == to_state:
             raise TrustTransitionError(
@@ -138,7 +157,4 @@ def transition(
         listing.trust_history.append(transition.to_dict())
         listing.updated_at = datetime.utcnow().isoformat()
 
-        if to_state == TrustState.INDEXED and not listing.indexed_at:
-            listing.indexed_at = datetime.utcnow().isoformat()
-
         return listing