从源码理解Cordis-Core

前言

最近 DeepSeek 开源了 DSH，Harness + Agent 的架构可能会成为后面一个比较重要的方向.
DSH 底层用了 Cordis,Cordis 是一个元框架，拥有高可组合性的插件系统.

所以想顺着 DSH 看一下 Cordis 的源码.

Context

是什么,负责什么?

Context 是每个插件各自持有的作用域句柄，也是该插件与框架交互的入口.
Context 可以不断派生，因此多个 Context 之间通过原型链共享上层环境，同时又可以拥有自己的隔离和拦截状态.

查看 Context 接口

export interface Context {
  [symbols.isolate]: Dict<symbol>
  [symbols.intercept]: Dict
  /** @experimental */
  root: this
  baseUrl?: string
  events: EventsService
  logger: LoggerService
  reflect: ReflectService
  registry: RegistryService
}

• baseUrl 模块/资源解析时使用的基础路径
• root 当前 Context 所属的根 Context
• isolate 记录当前 Context 的隔离信息
• intercept 记录当前 Context 的拦截信息
• Events Context 对应的事件系统
• Registry 服务的注册与管理
• Reflect 提供运行时反射/依赖关系相关能力
• Logger 日志服务
• Fiber 插件实例

Context是全局单例吗?

不是, 但会有一个 根Context.
之后每在某个 ctx 上挂插件（或 isolate / intercept），都会派生出一个子 Context.插件拿到的就是这份子 Context，而不是去改根.
于是整个应用形成一棵 Context 树

root ctx          ← new Context()
 ├─ 插件 A 的 ctx  ← extend，fiber = A
 │   └─ 插件 C 的 ctx
 └─ 插件 B 的 ctx

看下 Context 中关于 extend 的部分

 extend(meta = {}): this {
    const shadow = Reflect.getOwnPropertyDescriptor(this, symbols.shadow)?.value
    const self = Object.create(getTraceable(this, this))
    for (const prop of Reflect.ownKeys(meta)) {
      Object.defineProperty(self, prop, Reflect.getOwnPropertyDescriptor(meta, prop)!)
    }
    if (!shadow) return self
    return Object.assign(Object.create(self), { [symbols.shadow]: shadow })
  }

Shadow 和 Tracker 暂时跳过.
它们负责 Context 派生过程中的追踪与影子状态传递，属于框架为了实现运行时追踪而增加的辅助机制.它们不影响我们理解 Context 的基本模型.

上面的核心就是 for循环 ,将 meta 中的属性都复制到派生的 Context 中.
派生后的Context只包含部分属性,调用不存在的属性时,会一级一级往上往上查找.

Fiber

是什么,负责什么

Fiber 是一个插件挂载的运行实例.
管这条插件从启动到卸载的整段生命周期，以及期间登记的所有可逆副作用.

看下 Fiber 的属性

• uid 标识这个 Fiber 实例
• parent 是父 Context,表示从哪个 父Context 进行派生当前插件的 Context.
• ctx / context 插件实际运行时拿到的环境
• inject 插件依赖哪些服务
• runtime 描述这个 Fiber 要运行哪个插件、怎么运行
• config 本次插件运行使用的配置
• runner / _runner 真正驱动插件执行、控制执行状态
• effect 注册插件运行期间产生的资源
• dispose 结束这个 Fiber，并触发资源清理
• state 插件状态
• store / _store 保存 inject 对应的实际服务实现

怎么创建出来的

Cordis 中只有两个地方调用了 Fiber.

Context中的构造函数, 只有 根Context 会调用到构造函数.
传入的都是空值.

this.fiber = new Fiber(self, {}, Object.create(null), null, () => [])

Registry 中的 plugin 函数
这边会传递插件配置,runtime,依赖.

const fiber = new Fiber(this.ctx, config, Inject.resolve(plugin.inject), runtime, getOuterStack)

回到 Fiber 构造函数,只看有 runtime 这部分.

设置uid, counter 中每次 get 递增1,所以每次启动都会有不同的 uid.
从 父context 派生 ctx ,并将当前 fiber 传入
保存依赖的服务

this.uid = parent.registry.counter
this.ctx = this.context = parent.extend({ fiber: this })

