@ crossdelta/ 流程核心
用于CLI,MCP,测试和代理的可执行,类型安全的流动.
定义 会发生什么 -没有 如何进行。 。 。 .
flowcore 是一个用于构建的小型功能工具包 宣告性、可执行流动
可以在不同的主机运行 :
- 互动的CLI
- 非交互式(测试、自动化)
- MCP / AI 工具
- 未来的 UI 或代理
没错 没有 一个快速的库.
这是一个 流动的核心执行原始。 。 。 .
为什么流芯存在
大部分CLI工具是紧配对的 互动 与 执行数字 :
- 用户提示
- 控制流量
- 副作用
- /TTY问题
这种接合使得:
- 没有终端的测试流程
- cLI 以外的再利用逻辑
- 向 MCP 或 AI 工具曝光相同的行为
- 计划、模拟或在执行前审查副作用
flowcore 分离 发生了什么事 从 如何驱动。 。 。 .
一个流. 许多主持人。 零重写.
核心构想
Flow = declarative steps + typed context
Runner = decides how the flow is executed
你说 发生了什么事。 。 。 .
跑者决定 互动如何运作。 。 。 .
安装
bun add @crossdelta/flowcore
# or
npm install @crossdelta/flowcore
快速实例
import { createFlow, runFlow, type FlowContext } from '@crossdelta/flowcore'
interface UserContext extends FlowContext {
name: string
role: 'admin' | 'user'
}
const flow = createFlow<UserContext>()
const steps = [
flow.input('name').build(),
flow.select('role')
.choices(['admin', 'user'])
.build(),
flow.task('Save user')
.steps([
(ctx) => {
console.log(`Hello ${ctx.name} (${ctx.role})`)
},
])
.produces('name', 'role')
.build(),
]
await runFlow(steps, {
initialContext: {},
})
是什么让流芯不同
主机不可知流量
相同的流量可以运行 :
- 交互的 CLI
- 在试验中不互动
- mCP 工具内
- 稍后的代理或用户界面
无分行相.
没有重复.
完全类型安全的背景
每一步骤都丰富了共享的、按键的上下文:
interface MyContext extends FlowContext {
eventType: string
service?: string
}
TypeScript 保证跨步骤的正确性.
非互动执行
await runFlow(steps, {
initialContext: {
name: 'Max',
role: 'admin',
},
})
完美为:
- 测试
- 自动化
- AI 执行
- 干线运行
步骤原始
投入
flow.input('eventType')
.prompt({
title: 'Event Type',
message: 'Event type to publish (e.g., order.created)',
})
.skipWhen((ctx) => ctx.eventType !== undefined)
.build()
选择
flow.select('role')
.choices(['admin', 'user'])
.optional('Skip')
.build()
任务
flow.task('Publish event')
.steps([
loadDataFromInput,
publishToNats,
])
.produces('data', 'published', 'usedMock')
.build()
测试流量
流量不依赖于TTY或渲染.
说明 .produces(...)
.produces(...)文档 保证存在或修改 任务完成后.没错 没有 更改运行时间行为,执行代码,或执行验证. 这是读者、工具手和未来的主机(CLI、MCP、代理商)的公开承诺.
test('publish event flow', async () => {
const result = await runFlow(steps, {
initialContext: {
eventType: 'order.created',
},
})
expect(result.context.published).toBe(true)
})
没有即时嘲笑.
没有制造器的根.
流程样式指南
这些规则不是由编译者强制执行,而是强烈推荐.
1. 从上到下流动
阅读器应该理解文件之间的流动而不跳跃.
不错:
input → abort → input → task
错误 :
task → input → hidden condition → side effect
2. 优先采取明确步骤
总是使用 steps([...]) 用于多个操作的任务.
不错:
flow.task('Publish event')
.steps([loadData, publish])
避免 :
.run(loadData).run(publish)
3. 流动决定何时使用跳过
skipWhen 表示控制流,而不是验证.
不错:
.skipWhen(hasExplicitData)
4. 提前明确中止
使用 abortIf 成为它自己的一步.
不错:
flow.abortIf(isInvalid, getErrorMessage)
5. 宣布上下文突变
总是记录任务变异.
不错:
.produces('data', 'published')
6. 使域逻辑远离流动
流出乐团.
域逻辑生活在由步骤所召唤的函数中.
7. 偏好小型、命名功能
匿名的羊肉只是些无关紧要的逻辑.
8. 任务之外没有副作用
投入和条件必须纯洁.
9. 一个流出,一个故事
如果流动感觉太长,就把它分了.
10. 流动是价值
将流量视为不同主机可以执行的数据.
流心适合时
flowcore 如果您:
- 使用复杂的向导构建 CLI
- 想要向 MCP 或 AI 工具曝光相同的逻辑
- 关注可检验性和确定性
- 想要将互动与执行分开
没错 没有 用于简单的一次性即时脚本.
变化跟踪公用事业
flowcore 为跟踪业务成果提供公用事业(创建/更新/减少).
这些是 一般和主机不可知 ——不绑入文件或任何特定域.
何时使用
使用 ChangeResult 当您需要时:
- 跟踪创建、更新或跳过的资源
- 记录文物槽的结果(主机、文件系统、数据库)
- 基于操作结果的不同反应
- 跟踪流量变化
非流利方法(建议默认)
import { change, recordChange, isCreated, type ArtifactsSink } from '@crossdelta/flowcore'
// Perform some operation
const result = await generateContract()
// Map to ChangeResult
const outcome = result.created
? change.created(result.path)
: change.skipped('Already exists')
// Record to artifacts sink
recordChange(ctx.artifacts, 'Contract', outcome)
// React based on outcome
if (isCreated(outcome)) {
ctx.effects.emit(contractCreated(outcome.ref, ctx.eventType))
}
为什么这是默认 :
- 明确和明确
- 在复杂的流动中易于跟踪
- 没有隐藏的行为
- 熟悉的必备样式
流利方法(可选糖)
import { change, trackChange } from '@crossdelta/flowcore'
trackChange(change.created('/path/to/contract.ts'))
.record(ctx.artifacts, 'Contract')
.onCreated((ref) => {
ctx.effects.emit(contractCreated(ref, ctx.eventType))
})
节制使用 :
- 简单的线性案例不错
- 在复杂的条件逻辑中避免
- 记住:
.record()财务报告和审定财务报表.onX()钩子是独立的(没有隐含的副作用)
API 概览
类型 :
type ChangeResult<Ref = string> =
| { status: 'created'; ref: Ref }
| { status: 'updated'; ref: Ref }
| { status: 'skipped'; reason?: string }
interface ArtifactsSink {
created(label: string, ref?: string): void
updated(label: string, ref?: string): void
skipped(label: string, reason?: string): void
}
构造器 :
change.created(ref) // => ChangeResult with status: 'created'
change.updated(ref) // => ChangeResult with status: 'updated'
change.skipped(reason?) // => ChangeResult with status: 'skipped'
类型警卫 :
isCreated(result) // narrows to { status: 'created'; ref: Ref }
isUpdated(result) // narrows to { status: 'updated'; ref: Ref }
isSkipped(result) // narrows to { status: 'skipped'; reason?: string }
非流利帮助者 :
recordChange(artifacts, label, result) // records to sink, returns result
流利的帮助者 :
trackChange(result)
.status() // get status
.ref() // get ref (or undefined if skipped)
.record(artifacts, label) // record to sink (returns this)
.onCreated(fn) // execute fn if created (returns this)
.onUpdated(fn) // execute fn if updated (returns this)
.onSkipped(fn) // execute fn if skipped (returns this)
重要:关注事项的分离
.record() 财务报告和审定财务报表 .onX() 钩子是 独立数字 :
// ✅ This ONLY records to artifacts (hook NOT called)
trackChange(result).record(artifacts, 'File')
// ✅ This ONLY calls the hook (artifacts NOT recorded)
trackChange(result).onCreated((ref) => console.log(ref))
// ✅ Both together work independently
trackChange(result)
.record(artifacts, 'File')
.onCreated((ref) => console.log(ref))
为什么这很重要:
- 没有隐藏的行为
- 对副作用的明确控制
- 在复杂的流动中容易解释
哲学
- 功能而非面向对象
- 声明,不是必须
- 默认可测试
- 最小面积
- 没有魔法
流动是价值。 处决是一个细节.
许可证
麻省理工学院 横跨德尔塔