Cordis五概念:插件框架的发动机|dsh源码篇EP02

$ ls ~/.dsh/profiles/
ls: No such file or directory      ← 第一次敲 dsh web 之前：目录不存在
$ dsh web
dsh web: http://127.0.0.1:3080     ← 首次使用自动从随包模板初始化
$ ls ~/.dsh/profiles/
web/                               ← 框架自己生成的（headless 同理）

constweb=program.command('web').description('boot the web profile ...')
web
// web 的参数原样透传给网页应用（--port 这些它不管）
  .argument('[args...]', 'arguments for the web app (see: dsh web --help)')
  .option('--patch <path>', 'extra patch-list overlay ... (repeatable)', collect)
  .option('--dump-config', 'print the composed web-profile tree ... and exit')
  .option('--dump-default-config', "print the web profile's bundle layers ...")
  .action((args:string[], options:BootOptions) => {
// web 不吃父命令的 --profile 等选项——给了就报错：
// 因为 web 自己就是 profile 名，重复指定是矛盾
rejectParentOptions('web')
// 归一成一个启动请求：模式 profile、名字 web。
// dsh web ≡ dsh --profile web，别名关系在这落地
resolved=resolveBoot(web, 'web', options, args)
  })

// 解析启动器自己的参数：--profile、--patch、--help 这些
// 无效命令、用错模式的选项，在这里就非零退出
constinvocation=parseDshArgs(process.argv.slice(2), readVersion())
switch (invocation.mode) {
case'profile': {
// 按模式动态 import 对应 runner——web 模式的启动路径里
// 根本不会加载别的模式的代码
const { runProfile } =awaitimport('./profile-boot.ts')
awaitrunProfile({
// 分层环境变量、选中的 profile、--patch 文件、剩余参数
environment:loadLayeredEnv('dsh'),
profile:invocation.profile,
patchFiles:invocation.patches,
args:invocation.args,
    })
break
  }
case'plugin': {
// 插件管理：转发给 pnpm。退出码直接透传——装失败了 dsh 也失败
const { runPlugin } =awaitimport('./plugin.ts')
process.exit(runPlugin(invocation.profile, invocation.args))
break
  }
case'dump-config': {
// 打印合成配置树（EP01 用过的那条命令的出处）
const { runDumpConfig } =awaitimport('./dump-config.ts')
runDumpConfig(invocation.profile, invocation.defaultOnly, invocation.patches)
break
  }
default:
// TypeScript 穷尽检查：将来加新模式而这里忘了处理，
// 编译期就报错，不会静默漏网
invocationsatisfiesnever
thrownewError(`dsh: unhandled invocation mode ${JSON.stringify(invocation)}`)
}

exportasyncfunctionrunProfile(options:RunProfileOptions) {
// 第一步：读你的 profile 目录，把四层配置叠好
constcomposed=composeProfile(options.profile, options.patchFiles)
// ...保险与信号的装配（第 8 站展开）
constctx=awaitboot(NAME, rootConfig, patches, (hostCtx) => {
// boot 会在建好根上下文后、挂任何插件之前调这个回调，
// 让你先往上下文里放启动时的数据。放两样：
// ① 环境变量快照——provide 把值挂到名字上让插件按名取用
//    （存值 + 通知等待者 + 登记撤销，源码在第 2 站拆）
hostCtx.provide(DSH_LAUNCH_ENVIRONMENT_KEY, options.environment)
// ② 命令行参数——你在 dsh web --port 3000 里写的
//    --port 3000 不归启动器管，它只是原样带到这，
//    交给树里的应用插件自己读。
//    exit 是一个受控退出函数：插件想退出时调它，
//    走的是 shutdown 流程（先拆树再退），不是 process.exit
provideCmdline(hostCtx, {
args:options.args,
exit:code => voidshutdown.shutdown(code),
    })
  })
app.current=ctx
// ...热重载监听的装配（第 7 站展开）
}

args 归一 → bin 分发 → runProfile 备场 → boot 建上下文
→ Loader 遍历 → plugin() 建档 → 上架唤醒

asyncfunctionboot(binName, path, patches, prepare) {
// 1. 建根上下文（插件的工位）
constctx=newContext()
// 2. 挂 Loader——加载器自己是第一个被挂的插件
awaitctx.plugin(Loader)
// 3. 跑你的回调：注入环境快照和命令行参数
awaitprepare?.(ctx)
// 4. 挂根 include 行（含三摞补丁）
awaitmountRootInclude(ctx, path, patches)
// 5. 等所有行挂完才返回
awaitctx.get("loader")?.await()
}