const injectEntries = Object.entries(this.inject)
if (injectEntries.length) {
 this.ctx[Context.intercept] = Object.create(parent[Context.intercept])
 for (const [name, config] of injectEntries) {
   if (isNullable(config)) continue
   this.ctx[Context.intercept][name] = config
 }
}

实例化 runner ,从这里可以看出 _runner 的 execute 实际调用的就是runtime.callback

this._runner = {
 epoch: INACTIVE,
 getOuterStack,
 execute: function () {
   if (isConstructor(runtime.callback)) {
  const instance = new runtime.callback(this.ctx, this.config)
  for (const hook of instance?.[symbols.initHooks] ?? []) {
    hook()
  }
  return instance?.[symbols.init]?.()
   } else {
  return runtime.callback(this.ctx, this.config)
   }
 },
 collect,
  }

这里就是在检查依赖的服务是否已经启动成功

for (const name of Object.keys(this.inject)) {
 this._checkImpl(name)
  }

卸载函数,用于 parent 卸载时,将这次挂载也卸载掉.
这里要注意, return 里面的才是实际卸载时会执行的.
在之前的会在构造函数里执行.
配置就是在这里加载的.

this.dispose = parent.fiber.effect(() => {
   const remove = runtime.fibers.push(this)
   try {
     this.config = resolveConfig(runtime, config)
     this._refresh()
   } catch (error) {
     this.ctx.logger.error(error)
     this._error = error
   }
   return async () => {
     this.uid = null
     this.context.emit('internal/plugin', this)
     if (this.ctx.registry.has(runtime.callback)) {
    remove()
    if (!runtime.fibers.length) {
      this.ctx.registry.delete(runtime.callback)
    }
     }
     this._setEpoch(INACTIVE)
     while (this.inertia) {
    await this.inertia
     }
   }
 }, 'ctx.plugin()')

就是如果依赖的服务一开始没启动会怎么样

在构造函数里有执行 _checkImpl.
前面说了 ctx 是由 parent 派生的,只传入了 fiber.
所以这里的 reflect 会一层一层向上找到 根部context 的 reflect.

这里判断依赖的服务是否已经启动,启动则保存到 _store 中.

_checkImpl(name: string) {
    const impl = this.ctx.reflect._getImpl(name, true)
    if (!impl) return delete this._store[name]
    try {
      if (impl.check && !impl.check.call(getTraceable(this.ctx, impl.value))) {
        return delete this._store[name]
      }
    } catch (error) {
      impl.fiber.ctx.logger.error(error)
      return delete this._store[name]
    }
    this._store[name] = impl
  }

在 dispose 中调用的 _refresh.
这里判断是否所有的依赖服务都启动成功,并以此设置状态.

_refresh() {
    let epoch: string | boolean = false
    epoch = ''
    for (const name of Object.keys(this.inject)) {
      const impl = this._store[name]
      if (!impl) {
        epoch = INACTIVE
        break
      }
      epoch += ':' + impl.fiber.uid
    }
    this._setEpoch(epoch)
  }

在 _setEpoch 中会调用 _updateState,直接看下 _updateState.

this.ctx.reflect.store 保存的是所有的服务,不论是否启动成功.
会判断如果状态发生变化会进行通知.
在 notify 会再一步过滤,只通知依赖了当前服务的服务.

private _updateState(callback: () => void | FiberState) {
    const oldState = this.state
    this.state = callback() ?? this._getState()
    if (oldState === this.state) return
    this.context.emit('internal/status', this, oldState)

    if (oldState !== FiberState.ACTIVE && this.state !== FiberState.ACTIVE) return
    for (const key of Reflect.ownKeys(this.ctx.reflect.store)) {
      const impl = this.ctx.reflect.store[key as symbol]
      if (impl.fiber !== this) continue
      this.ctx.reflect.notify([impl.name])
    }
  }

卸载流程怎么样的

在构造函数中 dispose 会调用 effect 函数并传入一个 execute .

在 effect 函数会用 execute 生成一个 EffectRunner.
它和插件 _runner 是同一种类型.
最终都是要通过 _execute 实际调用.

看下_execute 的代码.
逻辑都是把 execute 执行后返回的 effect 调用 collect 存起来.

