跳到主要内容

workflows 代理与 ActionableWorkflow

这章讲底层:props.workflows 这个对象怎么靠 JS Proxy 凭空生成、startNewInstance 怎么落到后端 createWorkflow、单实例的 ActionableWorkflow 操作面怎么实现,以及类型系统如何从 workflow 定义推断出 input/output/tags 的精确类型。

1. 两层对象,各管一摊​

先分清两个东西,别混:

对象你怎么拿到管什么类型
workflows(集合代理)处理函数 props.workflows / proxyWorkflows()按名字启动/列出实例WorkflowProxy
workflow(单实例操作面)处理函数 props.workflow操作当前这一个实例ActionableWorkflow
  • workflows.lintAll.startNewInstance(...) —— 启动一个新实例。
  • workflow.setCompleted() —— 收尾当前正在处理的实例。

2. workflows 代理:用 Proxy 凭空造方法​

proxyWorkflows 返回一个 JS Proxy,任何属性访问(workflows.随便什么名字)都即时返回一对方法 { listInstances, startNewInstance }:

// packages/sdk/src/bot/workflow-proxy/proxy.ts:34-62(真实源码,节选)
export const proxyWorkflows = (props) =>
  new Proxy({} as WorkflowProxy, {
    get: (_, workflowName) => ({
      listInstances: (input) => createAsyncCollection(/* 调 listWorkflows 分页 */),
      startNewInstance: async (input) => {
        const { workflow } = await props.client.createWorkflow({
          name: workflowName,
          status: 'pending',
          ...prefixTagsIfNeeded(input, { alias: undefined /* props.pluginAlias */ }),
        })
        return { workflow: wrapWorkflowInstance({ ...props, workflow }) }
      },
    }),
  })

要点:

  • 属性名即 workflow 名:get 拿到的 workflowName 直接当 createWorkflow 的 name。所以 workflows.foo 不需要预先注册键——Proxy 让它「看起来什么名字都有」。
  • 启动 = 建一个 pending 实例:startNewInstance 固定 status: 'pending',然后把返回的实例用 wrapWorkflowInstance 包成操作面。
  • 那个 undefined /* props.pluginAlias */ 是 §index 提到的 KKN-292 技术债:本应按插件 alias 前缀 tag,但后端校验不允许 #,所以暂时传 undefined(proxy.ts:10-32 的长 FIXME)。

3. startNewInstance 的真实用法:链式 workflow​

file-synchronizer 用它把两个 workflow 串起来——buildQueue 干完立刻启动 processQueue:

// plugins/file-synchronizer/src/hooks/workflow-continued/build-queue.ts(真实源码,节选)
await props.workflow.setCompleted()                 // 当前 workflow 收尾
await props.workflows.processQueue.startNewInstance({ // 启动下一个 workflow
  input: { jobFileId: workflowState.jobFileId },
  tags: {
    syncInitiatedAt: props.workflow.tags.syncInitiatedAt!,
    syncJobId: props.workflow.tags.syncJobId!,
    syncType: props.workflow.tags.syncType!,
  },
})

这段在干嘛:这就是 Botpress 里「子流程」的真实形态——不是父子树,而是「A 收尾 → 用 A 的 tags 启动 B」。input 类型化(必须给 jobFileId),tags 也类型化。

4. ActionableWorkflow:单实例操作面​

wrapWorkflowInstance 把一个后端 client.Workflow 包成带动作的对象。它先摊开后端实例的所有字段(id/status/tags/output...),再挂上 5 个方法:

// packages/sdk/src/bot/workflow-proxy/proxy.ts:65-91(真实源码,节选)
export const wrapWorkflowInstance = (props) => {
  let isAcknowledged = false
  return {
    ...(unprefixTagsOwnedByPlugin(props.workflow, { alias: undefined }) as ActionableWorkflow),
    async update(x) {
      const { workflow } = await props.client.updateWorkflow({ id: props.workflow.id, ...prefixTagsIfNeeded(x, ...) })
      await props.onWorkflowUpdate?.(workflow)
      return { workflow: wrapWorkflowInstance({ ...props, workflow }) }
    },
    // acknowledgeStartOfProcessing / setFailed / setCompleted / cancel ...
  }
}