plugin(plugin, config) {
// 校验：必须是函数或带 apply 方法的对象
constcallback=this.resolve(plugin)
// 建档案：同一段代码可挂多个实例，配置各管各的
letruntime=this._internal.get(callback)
if (!runtime)
runtime= { name:plugin.name,
callback, fibers:newDisposableList() }
// 建跟踪档案：inject 在这一刻被解析
constfiber=newFiber(ctx, config,
Inject.resolve(plugin.inject), runtime)
// 包一层 then——可以 await，boot 靠它等插件就绪
returnmakeAwaitable(fiber)
}

provide(name, value, check) {
returnthis.ctx.fiber.effect(() => {
// 存值：把 { 名字, 值, 归属fiber, check } 放进货架
this.store[key] = { name, value, fiber, check }
// 通知：告诉等这个名字的 fiber——东西来了
// → 它们的状态从 PENDING 被推进
// state 2 = ACTIVE 时才发通知
if (this.ctx.fiber.state===2)
this.notify([name])
// 返回撤销函数（effect 记下来，卸载时自动调）
returnasync () => {
// 从货架撤掉
deletethis.store[key]
// 再通知一轮
constfibers=this.notify([name])
await Promise.allSettled(fibers.map(f => f.await()))
    }
  })
}

exportfunctionapplyEntryPatches(data, patches, warn) {
data=structuredClone(data)
// 建 id → 行 的索引（含嵌套组）
constentryMap=new Map()
buildMap(data)
for (constpatchofpatches) {
const { id, insert, ...overrides } =patch
if (insert) {
// 插入行：带 id 插进那个组，不带 id 追加到列表尾
//（你的自定义插件走这条路）
if (id) entryMap.get(id).config.push(...insert)
elsedata.push(...insert)
// 新插的行也索引进来，后到的补丁能改它
buildMap(insert)
    } else {
// 覆盖：按 id 找到目标行，overrides 盖上去
// 后写胜出 = 循环从先到后，后面的覆盖前面的
      Object.assign(entryMap.get(id), overrides)
    }
  }
returndata
}

privateasync_init() {
// 第一步：import 包名 → 插件模块（带 apply 方法的对象）
plugin=this.loader.unwrapExports(
awaitthis.parent.tree.import(this.options.name))
// import 失败：错误信息带上清单行位置
// 第二步：挂载——调上面讲过的 ctx.plugin()
awaitthis._start(plugin)
}
privateasync_start(plugin) {
fiber=this.ctx.registry.plugin(plugin, this.options.config)
// 等 fiber 就绪
awaitfiber.await()
// 失败即回收：刚建的 fiber 当场拆掉，不留残缺
}

plugin= {
apply(ctx) { /* HTTP 服务器的启动逻辑 */ },
}

asyncawait() {
while (this.inertia) awaitthis.inertia
if (this._error) throwthis._error
returnthis
}

constwrapped= Object.create(fiber)
wrapped.then= (ok, fail) => fiber.await().then(ok, fail)

exportconstenumFiberState {
// 主路：等待 → 执行 → 在岗
PENDING,
LOADING,
ACTIVE,
// 岔路：出错 / 已移除不可重启 / 收尾执行中
FAILED,
DISPOSED,
UNLOADING,
}

// 前提：自己没被移除，父上下文也不在卸载（省略其余防御）
if (this.uid!==null&&parent.fiber.state!==FiberState.UNLOADING) {
// 逐个核对声明过的依赖名：那个服务真的上架了吗
for (constnameof Object.keys(this.inject)) {
this._checkImpl(name)
  }
// 根据核对结果推进状态：全齐进 LOADING，否则继续等
this._refresh()
}

notify(names) {
for (constfiberofallFibers) {
if (!(nameinfiber.inject)) continue
// ↑ 不依赖这个服务的 fiber，到这里就跳过
fiber._checkImpl(name)
fiber._refresh()
  }
}

[2ms]    B 已注册（它声明要等 svcA）——此刻 B 的 apply 还没跑
[155ms]  150ms 后：注册 A
[156ms]  A 构造，服务 svcA 准备上架
[156ms]  B 的 apply 终于跑了，此时 ctx.svcA = object