private _execute<T>(runner: EffectRunner<T>) {
    const oldEpoch = runner.epoch
    return composeError((info) => {
      const safeCollect = (dispose: void | Disposable) => {
        if (typeof dispose === 'function') {
          runner.collect(dispose)
        } else if (!isNullable(dispose)) {
          throw new TypeError('Invalid effect')
        }
      }
      const effect: Effect = runner.execute.call(this)
      if (typeof effect === 'function') {
        return runner.collect(effect)
      } else if (isNullable(effect)) {
        // return
      } else if (!isObject(effect)) {
        throw new TypeError('Invalid effect')
      } else if ('then' in effect) {
        return effect.then(safeCollect)
      } else if (Symbol.iterator in effect) {
        info.error = new Error()
        const iter = effect[Symbol.iterator]()
        while (true) {
          const result = iter.next()
          safeCollect(result.value)
          if (result.done) return
        }
      } else if (Symbol.asyncIterator in effect) {
        const iter = effect[Symbol.asyncIterator]()
        return (async () => {
          // force async stack trace
          await Promise.resolve()
          info.error = new Error()
          while (true) {
            if (runner.epoch !== oldEpoch) return
            const result = await iter.next()
            safeCollect(result.value)
            if (result.done) return
          }
        })()
      } else {
        throw new TypeError('Invalid effect')
      }
    }, runner.getOuterStack)
  }

前面说了 execute 会把 effect 收集起来.
这里是收集到 disposables 中.
在 dispose 会对disposables逆序进行调用卸载.
为啥要逆序?
是为了避免后面挂载的资源可能依赖前面的资源.
如果先注册的被卸载了,后注册的卸载可能有问题.

effect(execute: () => Effect, label = 'anonymous'): any {
    this.assertActive()

    const disposables: Disposable[] = []
    const dispose = () => {
      let task!: void | Promise<void>
      for (const dispose of disposables.splice(0).reverse()) {
        if (task) {
          task = task.then(dispose)
        } else {
          const result = dispose()
          if (isObject(result) && 'then' in result) {
            task = result as any
          }
        }
      }
      return task
    }

    const meta: EffectMeta = { label, children: [] }
    const runner: EffectRunner<boolean> = {
      execute,
      epoch: true,
      collect: (dispose) => {
        disposables.push(dispose)
        this._disposables.delete(dispose)
        if (dispose[symbols.effect]) {
          meta.children.push(dispose[symbols.effect])
        }
      },
      getOuterStack: buildOuterStack(),
    }

    let task: void | Promise<void>
    try {
      task = this._execute(runner)
    } catch (reason) {
      dispose()
      throw reason
    }

这段也是 effect 中的代码.
task 是上面 _execute 的结果.
这里针对的是返回值是Promise<void></void>的情况.
第一个 catch 是说 Effect 的异步执行失败了，那就执行 dispose().
第二个 catch 是说 dispose() 也执行失败了,那就兜底打印日志.
这里要注意,直到这里也还没调用 dispose(), 只是注册了错误处理器

接着在 dispose 外面又包裹了一层当成 wrapper.
再当返回值返回.
为啥包裹一层?使用 epoch 避免多次调用.

    task?.catch(dispose).catch((error) => this.ctx.logger.error(error))

    const wrapper = defineProperty(() => {
      if (!runner.epoch) return
      runner.epoch = false
      return task ? task.then(dispose) : dispose()
    }, symbols.effect, meta) as AsyncDisposable

    const disposeAsync = () => {
      if (!runner.epoch) return
      runner.epoch = false
      return dispose()
    }
    wrapper.then = async (onFulfilled, onRejected) => {
      return Promise.resolve(task)
        .then(() => disposeAsync)
        .then(onFulfilled, onRejected)
    }
    disposables.push(this._disposables.push(wrapper))
    return wrapper
  }

Service、Reflect 与 Registry

Service 是什么,负责什么

service 是一个基类,提供可被别的插件按名字使用的能力.
它可以将服务直接挂载到 Context上, 其他插件可以直接通过名字调用该插件.

