First implementation

src/feature_control_sdk/__init__.py

@@ -0,0 +1,181 @@
+"""
+feature-control-sdk
+
+Thin organization wrapper around OpenFeature SDK.
+
+See docs/canon-mapping.md for how context projects from ITC-ORG/ACCESS/LAND facts.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+try:
+    from openfeature import OpenFeatureAPI
+    from openfeature.evaluation_context import EvaluationContext
+    from openfeature.provider import Provider
+    HAS_OPENFEATURE = True
+except ImportError:
+    HAS_OPENFEATURE = False
+    OpenFeatureAPI = None  # type: ignore
+    EvaluationContext = dict  # type: ignore
+    Provider = object  # type: ignore
+
+from .providers.local import LocalProvider
+from .registry import FeatureDefinition, FeatureRegistry
+from .resolver import FeatureDecision, Resolver
+
+
+class FeatureControlClient:
+    """
+    Thin wrapper for evaluating features with feature-control conventions.
+
+    Usage:
+        client = FeatureControlClient()
+        client.set_provider(LocalProvider({"my.feature": True}))
+        value = client.get_boolean_value("my.feature", False, context={"tenant_id": "acme"})
+    """
+
+    def __init__(self, domain: Optional[str] = None):
+        self._api = OpenFeatureAPI.get_instance() if HAS_OPENFEATURE else None
+        self._domain = domain
+        if HAS_OPENFEATURE:
+            self._client = self._api.get_client(domain) if domain else self._api.get_client()
+        else:
+            self._client = None  # local mode, will use resolver or direct
+        self._resolver: Optional[Resolver] = None
+        self._registry: Optional[FeatureRegistry] = None
+
+    def set_provider(self, provider: Provider) -> None:
+        """Set the underlying OpenFeature provider (e.g. LocalProvider for tests)."""
+        if HAS_OPENFEATURE:
+            if self._domain:
+                self._api.set_provider(self._domain, provider)
+            else:
+                self._api.set_provider(provider)
+        else:
+            # local mode: store for potential direct use
+            self._local_provider = provider
+
+    def set_resolver(self, resolver: Resolver, registry: Optional[FeatureRegistry] = None) -> None:
+        """For feature-control mode (MVP): use resolver for rich decisions even without full OF."""
+        self._resolver = resolver
+        self._registry = registry or resolver.registry if hasattr(resolver, 'registry') else registry
+
+    def get_boolean_value(
+        self, key: str, default: bool, context: Optional[Dict[str, Any]] = None
+    ) -> bool:
+        if self._resolver is not None:
+            decision = self._resolver.evaluate(key, default, context)
+            return bool(decision.value)
+        if HAS_OPENFEATURE and self._client:
+            of_context = self._build_context(context)
+            return self._client.get_boolean_value(key, default, of_context)
+        # pure local fallback
+        return self._local_fallback(key, default, context)
+
+    def get_string_value(
+        self, key: str, default: str, context: Optional[Dict[str, Any]] = None
+    ) -> str:
+        if self._resolver is not None:
+            decision = self._resolver.evaluate(key, default, context)
+            return str(decision.value)
+        if HAS_OPENFEATURE and self._client:
+            of_context = self._build_context(context)
+            return self._client.get_string_value(key, default, of_context)
+        return str(self._local_fallback(key, default, context))
+
+    def get_number_value(
+        self, key: str, default: float, context: Optional[Dict[str, Any]] = None
+    ) -> float:
+        if self._resolver is not None:
+            decision = self._resolver.evaluate(key, default, context)
+            return float(decision.value)
+        if HAS_OPENFEATURE and self._client:
+            of_context = self._build_context(context)
+            return self._client.get_number_value(key, default, of_context)
+        return float(self._local_fallback(key, default, context))
+
+    def get_object_value(
+        self, key: str, default: Dict[str, Any], context: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        if self._resolver is not None:
+            decision = self._resolver.evaluate(key, default, context)
+            return dict(decision.value) if isinstance(decision.value, dict) else default
+        if HAS_OPENFEATURE and self._client:
+            of_context = self._build_context(context)
+            return self._client.get_object_value(key, default, of_context)
+        val = self._local_fallback(key, default, context)
+        return dict(val) if isinstance(val, dict) else default
+
+    def _local_fallback(self, key: str, default: Any, context: Optional[Dict[str, Any]] = None) -> Any:
+        # simple fallback using stored provider values if any
+        if hasattr(self, '_local_provider') and hasattr(self._local_provider, '_values'):
+            return self._local_provider._values.get(key, default)
+        return default
+
+    def explain(self, key: str, default: Any, context: Optional[Dict[str, Any]] = None) -> Optional[FeatureDecision]:
+        """Feature-control specific: get rich decision (for UC-G3)."""
+        if self._resolver is not None:
+            return self._resolver.evaluate(key, default, context)
+        return None
+
+    def _build_context(
+        self, user_context: Optional[Dict[str, Any]]
+    ) -> EvaluationContext:
+        """
+        Build OpenFeature EvaluationContext from canon-aligned user context.
+
+        Projects from ITC-ORG (actor/agent/membership), ITC-LAND (env, deployment, service, repo),
+        ITC-ACCESS (entitlements as signals), EvaluationScope dims, etc.
+
+        See docs/canon-mapping.md, EvaluationScope in specs/UseCaseCatalog.md, and WP-0003.
+        """
+        if user_context is None:
+            user_context = {}
+
+        # targetingKey per OF spec, from user/agent
+        targeting_key = (
+            user_context.get("targeting_key")
+            or user_context.get("user_id")
+            or user_context.get("agent_id")
+            or user_context.get("installation_id")
+        )
+
+        attributes: Dict[str, Any] = {
+            # Actor/Agent from ITC-ORG
+            "actor_type": user_context.get("actor_type", "human"),
+            "user_id": user_context.get("user_id"),
+            "agent_id": user_context.get("agent_id"),
+            "roles": user_context.get("roles"),
+            # Landscape from ITC-LAND
+            "installation_id": user_context.get("installation_id"),
+            "environment": user_context.get("environment"),
+            "deployment_id": user_context.get("deployment_id"),
+            "service": user_context.get("service"),
+            "repository": user_context.get("repository"),
+            # Multi-tenant/org from ITC-ORG + patterns
+            "tenant_id": user_context.get("tenant_id"),
+            "domain_id": user_context.get("domain_id"),
+            "organization_id": user_context.get("organization_id"),
+            "group_ids": user_context.get("group_ids"),
+            "plan": user_context.get("plan"),
+            "region": user_context.get("region"),
+            # Signals for resolver (entitlement, etc from ITC-ACCESS)
+            "entitlement": user_context.get("entitlement"),
+            "compute_class": user_context.get("compute_class"),
+            "trust_level": user_context.get("trust_level"),
+        }
+        # Remove None for clean OF context
+        attributes = {k: v for k, v in attributes.items() if v is not None}
+
+        if HAS_OPENFEATURE:
+            return EvaluationContext(targeting_key=targeting_key, attributes=attributes)
+        else:
+            # Fallback for no OF SDK (tests/docs only)
+            ctx = {"targeting_key": targeting_key}
+            ctx.update(attributes)
+            return ctx  # type: ignore[return-value]
+
+
+__all__ = ["FeatureControlClient", "LocalProvider", "FeatureDefinition", "FeatureRegistry", "FeatureDecision", "Resolver"]

src/feature_control_sdk/providers/__init__.py

@@ -0,0 +1 @@
+"""Local and test providers for feature-control-sdk."""

src/feature_control_sdk/providers/local.py

@@ -0,0 +1,97 @@
+"""
+LocalProvider: in-memory/test provider for development and unit tests.
+
+Implements a simple dict-backed provider. Supports boolean, string, number, object.
+No network, deterministic, safe defaults.
+
+See WP-0003 T01 and T04.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+try:
+    from openfeature.evaluation_context import EvaluationContext
+    from openfeature.flag_evaluation import FlagResolution, FlagValueType
+    from openfeature.provider import Provider
+    from openfeature.provider.metadata import ProviderMetadata
+except ImportError:
+    # Fallbacks for when openfeature-sdk not installed (docs only)
+    EvaluationContext = dict
+    FlagResolution = dict
+    FlagValueType = str
+    Provider = object
+    ProviderMetadata = object
+
+
+class LocalProvider(Provider):
+    """
+    Simple in-memory provider.
+
+    Example:
+        provider = LocalProvider({
+            "my.feature": True,
+            "string.feature": "value",
+            "num.feature": 42,
+            "obj.feature": {"key": "val"}
+        })
+        client.set_provider(provider)
+        assert client.get_boolean_value("my.feature", False) == True
+    """
+
+    def __init__(self, values: Optional[Dict[str, Any]] = None):
+        self._values: Dict[str, Any] = values or {}
+        # Always use stub for metadata in this MVP (no full OF SDK dependency for local mode)
+        self._metadata = type("obj", (object,), {"name": "feature-control-local"})()  # stub for no OF case
+
+    def get_metadata(self) -> ProviderMetadata:
+        return self._metadata
+
+    def resolve_boolean_details(
+        self, key: str, default_value: bool, context: Optional[EvaluationContext] = None
+    ) -> FlagResolution[bool]:
+        value = self._values.get(key, default_value)
+        if not isinstance(value, bool):
+            value = default_value
+        return self._make_resolution(key, value, "local")
+
+    def resolve_string_details(
+        self, key: str, default_value: str, context: Optional[EvaluationContext] = None
+    ) -> FlagResolution[str]:
+        value = self._values.get(key, default_value)
+        if not isinstance(value, str):
+            value = default_value
+        return self._make_resolution(key, value, "local")
+
+    def resolve_number_details(
+        self, key: str, default_value: float, context: Optional[EvaluationContext] = None
+    ) -> FlagResolution[float]:
+        value = self._values.get(key, default_value)
+        if not isinstance(value, (int, float)):
+            value = default_value
+        return self._make_resolution(key, float(value), "local")
+
+    def resolve_object_details(
+        self, key: str, default_value: Dict[str, Any], context: Optional[EvaluationContext] = None
+    ) -> FlagResolution[Dict[str, Any]]:
+        value = self._values.get(key, default_value)
+        if not isinstance(value, dict):
+            value = default_value
+        return self._make_resolution(key, value, "local")
+
+    def _make_resolution(self, key: str, value: Any, reason: str) -> FlagResolution[Any]:
+        return {
+            "value": value,
+            "reason": reason,
+            "variant": None,
+            "flag_metadata": {"provider": "feature-control-local"},
+            "error_code": None,
+            "error_message": None,
+        }  # type: ignore[return-value]
+
+    # For compatibility if real SDK expects class methods
+    resolve_boolean_details = resolve_boolean_details  # type: ignore
+    resolve_string_details = resolve_string_details  # type: ignore
+    resolve_number_details = resolve_number_details  # type: ignore
+    resolve_object_details = resolve_object_details  # type: ignore