classWeatherServiceextendsService {
// 声明键位：挂上后别人用 ctx.weather 调
static [Service.provide] ='weather'
constructor(ctx) {
// 这一行就完成上架
super(ctx, 'weather')
  }
// 别人调 ctx.weather.get('北京') 时执行
get(city) { returnfetch(`/api/weather/${city}`) }
}

- id: weather
name: ./weather.ts

// 第一步：起名（决定挂到哪个键位）
// 没传就取 provide 字段 → 'weather'
name??=this.constructor['provide']
// 第二步：上货架（provide = 挂到名字上让别人找到）
ctx.reflect.provide('weather', this, check)
//                 ↑ 键位     ↑ 实例  ↑ 就绪检查
//
// provide 内部（reflect.ts:277，省略隔离键细节）：
// 1. 冲突检测：同名已被注册 → 报错
// 2. 存值：{ 名字, 值, fiber, check } 放进货架
// 3. 通知：告诉等这个名字的 fiber——东西来了
//    → 它们的状态从 PENDING 被推进
// 4. 返回清理函数（卸载时自动调）：
//    撤掉值 + 再通知一轮 + 等依赖方消化完
//
// 外面包的 ctx.fiber.effect()：
// 执行你的代码 + 把返回的清理函数记下来，
// 插件卸载时自动调——你不用手动清理
// 第三步：上架后发通知
// → 所有 inject: ['weather'] 的 fiber 被唤醒
// → 它们的 apply 开跑，可以用 ctx.weather 了

classDatabaseServiceextendsService {
static [Service.provide] ='database'
privatepool:Pool|null=null
constructor(ctx) {
super(ctx, 'database')
// 构造只是启动连接，不等完成
createPool().then(p => { this.pool=p })
  }
// 框架会调这个函数问「就绪了吗」
// 返回 false 时，依赖 database 的插件不会启动
async [Service.check]() {
returnthis.pool!==null
  }
}

_checkImpl(name) {
constimpl=reflect._getImpl(name)
// 从货架取：实例 + check 函数
if (!impl.check||impl.check(impl.value))
this._store[name] =impl
// ↑ check 通过（或没有 check）→ 存快照 → 依赖满足
else
deletethis._store[name]
// ↑ check 返回 false → 依赖不满足 → 继续等
}

// ===== 完整例子：你调 ctx.tools.execute('bash', 'ls -la') =====
// ① inner：真正执行命令的函数，最底端
// 它起进程、跑命令、返回结果——不检查任何东西
constinner=async (tool, input) => {
constproc=spawn(tool, input)
returnawaitproc.output
}
// ② 监听器：包裹在 inner 外面的壳
// 它多收到一个参数 next——「调下一层」的钥匙
ctx.on('tools/execute', async (tool, input, next) => {
// 检查权限
if (!isAllowed(input)) return'拒绝'
// next() = 把参数传给下一层
// 队列里还有别的监听器 → 调下一个
// 队列空了 → next() 落到 inner（真正干活）
returnawaitnext()
// 如果不调 next() → 后面的层和 inner 全部跳过
// → 命令不会被执行 → 这就是否决权
}, 'waterfall')
// ③ waterfall 内部，核心三行：
// 把监听器排成队列
constcbs= [权限检查]
// 取出参数表最后那个 = inner（真正干活的函数）
constinner=args.pop()
// next = 取下一个，队列空了就落到 inner
constnext= () => {
constcb=cbs.shift() ??inner
//    ↑ 有下一个取下一个    ↑ 没了就调 inner
returncb(...args, next)
}
// ④ 实际执行顺序（从 next() 开始转动）：
// next() → 取出权限检查 → 权限检查跑
//   → 它调 next() → 队列空了
//   → cbs.shift() 返回 undefined
//   → ?? inner → 调 inner → 命令真的跑了
//   → 结果从 inner → 权限检查 → 你手上

waterfall(...args) {
constcbs=this.dispatch('waterfall', args)
constinner=args.pop()
constnext= () => {
constcb=cbs.shift() ??inner
returncb(...args)
  }
args.push(next)
returnnext()
}

