generated from coulomb/repo-seed
feat(engine): per-shard extension activation (WP-0014 T3, ADR-0001)
engine/activation.py: ActivationContext (shard/tenant, no authz), pluggable ActivationProvider protocol, StaticProvider standalone default (zero-dep, global flags + per-shard scoping + per-ext config), ActivationResolver (candidate ids -> active set / activation profile), and feature_control_provider() lazy factory (returns None when feature_control_sdk absent -> degrade to static; OpenFeature- shaped when present). Availability only. 6 tests green, coverage held, pyflakes clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -5,6 +5,13 @@ is *one shard*: it is consumed by the orchestrator only via its `EngineShardAdap
|
||||
imports the derived tier (`union`/`projection`).
|
||||
"""
|
||||
|
||||
from shard_wiki.engine.activation import (
|
||||
ActivationContext,
|
||||
ActivationProvider,
|
||||
ActivationResolver,
|
||||
StaticProvider,
|
||||
feature_control_provider,
|
||||
)
|
||||
from shard_wiki.engine.extension import (
|
||||
ActiveExtensions,
|
||||
Extension,
|
||||
@@ -23,4 +30,9 @@ __all__ = [
|
||||
"ExtensionError",
|
||||
"ExtensionRuntime",
|
||||
"ActiveExtensions",
|
||||
"ActivationContext",
|
||||
"ActivationProvider",
|
||||
"StaticProvider",
|
||||
"ActivationResolver",
|
||||
"feature_control_provider",
|
||||
]
|
||||
|
||||
129
src/shard_wiki/engine/activation.py
Normal file
129
src/shard_wiki/engine/activation.py
Normal file
@@ -0,0 +1,129 @@
|
||||
"""Per-shard extension activation (WikiEngineCoreArchitecture E-4/E-8, ADR-0001).
|
||||
|
||||
Decides *which registered extensions are active* for a given shard — **availability only, never
|
||||
authorization**. The mechanism is OpenFeature-shaped and **provider-pluggable**:
|
||||
|
||||
- :class:`StaticProvider` is the **standalone / L0 default** — zero external dependency, in-process
|
||||
flags with optional per-shard scoping. A kernel-only or offline shard uses this.
|
||||
- :func:`feature_control_provider` lazily wraps the helix_forge ``feature_control_sdk`` (over
|
||||
OpenFeature) when it is installed; absent, it returns ``None`` and the caller falls back to the
|
||||
static default. This mirrors shard-wiki's identity-provider ladder (local default → external
|
||||
when present), and keeps the engine core pure-stdlib.
|
||||
|
||||
An *activation profile* is ``{extension id → config}``; the active id set then feeds the
|
||||
extension runtime's `activate()` (T2) and the derived capability profile (T4, E-5).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
__all__ = [
|
||||
"ActivationContext",
|
||||
"ActivationProvider",
|
||||
"StaticProvider",
|
||||
"ActivationResolver",
|
||||
"feature_control_provider",
|
||||
]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ActivationContext:
|
||||
"""Scope for an activation decision. Carries no principal/authz — availability only."""
|
||||
|
||||
shard_id: str
|
||||
tenant_id: str | None = None
|
||||
extra: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
d: dict[str, Any] = {"shard_id": self.shard_id}
|
||||
if self.tenant_id is not None:
|
||||
d["tenant_id"] = self.tenant_id
|
||||
d.update(self.extra)
|
||||
return d
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ActivationProvider(Protocol):
|
||||
"""Evaluates feature availability for an extension key in a context (OpenFeature-shaped)."""
|
||||
|
||||
def is_active(self, feature_key: str, context: Mapping[str, Any]) -> bool: ...
|
||||
|
||||
def config(self, feature_key: str, context: Mapping[str, Any]) -> Mapping[str, Any]: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StaticProvider:
|
||||
"""The standalone default: in-process flags, optionally overridden per shard. Zero deps.
|
||||
|
||||
``flags`` is the base availability map; ``per_shard[shard_id]`` overrides it for one shard;
|
||||
``configs[feature_key]`` supplies per-extension config. Unknown keys → ``default``.
|
||||
"""
|
||||
|
||||
flags: Mapping[str, bool] = field(default_factory=dict)
|
||||
per_shard: Mapping[str, Mapping[str, bool]] = field(default_factory=dict)
|
||||
configs: Mapping[str, Mapping[str, Any]] = field(default_factory=dict)
|
||||
default: bool = False
|
||||
|
||||
def is_active(self, feature_key: str, context: Mapping[str, Any]) -> bool:
|
||||
shard = context.get("shard_id")
|
||||
scoped = self.per_shard.get(shard, {}) if shard is not None else {}
|
||||
if feature_key in scoped:
|
||||
return scoped[feature_key]
|
||||
return self.flags.get(feature_key, self.default)
|
||||
|
||||
def config(self, feature_key: str, context: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
return self.configs.get(feature_key, {})
|
||||
|
||||
|
||||
class ActivationResolver:
|
||||
"""Maps candidate extension ids → the active set / activation profile for a context."""
|
||||
|
||||
def __init__(self, provider: ActivationProvider) -> None:
|
||||
self.provider = provider
|
||||
|
||||
def active_extensions(
|
||||
self, candidate_ids: Iterable[str], context: ActivationContext
|
||||
) -> set[str]:
|
||||
ctx = context.as_dict()
|
||||
return {eid for eid in candidate_ids if self.provider.is_active(eid, ctx)}
|
||||
|
||||
def activation_profile(
|
||||
self, candidate_ids: Iterable[str], context: ActivationContext
|
||||
) -> dict[str, Mapping[str, Any]]:
|
||||
"""``{extension id → config}`` for the active subset."""
|
||||
ctx = context.as_dict()
|
||||
return {
|
||||
eid: self.provider.config(eid, ctx)
|
||||
for eid in candidate_ids
|
||||
if self.provider.is_active(eid, ctx)
|
||||
}
|
||||
|
||||
|
||||
def feature_control_provider(domain: str | None = None) -> ActivationProvider | None:
|
||||
"""Return a feature-control-backed provider if ``feature_control_sdk`` is importable, else
|
||||
``None`` (caller falls back to :class:`StaticProvider`). Lazy import keeps the engine core
|
||||
dependency-free (ADR-0001)."""
|
||||
try: # optional engine extra — not a core dependency
|
||||
from feature_control_sdk import FeatureControlClient # type: ignore
|
||||
except ImportError:
|
||||
return None
|
||||
|
||||
client = FeatureControlClient(domain=domain)
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _FeatureControlProvider:
|
||||
_client: Any
|
||||
|
||||
def is_active(self, feature_key: str, context: Mapping[str, Any]) -> bool:
|
||||
return bool(
|
||||
self._client.get_boolean_value(feature_key, False, context=dict(context))
|
||||
)
|
||||
|
||||
def config(self, feature_key: str, context: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
getter = getattr(self._client, "get_object_value", None)
|
||||
return dict(getter(feature_key, {}, context=dict(context))) if getter else {}
|
||||
|
||||
return _FeatureControlProvider(client)
|
||||
Reference in New Issue
Block a user