直接看下构造函数.
name ??= this.constructor['provide'] as string 判断是否有传入 name ,如果没有则取服务的provide做为name.
service 是基类, this.constructor 指向的是继承它的子类.

tracker 主要用于内部追踪.

关键是 self.ctx.reflect.provide(name, self, this[symbols.check])
实际也是调用 reflect 将服务和服务名绑定注册在 根context的 reflect 上.

constructor(protected ctx: Context, name: string) {
    name ??= this.constructor['provide'] as string

    let self = this
    const tracker: Tracker = {
      associate: name,
      property: 'ctx',
    }
    if (self[symbols.invoke]) {
      self = createCallable(name, joinPrototype(Object.getPrototypeOf(this), Function.prototype), tracker)
    }
    self.ctx = ctx
    self.name = name
    defineProperty(self, symbols.tracker, tracker)

    self.ctx.reflect.provide(name, self, this[symbols.check])
    return self
  }

Reflect 是什么,负责什么

ReflectService 是 Cordis 的 Context 属性与 Service 运行时管理中心.
负责把 Context 上的属性访问映射到 Service / Accessor.
同时负责 Service 的注册、查找、更新和依赖通知.

回到 context 的构造函数中,有两条很关键的语句
这里为 context 创建代理.
对 context 中属性的访问,会走到ReflectService.handler中,再由他进行处理.

const self = new Proxy<this>(this, ReflectService.handler)
this.root = self

Reflect 如何映射属性

通过 get,set,has 三个方法对要访问的属性进行处理.
通过 mixin 将额外的属性/方法扩展到 context 中.

get

直接看get的函数体.
if (isSpecialProperty(prop)) 如果是特殊属性,直接使用JS原生,不解析,例如

ctx._xxx
ctx[symbol]
ctx.then
ctx.prototype
ctx[0]

if (Reflect.has(target, prop)) 对于 Context 已有的属性,
getTraceable 以后再返回.例如

ctx.events
ctx.logger
ctx.reflect
ctx.registry

如果前面的属性都访问不到.则从当前的实例的 props 进行获取.
对获取到的属性进行类似判断,判断是 accessor 还是 service.

如果获取的属性类似是 accessor.
直接调用该属性的get

if (def?.type === 'accessor') {
  return def.get.call(ctx, ctx[symbols.receiver], error)
}

在 fiber 的时候说过, 根context 创建的 fiber没有 runtime.
这里是对 根context 的处理

 (!ctx.fiber.runtime) return ctx.reflect.get(prop, false)

接着就是类型是 Service 的处理.
这里涉及到 event 模块.
暂时记住这样写,会允许其他机制对 Service 属性 get 进行拦截/修改.
最后的函数是默认调用.也就是没有被拦截就走这里进行处理.

fiber.store 在前面说过,保存的是 fiber依赖的服务实现.
while 循环去找它依赖的服务中已经准备好的服务.如果找到直接返回.

如果属性在依赖的服务但是没有准备好直接抛出错误.

if (fiber.parent[symbols.isolate][prop] !== key) throw error这句用到symbols.isolate进行判断,暂时记住这是用于 Context 隔离模型的.
这里是确保 父fiber 的 prop 的symbols.isolate 必须和当前 fiber 的一致.
不一致则直接抛出错误.

如果当前 fiber 还是找不到就往上到父级 fiber 找

return ctx.events.waterfall('internal/get', ctx, prop, error, () => {
  const key = target[symbols.isolate][prop]
  let fiber = (ctx[symbols.shadow] as Context ?? ctx).fiber
  while (true) {
 const impl = fiber.store?.[prop]
 if (impl) return getTraceable(ctx, impl.value)
 if (prop in fiber.inject) {
   error.message = `cannot get required service "${prop}" in inactive context`
   throw error
 }
 if (!fiber.runtime) throw error
 if (fiber.parent[symbols.isolate][prop] !== key) throw error
 fiber = fiber.parent.fiber
  }
})

mixin

mixin 更准备的作用是将 source 中的属性直接映射到 context.可以直接通过 context访问.

在构造函数中可以看到类似 this.mixin('fiber', ['runtime', 'effect'])
这样就可以通过 context.runtime 直接访问 context.fiber.runtime.