# 解析后的 web 命令行参数，做成一个普通服务提供出来
# 它自己 inject 了 cmdlineArgs——等的就是第 1 站 boot 里
# 那个「任何插件挂载之前」的 provideCmdline 注入
- id: web-startup
name: '@deepseek-ai/dsh-web-app/startup'
# HTTP 服务器：inject webStartup，等上面那行就绪才启动。
# host/port 从 webStartup 里取，取不到就回落部署默认值——
# 3080 这个数字的出处就在这两行表达式里
- id: webserver
name: '@deepseek-ai/dsh-host-webserver'
inject: [webStartup]
config:
host: !!jsctx.webStartup.host ?? '127.0.0.1'
port: !!jsctx.webStartup.port ?? 3080
# web 运行时：找到打包好的前端文件挂上静态服务、
# 注册网页版的提示词段——终端里那行 URL 就是它打的
- id: web-runtime
name: '@deepseek-ai/dsh-web-app'
inject: [webStartup]
config:
printUrl: true

// 组合里没有 hmr 服务时（web 禁用了共享模块重载），
// 挂一个只监听文件、不带模块根的 hmr 实例——
// 保证每个长期运行的界面都支持 cordis.patch.yml 热重载
if (ctx.get('hmr') ===undefined) {
awaitctx.loader.create({ name:'@deepseek-ai/cordis-plugin-hmr', config: { root: [] } })
}
// 两份用户补丁各挂一个监听：profile 自己的 + 全局的
awaitwatchUserPatches(ctx, {
binName:NAME,
filename:composed.profile.patchPath,
// 重组函数：bundle 在下、覆盖在上、全部深拷贝
compose:composeLive,
})
awaitwatchUserPatches(ctx, {
binName:NAME, filename:homePatchPath(), compose:composeLive,
})

hmr.registerConfig(filename, async () => {
// 重新读这份补丁文件（每次都重读，两份监听不串对方的旧内容）
constuserPatches=loadOptionalPatches(binName, filename) ?? []
// 重组：bundle 层在下、用户层在上，全部深拷贝
constpatches=compose(userPatches)
// 关键一步：更新根 Include 条目的 patches 配置。
// 整棵树都由这一个根条目挂载——它的配置一变，
// include 插件重放补丁到树上，差异自然产生
awaitentry.update({ config: { ...includeConfig, patches } })
})

// 退出信号。supervisor（进程看护器）指 systemd、Docker、pm2 这类
// 把 dsh 当服务托管的工具——它们停止服务发 SIGTERM，
// 按「正常退」处理，退出码 0；
// SIGINT 是你手动 Ctrl+C 的打断，退出码 130——启动窗口期内同样生效
process.on('SIGTERM', () => interrupt(0))
process.on('SIGINT', () => interrupt(130))
// fail-loud 守卫（installFailLoud）：快速失败——
// 未捕获异常先把整棵插件树 dispose 干净，再带着错误退出
installFailLoud(NAME, process, async () => {
awaitapp.current?.fiber.dispose()
})

constcomposeLive= ():PatchOptions[] => structuredClone([
// bundle 层在下
  ...composed.bundlePatches,
// 你的 profile 层
  ...loadOptionalPatches(NAME, composed.profile.patchPath) ?? [],
// 全局层
  ...loadOptionalPatches(NAME, homePatchPath()) ?? [],
// --patch 临时层在最上
  ...composed.overlays,
])

# dsh-base 清单的核心行（简化）
- id: agent-loop# Agent 主循环：turn/step 双层循环
- id: llm# 模型接口：流式请求、多适配器
- id: session# 会话日志：追加式记录
- id: tools# 工具注册表 + 三段瀑布把关
- id: subagent# 委派：八种子 agent 提供方
- id: system-prompt# 提示词段注册表
- id: sandbox# 沙箱与存储
- id: timer# 定时任务

第 1 站  dsh web → 参数归一 → boot 建上下文（注入环境快照和参数）
第 2 站  boot 五步依次拆：Context → plugin → 回调 provide → 合并挂树 → await
第 3 站  fiber 状态机：PENDING → _checkImpl → _refresh → ACTIVE
第 4 站  Service 上架：super() → provide 存值+通知 → ctx.weather 可用
第 5 站  waterfall：inner（干活函数）+ next（层间传话人）→ 否决权
第 6 站  web 三行叠出 3080 端口（inject 等待 + !!js 延迟求值）
第 7 站  热重载：保存补丁 → 深拷贝重组 → entry.update → 树算差异
第 8 站  干净退出：信号 → fail-loud → dispose 树 → 带着错误退出