Extending

# myadapter/pyproject.toml
[project.entry-points."corespine.vector_store"]
# 名字(左) -> "模块:工厂可调用对象"(右);工厂签名为 (**kwargs) -> 实现实例
pgvector = "myadapter.pg:make_pgvector"

# myadapter/pg.py
def make_pgvector(**kwargs):
    """工厂:返回一个实现该缝 Protocol 的实例(此处省略真实细节)。"""
    return PgVectorStore(**kwargs)

from corespine import Registry

# 同一条缝名 "vector_store" 必须与 entry-point group 后缀一致。
registry: Registry = Registry("vector_store")

# 内置名找不到 → 自动扫 group "corespine.vector_store" 发现 myadapter 的 pgvector。
store = registry.make("pgvector", dsn="postgresql://...")

# 解析大小写 / 连字符 / 留白不敏感:"PG-Vector" / " pgvector " 都解析到同一项。
registry.names()  # 列出【内置 + entry-point 发现】的全部可用名(字典序、去重)

# 你的包 pyproject.toml:真实后端依赖声明为可选 extra(以后端命名)
[project.optional-dependencies]
redis = ["redis>=5"]
openai = ["openai>=1.0"]

from corespine import lazy_extra_import


def make_redis_queue(**kwargs):
    # 只在真正构造该 adapter 时才 import;没装 redis 不影响核心离线默认路径。
    redis = lazy_extra_import("redis", pkg="myadapter", extra="redis")
    client = redis.Redis(**kwargs)
    return RedisQueue(client)

ImportError: 缺少可选依赖 'redis':请先 `pip install myadapter[redis]` 再重试。

from corespine import ConformanceSuite, InvariantPack

# 1) 实现注册表:名字 -> 无参工厂(每格各新建实例,杜绝实现间状态串味)
impls = {"counter": Counter, "broken": BrokenCounter}

# 2) app 自己的不变量包(corespine 核心不含任何具体不变量)
pack = (
    InvariantPack("counter-contract")
    .add("first-add-returns-n", lambda c: assert_eq(c.add(3), 3))
    .add("accumulates", lambda c: assert_eq((c.add(2), c.add(5)), (2, 7)))
)

# 3) 绑成笛卡尔积,逐格跑
suite = ConformanceSuite(impls, pack)
for impl, invariant in suite.cases():
    suite.check(impl, invariant)   # 失败即抛,定位到具体格子
# 或:suite.run() 收集全部结果(不抛);suite.passed() 便捷判全过

from corespine import InProcessPrivacyTraceSink, TraceError

class PrivacyGatedTrace:
    def __init__(self) -> None:
        self._gate = InProcessPrivacyTraceSink()

    def emit_trace(self, code: str, **fields: object) -> None:
        self._gate.emit(code, **fields)   # 命中 FORBIDDEN_KEYS 直接抛,不会落到下游
        ...                               # 过闸后才落自家日志 / DB

from corespine import TraceSink

def run_step(task: str, *, trace: TraceSink | None = None) -> str:
    output = do_work(task)
    if trace is not None:
        trace.emit("step", task_chars=len(task), output_chars=len(output))  # 只发长度
    return output

from corespine import InProcessPrivacyTraceSink, InProcessTraceExporter, TraceExporter, lazy_extra_import

class OTelTraceExporter:                    # 在【你的包】里,不在 corespine 核心
    def __init__(self) -> None:
        # OTel SDK 经你自己的可选 extra 声明,延迟到真正构造时才 import。
        self._otel = lazy_extra_import("opentelemetry.trace", pkg="myadapter", extra="otel")
        ...
    def export(self, event) -> None:        # 只会收到已过隐私校验的 TraceEvent(无需再自查禁词键)
        ...                                 # 建 span / 记 metric:event.code / event.fields(只含元数据)

# host 组装:合法事件先本地记录、再扇出到你的 OTel exporter
sink = InProcessPrivacyTraceSink(exporters=[OTelTraceExporter()])
# 或注入点里的 sink 也能挂:sink.add_exporter(OTelTraceExporter())