@bpos/activity-sdk (13.8.0)
Installation
@bpos:registry=https://gitea.akhr-ai.com/api/packages/ChaiHao/npm/npm install @bpos/activity-sdk@13.8.0"@bpos/activity-sdk": "13.8.0"About this package
@bpos/activity-sdk —— 薄封装
「活动侧感觉像进程内,边界却在线材上。」 可 curl、可 mock、可单测活动而不起系统;将来活动不是 Node 写的也不破。
它托管三件事:收事件(在系统给的端口上听着,把投来的每一条过一遍契约再交给你)、 调能力面(每个名词一个方法)、交出。除此之外什么都不做——怎么分拍、什么时候派工单、 什么时候问模型、什么时候交出,是活动自己的事,系统不规定、不解读。
export * from '@bpos/activity-contract':活动只 import 这一个包就够。
规范在《操作系统系统设计》§6.2(进程与事件)· §6.4(名词表)· §6.5(事件表)· §7.3(交出)。
公开 API 一览
| 名字 | 是什么 |
|---|---|
defineActivity(handle, opts?) |
常用形状:替你做掉去重与实例状态存取 |
runActivity(opts) |
低阶:只收事件与 cap,去重和状态归你 |
readEnvelope(env?) |
从环境变量读信封;七样必有,缺一样当场抛;业务时区缺席答 UTC |
Envelope |
taskId · nodeRunId · nodeId · token · base · eventPort · sandboxDir · timezone |
dayIn(zone, at?) |
那个时区里的哪一天(YYYY-MM-DD):算「今天」「晚了几天」用它,别 slice(0, 10) 一个 ISO 串 |
Capabilities |
名词表的 client;cap.call(名词, 载荷) 是底层,别的是它的糖(order · readOrder · writeOrder · voidOrder · say · input · model · read · state / putState · memory / putMemory · getFile · fileText · putFile · fileInfo · docs · timer · node · note) |
cap.read({ address, … }) |
按地址读对象、单据与台账(只读:写业务库只有人提交工单那一条路);不给 retired 就只答在用行(单据上的废弃标记 = 冲销),要连冲掉的一起看给 retired: 'all' |
cap.handoff({ output, summary }) |
交出,答 accepted / refused |
CapabilityError |
能力面拒绝了:capability · code(封闭集)· message · hint |
CapabilityErrorCode |
那组码的类型(契约包里同名的是类型,这里同名的是类,所以换个名字再导一次) |
readPayload(event, kind) |
按种类过一遍契约的真 zod;种类对不上、不合契约都当场抛 |
ActivityContext<T> |
处理函数每一拍收到的那一份:event · cap · env · state · save · handoff |
ActivityHandle |
ready(答真正绑上的端口)· close |
fakeCapabilityFace(opts?) |
一台最小的假能力面,单测用(见下) |
没有的:图构建器、步骤函数表、门钩子——活动包里没有图。一个活动 = 一个节点,
路径归实践、在系统侧编排;分流按你产出里的枚举字段接线。清单的声明帮手也没有:
bpos-activity.json 是手写的。但校它的尺子有——manifestSchema 随契约包一并再导出,
构建时 manifestSchema.parse(JSON.parse(...)) 一句,写坏了停在你改得动的地方
(两个范例包的 build.ts 就是这么干的)。
defineActivity 一拍的顺序
import { defineActivity, readPayload } from '@bpos/activity-sdk';
defineActivity<{ orderId?: string }>(async ({ event, cap, state, save, handoff }) => {
if (event.kind === 'enter') {
const { input } = readPayload(event, 'enter');
// 要落的值预填进件;落到哪写在清单里这张单的 commit 上——活动不写库
const { orderId } = await cap.order('diff-review', {
data: { part_no: input['part_no'] },
brief: `核一下 ${String(input['part_no'])} 这几项差异`,
});
await save({ orderId }); // 停在这一拍,等人办完
return;
}
if (event.kind === 'work-order-reply') {
const p = readPayload(event, 'work-order-reply');
// 人提交即落定:这一次落了什么在落账摘要里(p.landing),不回头去库里捞
const rows = p.landing.items.reduce((n, i) => n + i.rows, 0);
await handoff({ output: { conclusion: String(p.decision ?? '') }, summary: `差异核完,落了 ${rows} 行` });
}
});
state.get读实例状态(SDK 存的是{ seen: [事件 id…], own: 你的状态 })。- 这条
id在seen里 → 直接返回,你的处理函数不跑。 - 跑你的处理函数,
ctx.state是第 1 步读到的own(第一次是undefined)。 - 每次
save(next)写一次state.put——只碰own,不记「见过这一条」。 - 处理函数正常返回,收尾再写一次:这一次才把
event.id记进seen。 - 交出之后收尾那一次必被拒(
node-closed,实例状态已冻结),SDK 吞掉这一个码—— 每一次成功的交出都在 stderr 上留一行错误的话,那正是「看起来失败,其实成功了」。 别的拒绝照旧抛。
ctx.state 在同一拍里 save() 之后不跟着变(它是这一拍开头那一份);口径是
「上一拍存下的状态」。
opts:env(不读环境变量,直接给信封——测试用)· fetchImpl(换掉 fetch——假能力面)·
keepSeen(去重表长度,缺省 200)· ackAfterMs(下面那条)· onError。
确认的时机 —— 一条一条处理,处理完再答
系统那一侧一次只有一趟投递在飞,可**「投到了」不等于「处理完了」。收下就答 200 的话, 同一个进程会有两拍同时在跑,读同一份实例状态、各写一次,后写的赢——去重表与你的状态 都会被抹掉一截。所以 SDK 里有一条进程内的 FIFO 队列**:事件一条一条处理。
| 局面 | 答什么 | 为什么 |
|---|---|---|
| 队列空 · 处理函数正常返回 | 200 { ok: true } |
确认 |
| 队列空 · 处理函数抛了 | 500 { ok: false },不确认 |
抛了 = 这一拍没发生。系统按「没送到」退避重投,三次不成保护性挂起并留痕——那是设计里那条看得见的失败路 |
处理超过 ackAfterMs(缺省 20 秒) |
先答 200,处理接着在后台跑 |
长活不能让系统超时重投(服务端的确认时限缺省 30 秒) |
| 队列非空 | 立刻 200,这一条排队 |
确认时限是从这条事件投出去那一刻算的,排队等待也在里面;它前面有事在跑,等不到自己那一拍 |
nodeRunId 不是自己的 |
409,不确认 |
一个本机端口会被先后分给两个任务节点。替别人签收的那一条,真正该收它的那一步从此永远停着 |
| 信封或载荷不合契约 | 400,不确认 |
这一条谁也没处理 |
答过 200 之后才抛的(排队进来的那些条、软时限之后才抛的):系统那一侧已经记成「送到了」,
所以 SDK 替你报一声 event.fail(这一条的 id 与抛出来的那句话),系统把它放回队里、按「没送到」
那一套退避重投、次数接着数,三次不成保护性挂起,理由里写着报上来的那句话。挂起中、已收场的不报
(报了也只会被拒);报不出去的只能经 onError 说出来(缺省写 stderr,进工程日志)。
所以照旧:能存的先存;拿不准的地方 catch 住、cap.say 说给人听。
ackAfterMs 只影响「多久之后先答 200」,不影响处理本身——处理照旧跑到完。
它必须小于服务端的确认时限(缺省 30 秒,BPOS_LIMIT_EVENT_ACK_MS)。
去重
事件至少送到一次:同一条可能来两遍(进程崩在确认之前、网络抖动)。系统不重放,
去重是活动的事——defineActivity 把处理过的 id 记在实例状态里(keepSeen,缺省最近 200 条)。
自己写 runActivity 的话,这件事得自己做。
只有处理成功的那一条才进去重表:抛了就走不到收尾那一句,重投回来会重新跑一遍。 200 条这个窗口够用,因为系统只重投没确认过的那些条,而它一次只投一条。
脱离系统单测
边界在线材上,所以换掉 fetch 就能把一个活动跑一遍。假能力面由 SDK 提供:
import { test } from 'node:test';
import { strict as assert } from 'node:assert';
import { EVENT_PATH, fakeCapabilityFace } from '@bpos/activity-sdk';
import { runMyActivity } from '../src/activity'; // 工厂:import 进来没有副作用
test('进节点:派一张单、交出', async () => {
const face = fakeCapabilityFace();
const app = runMyActivity({ env: face.env, fetchImpl: face.fetchImpl });
const port = await app.ready; // 传 0 让系统派口,答真正绑上的那个
const res = await fetch(`http://127.0.0.1:${port}${EVENT_PATH}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ id: 'e1', nodeRunId: face.env.nodeRunId, kind: 'enter',
payload: { input: { subject_id: 'W-001' }, label: '回声' }, at: new Date().toISOString() }),
});
assert.equal(res.status, 200); // 队列空的那一条处理完才答,不用 sleep
assert.deepEqual(face.used, ['state.get', 'order.create', 'state.put', 'state.put']);
assert.deepEqual(face.handoff?.output, { said: '回声:W-001' });
await app.close();
});
fakeCapabilityFace() 守两条:递出去的每一次过 capabilityPayload[名词](你的载荷写错,
红在你调的那一行)、答回去的每一次过 capabilityResult[名词](假面自己答错也红在这里,
而不是三步之后活动那一侧报「系统答的结果不合契约」——那句话指的是系统)。
它还替你守住一处只写在契约里的规矩:state.get / memory.get 一次没存过答 {} 不答 null。
读得到的:face.calls(每一次调用与解过的载荷)· face.used(只要名词)· face.handoff ·
face.handedOut · face.state · face.memory · face.env。
调得动的:answers(某几个名词自己答什么)· refuse(让某个名词拒绝,验你 catch 分流的那一支)·
outcome(交出答什么)· state / memory 的初值 · node(node.read 答的那几格)。
把活动写成一个工厂(export const runMyActivity = (opts = {}) => defineActivity(handler, opts)),
起进程那一句单独放在构建入口里——模块顶层直接 defineActivity(...) 的话,测试一 import
就去读环境变量、连真能力面。范例包 activities/echo 与 L1 的走查夹具
(tests/e2e/fixtures/walkthrough/)都是这个形状。
假面不模拟系统那一侧的顺序性(事件按序、至少送到一次、空闲回收):那要起真系统跑端到端。 这里验的是「这个活动在这一拍里做了什么」。
13.4.0 同步再导出活动契约的表格只读列 hidden;运行时能力调用接口不变。
重做核对单
收到验收单 result: 'redo' 后,持久化原单 orderId,正常重新准备表单,再调用 cap.order(definition, { data, redoOf: orderId })。系统核验来源并按原单保存的理由修订,原件和人工改动可读,新单仍需人核对。恢复时沿用同一 redoOf;不能把其他节点或不同定义当作来源。不应按重做次数清空表格或改派手填单。
Dependencies
Dependencies
| ID | Version |
|---|---|
| @bpos/activity-contract | ^13.8.0 |
| zod | 4.4.3 |
Development Dependencies
| ID | Version |
|---|---|
| @types/node | ^24.0.0 |
| tsx | ^4.19.0 |
| typescript | ^5.6.0 |