mixin 会调用 accessor, 先看这部分.
accessor 的作用是将属性注册到 this.props 中, 同时返回一个“清理函数”用于删除该属性.
这里会将属性的类型注册成 accessor 类型.
然后如果直接访问的话就会走上面 get 小节的判断,再调用 mixin 这边传入的 get 方法.
this.ctx.fiber.effect 卸载的时候说过,算个副作用管理器.用于清理操作.
这部分后面估计要转门讲下.内容太分散.

accessor(name: string, options: Omit<Property.Accessor, 'type'>) {
    return this.ctx.fiber.effect(() => {
      if (name in this.props) {
        throw new Error(`property "${name}" is already declared as ${this.props[name].type}`)
      }
      this.props[name] = { type: 'accessor', ...options }
      return () => delete this.props[name]
    }, `ctx.accessor(${JSON.stringify(name)})`)
  }

mixin 要讲的话又要展开 Effect 的副作用管理器,放到后面吧.
暂时知道是卸载服务的时候用于清理工作.

  mixin(source: any, mixins: string[] | Dict<string>) {
    const self = this
    return this.ctx.fiber.effect(function* () {
      const entries = Array.isArray(mixins) ? mixins.map(key => [key, key]) : Object.entries(mixins)
      const getTarget = (ctx: Context, error: Error) => {
        // TODO enhance error message
        return ctx[source]
      }
      for (const [key, value] of entries) {
        yield self.accessor(value, {
          get(receiver, error) {
            const service = getTarget(this, error)
            if (isNullable(service)) return service
            const mixin = receiver ? withProps(receiver, service) : service
            const value = Reflect.get(service, key, mixin)
            if (typeof value !== 'function') return value
            return value.bind(mixin ?? service)
          },
          set(value, receiver, error) {
            const service = getTarget(this, error)
            const mixin = receiver ? withProps(receiver, service) : service
            return Reflect.set(service, key, value, mixin)
          },
        })
      }
    }, `ctx.mixin(${JSON.stringify(source)})`)
  }

set 和 has都比较简单,不展开了.

Reflect 如何将 Service 提供出来

关于如何注册和提供服务的部分,都在 provide 中
这部分一目了然,在前面讲 service 的时候也带过了.

大概流程如下

 Service 启动
     ↓
 provide
     ↓
 创建 / 更新 Impl
     ↓
 Reflect.store
     ↓
 notify
     ↓
 依赖它的 Fiber
     ↓
 _refresh()

Registry 是什么，负责什么

Registry 是 Cordis 中管理插件定义与运行实例的中心.
它负责记录插件对应的 Runtime，并通过 plugin() 创建和管理插件的 Fiber

Registry 怎么创建插件的

插件都是通过 Registry 中的 plugin 进行插件的创建.
里面的代码蛮简单的,不分析,画个大概流程.

ctx.plugin(A)
      ↓
Registry.plugin()
      ↓
找到 / 创建 Runtime
      ↓
解析 inject
      ↓
new Fiber(...)
      ↓
Fiber.effect(...)
      ↓
插件进入运行状态

三者之间的关系

先总结下:
Registry 负责把插件装进系统（Runtime / Fiber）；
Reflect 负责把插件贡献的能力挂到 Context 上并供依赖方查找；
Service 是提供这些能力时的标准基类，构造时通过 Reflect 注册自己。

           ctx.plugin / Loader
                   │
                   ▼
            ┌─────────────┐
            │  Registry   │  Runtime（定义）
            │             │  plugin() → 创建 Fiber
            └──────┬──────┘
                   │
                   ▼
            ┌─────────────┐
            │    Fiber    │  inject 就绪后执行插件
            └──────┬──────┘
                   │
      ┌────────────┼────────────────┐
      ▼            ▼                ▼
   Effect      Service 等      其它 Reflect API
on/effect…   provide(...)     mixin / accessor
      │            │                │
      │            ▼                ▼
      │         store             props
      │            └───────┬────────┘
      │                    ▼
      │             ┌─────────────┐
      │             │   Reflect   │  + Proxy
      │             └──────┬──────┘
      │                    │
      ▼                    ▼
生命周期清理          ctx.xxx（读服务 / mixin）

Event

