Cordis Python 版：介绍与实现思路

1. 论文要点：为什么需要 Cordis

论文提出现代软件在“动态组合”上缺少形式化基础，并把动态组合拆成两个正交维度：

时间可组合性（Temporal Composability）：组件被移除时，它对共享环境造成的所有副作用必须能够被完整、安全地撤销。
空间可组合性（Spatial Composability）：组件可以声明、发现、解析并响应其他组件的依赖变化。

静态场景下，这两个维度分别对应 RAII / 词法作用域和模块导入解析；在动态场景下，组件会在运行期到达或离开，因此需要运行期机制来支撑可逆副作用和响应式依赖。

核心机制

论文概念	含义
`revertible effects`	所有通过 Context 的修改都经由 `ctx.effect(callback)` 登记；callback 返回逆操作，卸载时按 LIFO 顺序执行，实现“效果可逆”。
`reactive coeffects`	组件用 `inject` 声明依赖；依赖对应“coeffect”，其出现、消失或身份变化会触发依赖方自动激活/停用。
`unified context`	效果上下文和余效果上下文统一为同一个 Context，通过它中介所有读写，从而获得观察等价性。
`component / fiber`	组件是可声明依赖、可执行效果函数的代码单元；fiber 是组件的一次运行实例，有完整生命周期。

实现分层

核心库：实现 effect tracking、coeffect resolution、fiber 生命周期、Context 访问。
组件加载器：把声明式配置（YAML/JSON 等）落实为 fiber 树，支持增量协调。
应用框架：在 Cordis 之上构建具体领域框架（论文中案例为 Koishi）。

Python 版定位：不是重写一个具体应用框架，而是实现“元框架”核心——提供通用的动态组合语义，让上层应用或 Agent Harness 可以像使用 Node 版 Cordis 一样使用它。

2. Node Cordis 包参考

Node 版 Cordis（v4）已开源并作为 DeepSeek Harness 的底层插件框架。它是我们 Python 移植的主要 API 与语义参照。

2.1 快速示例（TypeScript 惯例）

const root = new Context()

class Counter extends Service {
  value = 0
  constructor(ctx: Context) {
    super(ctx, 'counter')
  }
  next() { return ++this.value }
}

const greeter = Object.assign((ctx: Context) => {
  ctx.on('app/ready', (msg) => console.log(ctx.counter.next(), msg))
}, { inject: ['counter'] })

await root.plugin(Counter)
await root.plugin(greeter)
root.emit('app/ready', 'started')
await root.fiber.dispose()

2.2 核心对象

对象/API	作用
`Context`	根依赖容器，也是插件的运行上下文；通过 Proxy 支持 `ctx.serviceName` 属性访问。
`ctx.plugin()`	装载一个插件，返回 `Fiber`；Fiber 可 `await`，等待启动完成。
`ctx.inject()`	声明依赖后运行函数式插件；依赖不满足时等待，满足后自动生命周期管理。
`ctx.effect()`	登记可逆副作用；支持函数返回 disposer、同步/异步 iterable 等形态。
`ctx.provide()`	注册服务，自动在 fiber 卸载时注销并通知依赖方。
`ctx.get() / ctx.set()`	反射式读写 coeffect 值（不强制通过 inject 声明）。
`ctx.isolate() / ctx.intercept()`	创建子 Context：隔离某个服务的绑定域，或为服务访问追加拦截/配置元数据。
`ctx.on() / ctx.emit()`	类型化事件总线，监听器随 fiber 自动注销。
`ctx.parallel / serial / bail / waterfall`	不同事件分发模式：并行、串行、短路、中间件式瀑布。
`Fiber`	组件实例，状态机：PENDING → LOADING → ACTIVE → UNLOADING → DISPOSED，以及 FAILED。
`Service`	基类；子类构造时调用 `super(ctx, name)` 即自动注册为服务。
`RegistryService`	插件规范化、运行时记录、fiber 集合管理。
`ReflectService`	支撑 Context Proxy：属性解析、provide/get/set、accessor、mixin。

2.3 加载器与 HMR

Node 生态中，@cordisjs/plugin-loader、@cordisjs/plugin-include、@cordisjs/plugin-group、@cordisjs/plugin-hmr 提供声明式配置和热模块替换。Python 版应保留这些概念：

Entry：声明一个 fiber 的稳定记录，含 id / url / isolate / intercept / config / disabled。
Group：以子条目列表为配置，递归协调子树。
Include：把外部 YAML/JSON 配置文件接枝到配置树中。
HMR：模块分类 → 过期条目检测 → 事务式重载。

3. Python 版总体设计

3.1 设计目标