src/feature_control_sdk/registry.py

@@ -0,0 +1,106 @@
+"""
+Canonical feature registry (T02 in WP-0003).
+
+Git-backed declarative baseline for FeatureDefinition.
+
+For MVP: simple in-memory + file (json) store simulating Git.
+Validation: owner required, temp features need review/expiry.
+Integrates with resolver (T03).
+
+Per specs/UseCaseCatalog.md UC-G1, PRD FR-2, canon-mapping.md (Feature as ProducerCapability).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class FeatureDefinition:
+    """Core metadata for a feature. Matches PRD data model sketch."""
+
+    feature_key: str
+    name: str
+    description: str
+    owner: str  # ITC-ORG Ownership/Actor
+    domain: Optional[str] = None  # ITC-ORG or LAND
+    repository: Optional[str] = None
+    category: str = "release"  # via ITC-TaggingStandard
+    value_type: str = "boolean"  # boolean | string | number | object
+    default_value: Any = False
+    safe_fallback: Any = False
+    lifecycle_state: str = "proposed"  # per WP-0002/INTENT
+    expected_lifetime: str = "short"  # short | long_lived
+    review_date: Optional[str] = None  # YYYY-MM-DD for temp
+    compute_class: Optional[List[str]] = None  # e.g. ["gpu_heavy"]
+    security_sensitivity: str = "low"
+    tenant_configurable: bool = False
+    requires_approval: bool = False
+    documentation: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "FeatureDefinition":
+        return cls(**data)
+
+
+class FeatureRegistry:
+    """
+    Git-style registry for FeatureDefinitions.
+
+    MVP: load/save to JSON file (commit this to Git in real use).
+    In production: could use gitpython or just treat files as declarative.
+
+    Usage:
+        reg = FeatureRegistry("features.json")
+        reg.register(FeatureDefinition(...))
+        reg.save()  # "git commit" the file
+    """
+
+    def __init__(self, storage_path: str = "features.json"):
+        self.storage_path = Path(storage_path)
+        self._features: Dict[str, FeatureDefinition] = {}
+        self.load()
+
+    def load(self) -> None:
+        """Load from "Git" baseline (json file)."""
+        if self.storage_path.exists():
+            data = json.loads(self.storage_path.read_text())
+            self._features = {
+                k: FeatureDefinition.from_dict(v) for k, v in data.items()
+            }
+
+    def save(self) -> None:
+        """Save to Git baseline. In real: git add + commit the json."""
+        data = {k: v.to_dict() for k, v in self._features.items()}
+        self.storage_path.write_text(json.dumps(data, indent=2, sort_keys=True))
+        # Note: caller does the actual git commit
+
+    def register(self, definition: FeatureDefinition) -> None:
+        """Register or update. Validates per PRD FR-2 / UC-G1."""
+        if not definition.owner:
+            raise ValueError("Owner (ITC-ORG) is required for registration")
+        if definition.expected_lifetime == "short" and not definition.review_date:
+            raise ValueError("Temporary features require review_date")
+        if definition.feature_key in self._features:
+            # Update allowed for now (in real: approval for changes)
+            pass
+        self._features[definition.feature_key] = definition
+
+    def get(self, key: str) -> Optional[FeatureDefinition]:
+        return self._features.get(key)
+
+    def list_all(self) -> List[FeatureDefinition]:
+        return list(self._features.values())
+
+    def keys(self) -> List[str]:
+        """For discovery / scanner (UC-A3)."""
+        return list(self._features.keys())
+
+    def to_dict(self) -> Dict[str, Dict[str, Any]]:
+        return {k: v.to_dict() for k, v in self._features.items()}

src/feature_control_sdk/resolver.py

@@ -0,0 +1,112 @@
+"""
+Multi-scope resolver with EvaluationScope and signals (T03).
+
+For MVP: simple in-memory resolver using registry + provider values.
+Composes signals per precedence in WP-0003.
+
+Rich FeatureDecision output.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from .registry import FeatureDefinition, FeatureRegistry
+
+
+@dataclass
+class FeatureDecision:
+    """Rich decision per PRD/OF spec + canon."""
+
+    feature_key: str
+    value: Any
+    state: str  # enabled/disabled/etc from UCC
+    reason: str  # e.g. "tenant_policy", "kill_switch", "default"
+    source: str  # e.g. "tenant_rule", "global_default"
+    scope: Optional[str] = None  # EvaluationScope e.g. "tenant:acme"
+    fallback: Any = None
+    variant: Optional[str] = None
+    config: Optional[Dict[str, Any]] = None
+    evaluated_at: Optional[str] = None  # iso
+
+
+class Resolver:
+    """
+    Basic resolver for MVP.
+
+    Uses registry for defs + provider for current values/signals.
+    Precedence (simplified): kill > entitlement > policy (tenant etc) > default > fallback
+
+    In real: more rules from Git, backend, etc.
+    """
+
+    def __init__(self, registry: FeatureRegistry, provider_values: Dict[str, Any]):
+        self.registry = registry
+        self.values = provider_values  # from provider or overrides (use ._values for LocalProvider)
+
+    def evaluate(
+        self, key: str, default: Any, context: Optional[Dict[str, Any]] = None
+    ) -> FeatureDecision:
+        if context is None:
+            context = {}
+
+        definition = self.registry.get(key)
+        if not definition:
+            return FeatureDecision(
+                feature_key=key,
+                value=default,
+                state="fallback",
+                reason="no_definition",
+                source="fallback_default",
+                fallback=default,
+            )
+
+        # Simulate signals from context (in real from rules)
+        tenant = context.get("tenant_id")
+        is_agent = context.get("actor_type") == "agent"
+        kill = self.values.get(f"kill:{key}", False)
+        entitlement = context.get("entitlement")  # simplified signal
+
+        value = self.values.get(key, definition.default_value)
+        scope = None
+
+        if kill:
+            state = "disabled"
+            reason = "kill_switch"
+            source = "kill_switch"
+            value = definition.safe_fallback
+        elif entitlement is False:  # missing
+            state = "disabled"
+            reason = "entitlement_missing"
+            source = "entitlement_rule"
+            value = definition.safe_fallback
+        elif tenant and self.values.get(f"tenant:{tenant}:{key}") is not None:
+            # tenant override
+            state = "enabled" if value else "disabled"
+            reason = "tenant_policy"
+            source = "tenant_rule"
+            scope = f"tenant:{tenant}"
+        else:
+            state = "enabled" if value else "disabled"
+            reason = "default"
+            source = "global_default"
+            scope = None
+
+        # For compute/agent examples
+        if definition.compute_class and context.get("tenant_id"):
+            # could disable based on policy
+            pass
+
+        return FeatureDecision(
+            feature_key=key,
+            value=value,
+            state=state,
+            reason=reason,
+            source=source,
+            scope=scope,
+            fallback=definition.safe_fallback,
+            variant=context.get("variant"),
+            config=None,
+            evaluated_at="2026-06-14T12:00:00Z",  # stub
+        )