Event 是 Context 内部的消息通知机制.
一个组件发布事件，其他组件订阅事件，在事件发生时执行对应的回调.

五种派发模式有什么不同

isBailed 的判定是：返回值 不是 null / false / undefined 就算「有效」，于是 serial / bail 会提前返回。

模式	同步/异步	执行方式	返回值	停止条件
emit	同步	依次调用所有 listener	void	不会主动停止；listener 抛异常会直接中断
parallel	异步	并行调用所有 listener	Promise<void>	等待全部结束；最后统一处理异常
serial	异步	依次 await listener	第一个有效返回值	isBailed(result) 为真
bail	同步	依次调用 listener	第一个有效返回值	isBailed(result) 为真
waterfall	同步	链式 next() 调用	最终 listener / default 的返回值	某个 listener 不调用 next() 就停止

waterfall 是这里面最特殊的,要单独讲下.但开始前要先讲 _resolve
_resolve 是 Events 的核心函数,看下它的实现.

this.args 支持 ctx.emit('foo', a, b) 和 ctx.emit(service, 'foo', a, b)两种形式.
第一句解析出实际的 this.args
this.emit('internal/dispatch', type, name, args, thisArg) 用于在派发非internal事件前,给其他想监听的人一个钩子.应该是用于调试的.
最后对事件进行 context 过滤.
对于同一个事件名,但是分属不同的 context,可以过滤只派发指定 context 上的.
如果 global 为 true 的话则派发全部.

private _resolve(type: string, args: any[]) {
    const thisArg = typeof args[0] === 'object' || typeof args[0] === 'function' ? args.shift() : null
    const name: string = args.shift()
    if (!name.startsWith('internal/') && this._hooks['internal/dispatch']?.length) {
      this.emit('internal/dispatch', type, name, args, thisArg)
    }
    const filter = thisArg?.[Context.filter]
    return [thisArg, (this._hooks[name] || [])
      .filter(hook => hook.global || !filter || filter.call(thisArg, hook.ctx)).map(hook => hook.callback)] as const
  }

其他派发模式,找出所有的监听器,for循环调用 callback 派发出去完事.
看下调用方式可以看到 waterfall 最后一个参数是个函数.
waterfall 和其他派发模式不一样的点在于它把最后一个参数抠出来当 inner，再给每个 listener 注入 next.
这里比较绕.要先记住这里的 waterfall 是派发,不是 listener.listener 是 on 和 once.

waterfall 在这里调用了一次 next(),会走 Reflect.apply(callback, thisArg, args).
这时候流程会走到 on 和 once 的 callback . next() 会当成参数也传递给 callback.
此时的控制权已经到了 on 和 once 的 callback.
在callback它可以选择是否继续执行 next().
如果不执行则直接中断不再执行同一事件后面的 listener 的 callback.

简单来说,就是其他派发器会循环执行所有callback,waterfall 可以拦截中断 callback循环

ctx.events.waterfall('internal/set', ctx, prop, value, error, () => {
  return ctx.reflect.set(prop, value, error)
})
ctx.on('loader/patch-context', (entry, next) => {
    ~~~
    next()
 ~~~
  })
waterfall(...args: any[]) {
    const [thisArg, callbacks] = this._resolve('waterfall', args)
    const inner = args.pop()
    const next = () => {
      const callback = callbacks.shift()
      return callback ? Reflect.apply(callback, thisArg, args) : inner(...args)
    }
    args.push(next)
    return next()
  }

和其他的事件机制有啥不一样的

会自动自动卸载,监听后不需要手动撤销监听.
注册事件的时候用的 effect , Fiber dispose 的时候会自动调用 unregister 撤销监听

register(label: string, hooks: Hook[], callback: any, options: EventOptions): () => void {
    const method = options.prepend ? 'unshift' : 'push'
    return this.ctx.fiber.effect(() => {
      hooks[method]({ ctx: this.ctx, callback, ...options })
      return () => this.unregister(hooks, callback)
    }, label)
  }

派发增加上下文过滤.
thisArg 如果有 Context.filter, 只会通知通过 filter 的 hook,不包括 global: true
具体代码在上面的 _resolve

派发类型多,正常框架可能就一两种,上面列出有5种.