忠实还原论文语义：所有 Context 修改都走可逆效果；依赖变化自动驱动生命周期。
贴近 Node Cordis API：降低从 TS 生态迁移的心理与概念成本。
Python 异步优先：使用 asyncio 承载 async effect、异步生命周期、async generator 效果迭代器。
保持精简可扩展：核心包不绑定具体领域；加载器、HMR、日志、计时器等作为扩展模块或独立插件。
可打包、可测试：采用 src 布局，配置 pytest、ruff/mypy（可选），用 uv 管理依赖与构建。

3.2 包结构

cordis_py/
├── pyproject.toml
├── README.md
├── LICENSE
├── docs/
│   └── cordis_py_intro.html        # 本文档
├── src/
│   └── cordis_py/
│       ├── __init__.py             # 公共导出
│       ├── context.py              # Context 根/子上下文、属性解析
│       ├── fiber.py                # Fiber、FiberState、effect 执行与回收
│       ├── events.py               # EventsService：emit/parallel/serial/bail/waterfall
│       ├── registry.py             # Plugin 规范化、RegistryService
│       ├── reflect.py              # 服务注册表、get/set/provide/accessor/mixin
│       ├── service.py              # Service 基类、@inject 装饰器
│       ├── loader.py               # Entry、Loader、Group、Include
│       ├── hmr.py                  # 热模块替换
│       ├── config.py               # YAML/JSON 配置解析与校验（可选依赖）
│       └── utils.py                # DisposableList、错误码、公共工具
├── tests/
│   ├── test_effects.py
│   ├── test_fiber_lifecycle.py
│   ├── test_events.py
│   ├── test_services.py
│   ├── test_loader.py
│   └── test_hmr.py
└── .python-version                 # 3.12

3.3 核心抽象

Fiber：组件运行实例

class FiberState(enum.Enum):
    PENDING = "pending"      # 等待依赖
    LOADING = "loading"      # 正在执行组件
    ACTIVE = "active"        # 已激活、正常提供
    FAILED = "failed"        # 启动/校验失败
    UNLOADING = "unloading"  # 正在回滚
    DISPOSED = "disposed"    # 已彻底移除

@dataclass
class Fiber:
    uid: int
    parent: Context
    inject: dict[str, Any]
    config: Any
    state: FiberState
    effects: list[Disposable]          # LIFO 逆操作账本
    committed: dict[str, Provider]     # 激活时依赖视图
    target: str | None                 # 当前目标状态
    inertia: asyncio.Task | None       # 在途 transition
    store: dict[str, Impl] | None      # 本 fiber 提供的服务

效果（Effect）

论文中 ctx.effect 接受回调，回调返回：

一个 disposer（同步或异步函数）；
一个同步/异步 iterable，逐个 yield disposer；
或者什么也不返回（表示无副作用）。

Python 实现可借助 inspect.isgenerator / inspect.isasyncgen，将每个 disposer 压入当前 fiber 的 effect 栈。卸载时倒序执行并 await 异步 disposer。

async def execute(callback, guard):
    iterable = callback()
    disposers = []
    if inspect.isasyncgen(iterable):
        async for d in iterable:
            if callable(d):
                disposers.append(d)
    elif inspect.isgenerator(iterable):
        for d in iterable:
            if callable(d):
                disposers.append(d)
    elif callable(iterable):
        disposers.append(iterable)
    # 若需要支持“边执行边允许中断”，则按论文 Algorithm 1 在每步检查 guard
    return disposers  # 由调用方组合为单个逆操作

Context：服务与依赖的中介

class Context:
    fiber: Fiber
    parent: Context | None
    _isolate: dict[str, object]       # 服务名 -> realm
    _intercept: dict[str, Any]        # 服务名 -> 拦截/配置元数据

    # 核心效果原语
    def effect(self, callback: EffectCallback) -> Disposable: ...
    # 依赖相关
    def plugin(self, plugin: Plugin, config: Any = None) -> Fiber: ...
    def inject(self, deps: Inject, callback: Callable) -> Fiber: ...
    # coeffect
    def get(self, name: str, strict: bool = True) -> Any: ...
    def set(self, name: str, value: Any) -> None: ...
    def provide(self, name: str, value: Any) -> Disposable: ...
    # 作用域
    def isolate(self, name: str, realm: object | None = None) -> Context: ...
    def intercept(self, name: str, metadata: Any) -> Context: ...
    # 事件
    def on(self, event: str, handler, *, prepend=False) -> Disposable: ...
    def emit(self, event: str, *args) -> None: ...
    async def parallel(self, event: str, *args) -> None: ...
    async def serial(self, event: str, *args) -> Any: ...
    def bail(self, event: str, *args) -> Any: ...
    def waterfall(self, event: str, *args, next_fn=None) -> Any: ...

属性访问与注入校验

Python 没有 JavaScript Proxy，但可用 __getattr__ 实现类似语义：当插件访问 ctx.model 时，从当前 fiber 的 committed 视图向上查找；若该名称在当前 fiber 的 inject 中但尚未 committed，则抛“依赖未激活”；若一直到根都没有声明，则抛“未声明访问”。这与论文 Algorithm 6 的语义一致。