三个统一模式,记住就懂全部 5 个动作:

  1. 每个动作都 = 一次 updateWorkflow(改 status 或字段)。
  2. 改完调 onWorkflowUpdate?.(workflow)——这是 §02 里 dispatcher 用来「捕获最新状态并传给下一个处理函数」的回调钩子。
  3. 返回一个重新包装的新实例(wrapWorkflowInstance({ ...props, workflow })),所以可以链式继续操作。

onWorkflowUpdate 是这套设计的暗线:操作面本身不持有可变状态,而是每次变更都把新状态「喊」给注册方。dispatcher 注册它来更新接力链的 currentWorkflowState(implementation.ts:251-262)。

5. 类型推断:从定义到精确 input/output/tags​

类型层让你写 workflow.setCompleted({ output }) 时,output 的形状精确匹配该 workflow 定义里的 schema。核心是 ActionableWorkflow 这个映射类型(packages/sdk/src/bot/workflow-proxy/types.ts:42-105):

// packages/sdk/src/bot/workflow-proxy/types.ts:46-67(真实源码,节选)
client.Workflow & {
  name: TWorkflowName
  input: Cast<TBot['workflows'][TWorkflowName], ...>['input']
  output: Partial<Cast<TBot['workflows'][TWorkflowName], ...>['output']>
  tags: Partial<TBot['workflows'][TWorkflowName]['tags']>
  update(x: AtLeastOneProperty<...>): Promise<{ workflow: ActionableWorkflow<...> }>
  // ...
}

几个值得记的类型决策:

字段/方法类型处理为什么
outputPartial<...>运行中可能只填了一部分 output
tagsPartial<...>tags 不一定都设了(所以示例里用 ! 断言)
update(x)AtLeastOneProperty<...>至少改一项,不许传空对象
setFailed(x)Required<Pick<..., 'failureReason'>>失败必须给原因
getWorkflow / updateWorkflow放弃推断只有 id 没有名字,推不出类型(client/types.ts:211-218 FIXME)

一句话:有名字的入口(workflows.X、处理函数注入的 workflow)是类型安全的;只有 id 的入口(client.getWorkflow)不是。 所以优先用前者。

6. 谁来注入这两个对象​

  • 处理函数里的 workflow / workflows:由 workflowHandlers getter 在调用你的处理函数前注入(implementation.ts:252-261)——workflow 来自 wrapWorkflowInstance,workflows 来自 proxyWorkflows(props)。
  • 插件里的 workflows:插件实现自己在 _getTools 里建一份 proxyWorkflows({ client, pluginAlias })(packages/sdk/src/plugin/implementation.ts:106)。

7. 关键细节 / 坑​

  • Proxy 让「任意 workflow 名」都有方法,类型才是唯一的「这个名字存不存在」约束;名字打错运行时不会立刻报错,而是后端建出一个你没处理函数的实例。
  • tags 前缀逻辑当前是 no-op(alias: undefined),这是 KKN-292 技术债,别依赖插件 tag 前缀。
  • 每个动作返回的是新包装实例,要拿最新状态用返回值,别复用旧引用。
  • onWorkflowUpdate 是状态接力的关键钩子,理解它才理解 §02 的处理函数链怎么拿到最新状态。

代码地图​

主题文件路径符号名
集合代理(启动/列出)packages/sdk/src/bot/workflow-proxy/proxy.tsproxyWorkflows
单实例操作面packages/sdk/src/bot/workflow-proxy/proxy.tswrapWorkflowInstance
代理 / 操作面类型packages/sdk/src/bot/workflow-proxy/types.tsWorkflowProxy / ActionableWorkflow
client 端类型推断入口packages/sdk/src/bot/client/types.tsCreateWorkflow / GetOrCreateWorkflow / ListWorkflows
插件侧代理构造packages/sdk/src/plugin/implementation.ts_getTools(proxyWorkflows)
链式 workflow 示例plugins/file-synchronizer/src/hooks/workflow-continued/build-queue.tshandleEvent
KKN-292 技术债说明packages/sdk/src/bot/workflow-proxy/proxy.ts顶部 FIXME 注释块