@bpos/activity-sdk (13.10.0)
Installation
@bpos:registry=https://gitea.akhr-ai.com/api/packages/ChaiHao/npm/npm install @bpos/activity-sdk@13.10.0"@bpos/activity-sdk": "13.10.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；不能把其他节点或不同定义当作来源。不应按重做次数清空表格或改派手填单。
发出一件事（13.10.0）
清单 declares.events 里声明过的事，在节点里用 cap.emit(event, data?, opts?) 发出：
await cap.emit('need-review', { item_no: 'A-1', level: 'high' });
- 不知道有没有人接：实践里有事件子流程接它，系统就另起一趟去办；没人接只记一笔。答复里只有
eventId，发完照常往下做。 data按那一件声明的几格给（没声明的键、类型不对、必填没带，系统答invalid并指名）；文件引用要是这一步够得着的。- 缺省带幂等键：「事件 id · 事件名 · 第几次」，与
cap.order()同一套、各数各的——同一条事件重投回来，第 n 次发的还是第 n 把键，系统认得出是同一件，不再起第二趟。按业务键各发一件、重跑也只认业务键的，自己给{ idempotencyKey }。 - 被中断型事件子流程收掉的那一步，下一次调能力面收
node-closed——照「这一步已收场」收尾即可。 - 假能力面答
{ eventId: 'emit-n' }；要验带了什么，看face.calls。
Dependencies
Dependencies
| ID | Version |
|---|---|
| @bpos/activity-contract | ^13.10.0 |
| zod | 4.4.3 |
Development Dependencies
| ID | Version |
|---|---|
| @types/node | ^24.0.0 |
| tsx | ^4.19.0 |
| typescript | ^5.6.0 |