def __getattr__(self, name: str):
    if name.startswith("_"):
        raise AttributeError(name)
    fiber = self.fiber
    while True:
        if name in fiber.committed:
            return fiber.committed[name]
        if name in fiber.inject:
            raise InactiveAccess(name)
        if fiber.parent is None:
            raise UndeclaredAccess(name)
        fiber = fiber.parent.fiber

插件形态

# 函数式
def my_plugin(ctx: Context, config: dict) -> Disposable | None:
    ctx.provide("foo", Foo())
    return lambda: cleanup()

my_plugin.inject = ["foo"]

# 类式
class MyPlugin:
    inject = ["foo"]
    def __init__(self, ctx: Context, config: dict):
        self.ctx = ctx
        ctx.provide("bar", Bar())

# 对象式
class MyObjectPlugin:
    def apply(self, ctx: Context, config: dict): ...

事件分发

方法	语义	Python 注意点
`emit`	同步广播，不等待异步监听器	同步广播即可；异步监听器可 `asyncio.ensure_future` 调度。
`parallel`	并发执行并等待全部完成	用 `asyncio.gather(..., return_exceptions=True)` 或 `asyncio.TaskGroup`。
`serial`	按注册顺序逐一 await，遇 bail 值短路	把非 `None/False` 返回值视为短路值。
`bail`	同步短路，返回第一个非空结果	适合“找到第一个能处理者”的场景。
`waterfall`	中间件链	监听器接收 `(*args, next)`；不调用 `next()` 即短路。

3.4 实现依据与边界

本实现不参考任何现有的 Python Cordis 包。

设计参考仅包括：

论文《A Programming Paradigm for Spatiotemporal Composability》；
Node.js 版 Cordis（Cordis v4 / DeepSeek Harness vendor）。

因此本项目不存在需要兼容的历史 Python 实现；API 以 Node 版 Cordis 为参照，并结合 Python 异步生态做了必要调整。

4. 核心 API 草案与示例

4.1 最小可运行示例

import asyncio
from cordis_py import Context, Service, inject


class Greeter(Service):
    def __init__(self, ctx: Context):
        super().__init__(ctx, "greeter")

    def hello(self, name: str) -> str:
        return f"Hello, {name}!"


@inject("greeter")
def greeter_plugin(ctx: Context, config: dict):
    def on_ready(msg: str):
        print(ctx.greeter.hello(msg))

    ctx.on("app/ready", on_ready)
    return lambda: print("greeter_plugin disposed")


async def main():
    root = Context()
    await root.plugin(Greeter)
    await root.plugin(greeter_plugin)

    root.emit("app/ready", "Cordis Python")
    await root.fiber.dispose()


asyncio.run(main())

4.2 生命周期与依赖响应

async def main():
    root = Context()

    # 先加载消费者，依赖未满足，进入 PENDING
    consumer = root.plugin(consumer_plugin)   # inject = ["model"]

    # 加载提供者后，消费者自动从 PENDING -> LOADING -> ACTIVE
    provider = root.plugin(model_plugin, {"provider": "deepseek"})
    await provider.await()
    await consumer.await()

    # 卸载提供者，依赖方应自动回滚并进入 PENDING 或 UNLOADING
    await provider.dispose()

    # 重新提供，消费者恢复
    provider2 = root.plugin(model_plugin, {"provider": "openai"})
    await provider2.await()
    await consumer.await()

    await root.fiber.dispose()

4.3 声明式 Loader 配置草案

# cordis.yml
plugins:
  - id: greeter
    url: "myapp.plugins:Greeter"
    config:
      language: "zh"
  - id: storage
    url: "myapp.plugins:Storage"
    config:
      backend: "sqlite"
    disabled: false
  - id: web
    url: "myapp.plugins:Web"
    isolate:
      db: true          # 本地 realm：随条目移动
    intercept:
      logger:
        level: "debug"
    config:
      port: 8080

from cordis_py.loader import Loader

loader = Loader(root)
await loader.include("cordis.yml")
# 修改配置后调用 reconcile，仅重建发生变化的条目
await loader.reconcile()

4.4 热模块替换（HMR）

模块分类：对变更模块集合计算 accepted/declined 集合（模块导入边由运行时追踪器 + AST 静态补全两路构成）。
过期条目检测：找出 import 依赖树中触及 accepted 模块的 Loader 条目。
事务式重载：快照旧插件与模块命名空间，dispose 旧 fiber，按拓扑序重载变更模块与条目模块，再重新 apply；任一步失败则回滚并重抛。

from cordis_py import HMR

hmr = HMR(loader)
await hmr.reload_file("src/plugins/worker.py")   # 变更文件 → 自动分类并事务式重载
await hmr.reload_module("myapp.helpers")         # 或按模块名手动触发
hmr.dispose()                                    # 退出时卸载依赖图追踪器

5. 实现路线

阶段	内容	验收标准
Phase 0	初始化 uv 工程、src 布局、py.typed、pytest、基本导出	`uv run pytest` 通过，`import cordis_py` 可用。
Phase 1	Context + effect 账本 + Disposable 组合	同步/异步 disposer、生成器效果可倒序回收。
Phase 2	Fiber 生命周期、依赖注入、服务注册与通知	先装消费者后装提供者能自动启动；卸载提供者触发依赖方回滚。
Phase 3	事件系统：on/once、四种异步模式和 waterfall	监听器随 fiber 自动清理；模式语义测试通过。
Phase 4	Service 基类、isolate/intercept、accessor/mixin、访问控制	多 realm 服务互不干扰；未声明访问抛错。
Phase 5	声明式 Loader：Entry、Group、Include、YAML/JSON、reconcile	增删改配置只影响相关 fiber，且最终状态等于全新加载。
Phase 6	HMR：文件监听、模块分类、过期检测、事务重载	修改组件源码后无需重启进程，旧副作用被清理。
Phase 7	文档、日志、计时器等扩展，性能与并发压力测试	提供完整 README 和 API 参考。

建议首版范围：先完成 Phase 0-4 作为 cordis-py 核心包；Phase 5-6 可作为同一包内的可选子模块或独立扩展包，避免核心依赖过重。

TODO / 实现状态

此节用于记录当前 Python 实现已经完成和尚未完成的功能点，避免后续迭代时遗忘。

已完成

Context / Fiber 核心运行时与生命周期状态机
Effect 追踪与 LIFO 清理
服务注册、依赖注入与响应式依赖
事件系统：on / once / emit / parallel / serial / bail / waterfall
Service 基类与 @inject 装饰器
按 realm 的服务隔离
intercept 服务级配置拦截
声明式 Loader：JSON / YAML / TOML、reconcile、disable / enable
HMR 依赖图分类：模块边追踪、accepted/declined 分类、事务式重载与回滚
Python entry points 插件发现
完整同步/异步双模式封装
单元测试与可运行 demo
服务版本约束、接口校验与配置校验
配置 overlay / 租户派生（internal/config 瀑布链）

待办 / 后续功能点

跨进程 Bridge / 远程服务
沙箱与不可信插件隔离
HMR 文件监听器与 watcher 集成（watchdogs 或手动触发）

实现依据说明：本项目仅参考论文《A Programming Paradigm for Spatiotemporal Composability》与 Node.js 版 Cordis（Cordis v4 / DeepSeek Harness vendor）进行设计，不参考任何现有的 Python Cordis 实现包。

6. 打包与工程化

仓库已使用 uv 和 Python 3.12。建议按以下方式组织 pyproject.toml：

[project]
name = "cordis-py"
version = "0.1.0"
description = "A Python implementation of Cordis: spatiotemporal composability for dynamic systems"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
dependencies = [
  # 核心无第三方依赖；可选扩展按 extra 引入
]

[project.optional-dependencies]
yaml = ["pyyaml>=6.0"]
hmr = ["watchfiles>=0.21"]
dev = ["pytest>=8", "pytest-asyncio>=0.23", "ruff>=0.6", "mypy>=1.10"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/cordis_py"]

常用命令：

# 安装开发环境
uv sync --extra dev

# 运行测试
uv run pytest

# 构建 wheel / sdist
uv build

# 本地安装为包
uv pip install -e .

7. 风险、取舍与待确认问题

Python 属性代理能力弱于 JS Proxy：__getattr__ 无法完全模拟 has 等陷阱，也无法拦截所有特殊属性。计划先在发布 API 层保证语义，内部真实存储用显式 _store，避免污染。
异步生命周期复杂度：论文的“惯性”要求 transition 期间新请求不能打断，需要仔细处理 inertia 与取消/并发。建议用单一事件循环任务队列和状态机单测覆盖。
模块热替换的 Python 局限：importlib.reload 不会自动清理旧模块的所有引用，且对扩展模块、C 扩展、已实例化类不友好。HMR 先面向纯 Python 开发场景，并明确“不可热替换则全量重启”的边界。
配置校验：Node 版使用 Standard Schema。Python 可先支持 pydantic 或轻量 callable 校验器，避免强依赖。
命名与兼容：本 Python 实现不参考任何现有的 Python Cordis 包，命名以 Node 版 Cordis 为参照；后续发布时应保持公共 API 稳定，并记录破坏性变更。
是否首版包含 Loader/HMR：若用户只想要核心运行时，可拆成 cordis-py 与 cordis-py-loader 两个包，与 Node 生态保持一致。