workflow 服务
1. 定位
声明式的「触发器 + DAG」编排引擎：把多步骤任务（工具调用 / 发消息 / 等待 / 派发给 agent）按依赖图拓扑分层并行执行，支持 cron / interval / once / event / manual 多种触发方式。
- 服务注册名：
'workflow'（消费方经workflow.use调用）。 - 契约包：
@aalis/api-workflow。 - 该契约有运行时服务：
-api包只导出 interface + DSL 类型 + 事件契约，实现由@aalis/plugin-workflow提供。 - 设计取向（
packages/api-workflow/src/index.ts）：工作流「定义」是用户/AI 资产，存workspace；「运行实例」是运行时记录，存data。
2. 契约
服务接口
// packages/api-workflow/src/index.ts
export interface WorkflowService {
/** 列出全部工作流定义 */
listWorkflows(): WorkflowDef[];
/** 按 id 获取定义 */
getWorkflow(id: string): WorkflowDef | undefined;
/**
* 注册/覆盖一个工作流定义；persist 不为 false 时（缺省即写）同步写入 workspace/workflows/<id>.yaml。
* 经这里定义或覆盖的没有定义者（运行身份见 runWorkflow）：服务认不出调用方是谁，定义者只由实现
* 自己认证过调用者的入口记下
*/
defineWorkflow(def: WorkflowDef, opts?: { persist?: boolean }): Promise<void>;
/** 删除工作流（若 persist 文件存在则删除；有文件的定义删不掉文件时抛错） */
removeWorkflow(id: string): Promise<boolean>;
/** 手动触发一次运行；vars 与定义中的 vars 浅合并 */
runWorkflow(
id: string,
vars?: Record<string, unknown>,
triggerSource?: string,
/**
* 调用者身份（缺省=匿名）。内部工具与投递的消息按定义者与调用者里不是 owner 的一方裁决：调用者是 owner 时取定义者
* （没有即匿名），否则取调用者。定时与事件触发以定义者作调用者。sessionId 是调起这次运行的会话（经 workflow_run 调起时
* 有）：它所在的一轮在等这次运行，指向它的 agent 节点立即失败
*/
caller?: { platform?: string; userId?: string; sessionId?: string },
): Promise<WorkflowRun>;
/** 取消一个运行中的实例（若执行引擎支持） */
cancelRun(runId: string): boolean;
/** 获取运行实例（含历史） */
getRun(runId: string): WorkflowRun | undefined;
/** 列出最近 N 个运行实例（按时间倒序） */
listRuns(limit?: number, workflowId?: string): WorkflowRun[];
}服务经描述符 workflow = defineService<WorkflowService>('workflow') 登记（packages/api-workflow/src/index.ts），类型随描述符走，消费方写进 uses 即得类型；事件类型另经 declaration merging 增广 AalisEvents（见下文「事件契约」）。
工作流定义（DSL）
// packages/api-workflow/src/index.ts
export interface WorkflowDef {
id: string; // 唯一 id（文件名/外部引用键）
name?: string;
description?: string;
vars?: Record<string, unknown>; // 默认变量；运行时可覆盖（浅合并）
trigger: TriggerSpec; // manual 表示仅手动调用
nodes: NodeSpec[]; // DAG 节点
enabled?: boolean;
}触发器（packages/api-workflow/src/index.ts）：
export type TriggerSpec =
| { type: 'cron'; expr: string }
| { type: 'interval'; seconds: number }
| { type: 'once'; runAt: string }
| { type: 'event'; event: string; filter?: Record<string, unknown> }
| { type: 'manual' };节点（packages/api-workflow/src/index.ts）。基础字段 id / type / deps?（上游依赖，空=根节点）/ out?（把结果存入 outputs[out] 供下游 {{outputs.<out>}} 插值）。四种类型：
tool：{ tool: string; args?: Record<string, unknown> }——调用已注册工具；args内字符串值会被插值。send-message：{ sessionId: string; platform?: string; content: string }——fire-and-forget 发一条消息。消息带source，正文写成指令的也不执行，当普通消息交给她（见 security-model）。wait：{ seconds: number }——等待 N 秒。agent：{ instruction: string; sessionId?: string; platform?: string; timeoutSeconds?: number }——把指令派发给 agent 并等待本轮回复，回复文本作为节点结果（详见 §6）。instruction同样不当指令系统的指令执行。
运行实例
// packages/api-workflow/src/index.ts
export type RunStatus = 'pending' | 'running' | 'success' | 'failed' | 'cancelled';
export type NodeStatus = 'pending' | 'running' | 'success' | 'failed' | 'skipped';
export interface WorkflowRun {
runId: string;
workflowId: string;
status: RunStatus;
triggerSource: string; // cron:/interval:/event:/manual: ... 前缀
startedAt: number;
finishedAt?: number;
vars: Record<string, unknown>; // 实例变量（def.vars + trigger payload/运行时传入）
nodes: NodeRunInfo[]; // 节点运行明细
error?: string | TextRef; // 整体错误（首个失败的简要）
}运行的 error 与节点的原因（NodeRunInfo.error）都是 string | TextRef，可以是文字对象，按读者语言显示；交给模型与写日志的取 toSource。
事件契约
-api 通过 declaration merging 往 AalisEvents 注入以下事件（packages/api-workflow/src/index.ts）：
'trigger:fired'：入站——由 scheduler / 其它触发源 emit，workflow 订阅。payload 形如{ source, type, workflowId?, payload? }，其中payload会被透传进运行实例vars。'workflow:run:start'/'workflow:run:done'/'workflow:run:error'：以WorkflowRun为参数，出站——workflow 在运行各阶段 emit。'workflow:node:done'：{ runId, node: NodeRunInfo }，每个节点完成时 emit。
@aalis/api-workflow/package.json 的 keywords 含 aalis-api——是契约包，非可加载插件。
3. 谁提供 / 谁消费
参考实现（provider）
唯一一等实现 @aalis/plugin-workflow：
- 注册：
provide.use(workflow, service)（packages/plugin-workflow/src/index.ts）。 - 模块拆分：
engine.ts（DAG 拓扑调度 + 节点执行 +{{...}}插值）、triggers.ts（TriggerManager，cron/interval/once/event 接线）、loader.ts（YAML 定义加载/持久化）、persistence.ts（RunStore运行历史滚动写盘）、index.ts（服务装配 + AI 工具 + WebUI 页）。 - 依赖（
package.jsonaalis.service与index.ts双源）：required: ['config', 'cron-engine', 'events', 'hooks', 'lifecycle', 'logger', 'provide']（周期型触发器全部委托 cron-engine，见 §6）；optional: ['authority', 'storage', 'tools', 'webui-server']（authority用来判 owner，定运行身份，见 §6）。
触发源（trigger:fired 的 emit 者）
@aalis/plugin-scheduler 在执行调度任务时同时广播 trigger:fired，供 workflow 订阅（packages/plugin-scheduler/src/index.ts）：
await events.use.emit('trigger:fired' as any, {
source: `scheduler:${jobName}`,
type: rt.config.cron ? 'cron' : 'interval',
payload: { jobName, sessionId, platform, content },
});注意 scheduler 不直接依赖 workflow-api（事件类型由后者增广，scheduler 用 as any 解耦）。
典型消费点
- AI 工具（同插件内自我消费）：
enableTools开启时向 LLM 暴露workflow_define/workflow_list/workflow_run/workflow_get_runs/workflow_remove（packages/plugin-workflow/src/index.ts），全部走tools服务注册（optional 依赖；登记口在 tools 提供者缺席时先挂账，上线后自动补挂）。 - 工作流页（同插件内）：数据源与动作处理函数是闭包，直接使用本次激活构造的 service，不按名回查容器（查到的可能是别人提供的同名服务）。
- 跨插件外部消费者：当前仓内 workflow 服务的主要消费方就是 workflow 自身的工具/WebUI 层 + scheduler 的事件桥；第三方插件可经
workflow.use编程式触发/查询。
4. 写一个 provider
多数场景不需要自己写 provider——直接用
@aalis/plugin-workflow即可。仅当你要替换执行引擎（如换调度策略、加新节点类型）时才重实现。
必须 vs 可选
WorkflowService 的 8 个方法都应实现（接口无可选成员）。最小语义：
listWorkflows/getWorkflow：返回内存中的定义。defineWorkflow：校验 id（参考实现拿新定义的 id 作新定义文件的文件名，只收字母（各种文字，含组合符号）、数字、下划线与连字符，要写的文件存着另一个 id 的定义时拒收；已载入的定义改写与删除都对它载入时的文件，removeWorkflow也只对已载入的定义动文件）与图（无环、deps 引用合法、节点必填字段齐全），event触发器订阅负载带会话内容的事件的报错、不保存（参考实现的列表见 plugin-workflow）；通过的注册触发器，persist !== false时落盘；经它定义或覆盖的没有定义者，覆盖即清掉原来的定义者（见 §6）。runWorkflow：拓扑执行，返回完整WorkflowRun。务必按运行身份执行内部工具调用与投递的消息，定时与事件触发以定义者作调用者（见 §6）。cancelRun/getRun/listRuns：运行实例管理（cancelRun可降级为始终返回 false，但参考实现经 AbortSignal 真实支持）。
aalis.service 与 definePlugin provides/uses 双源必须同步
DI 靠包清单 + 代码导出双源声明（见 manifest-metadata）。两处都要写 provides: ['workflow']：
package.json（参考 packages/plugin-workflow/package.json 的 aalis.service）：
{
"keywords": ["aalis", "aalis-plugin"],
"aalis": {
"service": {
"required": ["config", "cron-engine", "events", "hooks", "lifecycle", "logger", "provide"],
"optional": ["authority", "storage", "tools", "webui-server"],
"provides": ["workflow"]
}
}
}src/index.ts 入口是 export default definePlugin({ name, subsystem: 'workflow', provides: [workflow], uses: { cronEngine, tools: optional(tools), storage: optional(storage), webui: optional(webuiServer), authority: optional(authority), events, hooks, lifecycle, logger, config, provide }, apply })（packages/plugin-workflow/src/index.ts）。不要再写具名 export const subsystem。
最小可编译骨架
import { cronEngine } from '@aalis/api-cron-engine';
import { storage } from '@aalis/api-storage';
import { tools } from '@aalis/api-tools';
import { webuiServer } from '@aalis/api-webui';
import type { WorkflowDef, WorkflowRun, WorkflowService } from '@aalis/api-workflow';
import { workflow } from '@aalis/api-workflow';
import { definePlugin, optional, provide } from '@aalis/core';
export default definePlugin({
name: '@yourscope/plugin-workflow-foo',
subsystem: 'workflow',
provides: [workflow],
uses: {
provide,
cronEngine,
tools: optional(tools),
storage: optional(storage),
webui: optional(webuiServer),
},
apply({ provide, cronEngine, tools, storage, webui }) {
void cronEngine;
void tools;
void storage;
void webui;
const defs = new Map<string, WorkflowDef>();
const runs: WorkflowRun[] = [];
const service: WorkflowService = {
listWorkflows: () => [...defs.values()],
getWorkflow: id => defs.get(id),
async defineWorkflow(def) {
// TODO: 校验无环 + deps 合法 + 拒绝订阅不可作触发源的事件 + 注册触发器（参考 engine.validateGraph / BLOCKED_EVENT_TRIGGERS / TriggerManager）
defs.set(def.id, def);
},
async removeWorkflow(id) {
return defs.delete(id);
},
async runWorkflow(id, vars, source, caller) {
const def = defs.get(id);
if (!def) throw new Error(`workflow "${id}" 不存在`);
const run: WorkflowRun = {
runId: `${id}-${Date.now()}`,
workflowId: id,
status: 'running',
triggerSource: source ?? 'manual',
startedAt: Date.now(),
vars: { ...(def.vars ?? {}), ...(vars ?? {}) }, // 浅合并：运行时覆盖默认
nodes: def.nodes.map(n => ({ id: n.id, type: n.type, status: 'pending' })),
};
runs.push(run);
// TODO: 拓扑执行 def.nodes；工具节点必须带 caller 身份过 authority（见 §6）
run.status = 'success';
run.finishedAt = Date.now();
return run;
},
cancelRun: () => false,
getRun: id => runs.find(r => r.runId === id),
listRuns: (limit, wfId) => {
const arr = (wfId ? runs.filter(r => r.workflowId === wfId) : runs)
.slice()
.sort((a, b) => b.startedAt - a.startedAt);
return limit && limit > 0 ? arr.slice(0, limit) : arr;
},
};
provide.use(workflow, service);
},
});priority / entryId / label
provide.use(workflow, instance, { priority?, label?, entryId? })：
priority：默认0。同名服务竞争时消费方选中哪一家按 本插件偏好 > 全局偏好 > priority > 注册顺序（无能力匹配，0.5.0 已移除——能力挂在实例上而非 DI 层）。普通第三方实现保持0，让用户在 WebUI 用 preference 选；要默认压过参考实现才取更高值（如50）。entryId：条目的逻辑身份，默认lifecycle.use.id；格式由契约约定，core 不检查。卸载时按激活清理，与 entryId 无关。label：WebUI 选择器展示名。
详见 service-model 与 core/service。
5. 标准消费方式
业务句柄是活的
workflow.use 是活的业务句柄，选择、缺席与固定同一家的规则见 选择层与业务层。可选依赖先看 present：
if (!workflow.present) return { ok: false, error: 'workflow 服务未就绪' };
const run = await workflow.use.runWorkflow(id, vars, 'manual:foo', { platform, userId });编程式触发
if (!workflow.present) return; // optional 依赖：先看 present 降级
const run = await workflow.use.runWorkflow('daily-report', { date: '2026-06-22' }, 'event:custom');
if (run.status !== 'success') logger.use.warn(`工作流运行未成功（${run.status}）: ${toSource(run.error) ?? ''}`);用事件触发（解耦，推荐跨插件做法）
不直接拿服务，而是 emit trigger:fired（像 scheduler 那样）——workflow 订阅后会按 workflowId 跑，并把 payload 注入 vars：
await events.use.emit('trigger:fired', {
source: 'myplugin:something',
type: 'event',
workflowId: 'my-workflow',
payload: { foo: 'bar' }, // 工作流里用 {{vars.foo}} 读取
});注意：trigger:fired 只有携带 workflowId 时才会被执行（packages/plugin-workflow/src/index.ts：if (!info?.workflowId) return;）。scheduler 当前广播的 trigger:fired 不带 workflowId，因此那条桥目前不会自动触发任何 workflow——它是为平滑迁移预留的（scheduler/src/index.ts 注释）。
错误边界
runWorkflow在「workflow 不存在 / 已禁用」时会抛（index.ts），消费者要兜 try/catch（工具 handler 与 WebUI action 都这么做）。- DAG 内部某节点失败不抛到
runWorkflow——整个 run 标记failed、run.error带首个失败原因，runWorkflow仍正常 resolve（engine.ts）。所以判结果要看run.status，不能只靠 try/catch。
6. 能力 / 风险 → 影响
authority：运行身份（核心安全约束）
工作流按定义者与调用者里不是 owner 的一方裁决，杜绝借他人 workflow 提权，也借不到 owner 的免确认。参考实现各入口的定义者与调用者、哪几种编排没有定义者，见 plugin-workflow。provider 作者重实现时必须守住下面几条：
- 定义者：只由实现自己认证过调用者的入口记下（参考实现是
workflow_define工具与工作流页），经服务的defineWorkflow定义或覆盖的没有定义者：服务认不出调用方是谁，契约也不收调用方报上来的定义者。定义者不写进定义文件，也不从定义文件读（工作区能被别的工具写，文件里的定义者伪造得出来）；记下定义者时连同写进文件的文本的哈希，载入时哈希对得上才认。 - 运行身份：调用者是 owner 时取定义者（没有即匿名），否则取调用者。
- 触发：定时与事件触发以定义者作调用者。事件参数与
trigger:fired的payload作为vars进入这次运行，插值进tool节点的参数、agent节点的指令或send-message的正文时就以定义者的身份下达，订阅的事件与trigger:fired的发出方要可信：event触发器不订阅负载带会话内容的事件（参考实现的列表见 plugin-workflow）。 - 传递：运行身份是
toolCallContext的唯一参与者（participants，匿名时是无主体；platform保持'workflow'），tool节点按它过 authority 闸，send-message与agent节点投递的消息以它为actor（匿名的不带）；匿名只能跑public（risk safe、minLevel 0）工具。
漏掉其中一条，匿名触发的工作流就能跑 owner 才允许的危险工具，或 owner 调起成员改写的编排时按 owner 执行，等于绕过 authority。risk{safe/sensitive/dangerous}→minLevel、确认（confirm 轴）等都在 tools.use.execute 那层裁决，workflow 只负责传对身份。
agent 节点：经网关送达，只认自己那一轮
agent 节点经 gateway（optional）的 ingressMessage 把指令（triggerType: 'proactive'，来源 workflow:<wf>:<nodeId>）送进目标会话，送达之前注册 agent:turn:after middleware（engine.ts）。约束：
- 指令在目标会话里单独成一轮，排在那里在跑的一轮之后。送达在节点自己的异步上下文（
AsyncLocalStorage）里调用，agent 让一轮跑在开它的那条输入的调用的上下文里，节点只认这个上下文里、目标会话的那一轮：同一会话里别的回合的回复不会被当成本节点的结果，同一编排两次运行的同名节点各认各的。省略sessionId时引擎自动生成一次性子会话workflow::<runId>::<nodeId>（engine.ts；契约AgentNodeSpec.sessionId的注释同此）。这类会话没有出生平台，选层按入口平台，场所在默认工作区（见 会话管理）。 - 默认
timeoutSeconds = 120；超时或outcome=error/aborted→ 节点失败（engine.ts）。 - 立即失败、不等满超时：网关不在；指向调起这次运行的会话（
runWorkflow的caller.sessionId，经workflow_run调起时有：那一轮在等这次运行，节点开不了轮，必然互等到超时）；ingressMessage返回（这条输入落定）时还没认到结局，即指令在网关相位被吞或排着时被放掉。目标会话处于禁言期，或在 flow-control 的生效范围内且限速窗口已满时，指令被 flow 相位吞掉。禁言期不看生效范围一律吞；除此之外，带source的编排消息不经触发策略、不受回复后冷却；限速只在目标会话在 flow-control 的生效范围内时生效（engine.ts）。
触发器全部委托 cron-engine
cron / interval 触发器都转成 cron-engine 的 subscribe（interval → @every Ns），与 scheduler 共享整分钟 tick（triggers.ts）。所以 cron-engine 是硬依赖（required）。once 用 setTimeout，触发即把 firedAt 记入 runsFile，一生只触发一次；runsFile 读不出（不存在以外的读取错误、解析失败、结构不对）时本次运行拒写该文件，也不安排任何 once（RunStore.onceLedgerReadable），避免拿空记账重放已触发过的一次性工作流；定义不存在时记账随之清除——removeWorkflow 删掉定义文件后当场清（有文件的定义删不掉文件时抛错、不清账；只在内存里的定义照常删除），手删 yaml 则由启动时的 prune（扫完定义、注册触发器之前）按现存定义集补清，定义者账本同样清；event 用 events.use.on 订阅（triggers.ts）。
storage 不是沙盒
定义存 defsDir（默认 workspace:/workflows）、运行历史存 runsFile（默认 data:/workflow-runs.json），都只接受 storage URI，写成相对路径等其它形态时拒绝激活（index.ts、isStorageUri）。storage 按 root 做权限位但不是沙盒，见 storage-uri-grammar。send-message / tool 节点能触达任意会话与已注册工具，等价于 owner 资产的执行面——参考实现的 workflow_define 要 owner 审批（参与者都是 owner 的回合可由免确认白名单免），批过的编排按批的人运行：低权限用户能让她写编排，但原文要经 owner 过目。
7. 边界与注意事项
- event 触发的 payload 形态：
event触发时TriggerManager把监听到的事件参数包成{ args }传给 fire（triggers.ts：this.fire(def.id, ..., { args })），fire 再把它整体作为extraVars注入运行实例vars（index.ts）。因此事件触发的工作流里要用{{vars.args}}（或{{vars.args[0].xxx}}走路径访问）读取事件负载，而不是{{vars.xxx}}直接读事件字段。对比之下trigger:fired（含 webhook/scheduler 桥）的info.payload是直接展开进 vars 的（index.ts），其字段名直接用{{vars.字段}}。两条入口的 vars 形态不同，写定义时需区分。 - event filter 只做顶层等值匹配：
filter的每个 key 必须在事件第一个参数（顶层）等值命中；payload 不是对象时只有空 filter 通过（triggers.ts）。无嵌套/范围匹配。 - DAG 失败即停：任一节点失败，引擎不再调度新批次，未跑的节点标
skipped，整 run =failed（engine.ts）。没有节点级重试 / 部分继续。 - 取消打断阻塞中的节点：
cancelRun与插件拆卸（停机、卸载、bounce）都中止这次运行的信号：不再调度新节点；阻塞中的wait/agent节点立即失败（「运行已取消」），已派发的 agent 回合不随之中止，照常跑完；tool节点经ToolCallContext.signal收到中止，是否提前返回由工具决定（engine.ts）。拆卸等在飞 run 收尾、终态写进runsFile才结束，不响应中止的工具至多让它等到disposeTimeoutMs（index.ts）。 - 运行历史是滚动 + 整体重写：
RunStore超过maxRuns（默认 200、最小 10）裁剪最旧的已结束的运行，在跑的不裁（收尾时照常更新它），每次变更整体重写文件、写入串行化（persistence.ts）。载入时还在running的是进程没有正常退出时留下的，收口成cancelled（与拆卸时被打断的同一状态），error写进程退出时中断，不写finishedAt；节点与引擎取消时同样收：running的记failed（error同一句），pending的记skipped。历史不是无限审计日志。同文件还存 once 的firedAt与定义者账本（{ runs, onceFired, definers }）。 - 节点
output截断：存入outputs的是完整结果，但NodeRunInfo.output展示字段截断到 1000 字符（engine.ts）。下游插值用的是完整值，UI 看到的是预览。 - 同名服务无能力选择：0.5.0 起 DI 不做能力匹配；多个 workflow provider 并存时靠 preference/priority 选，能力信息在实例上自管（见 service-model）。
8. 交叉链接
- 概念：service-model（DI 按名解析 / 同名竞争）、lazy-service-access（业务句柄是活的）、manifest-metadata（aalis.service 与 definePlugin provides/uses 双源）、storage-uri-grammar（定义/历史存储）、security-model、message-llm-pipeline（agent 节点经网关的
ingressMessage接入主链路）。 - 核心：core/service、core/context、core/plugin、core/events、plugins/plugin-authority、plugins/plugin-tools。
- 相关服务：services/tools（
tool节点的执行面 +workflow_*工具）、services/agent（agent节点的回复 join）、services/storage（定义/历史落盘）。相关插件：@aalis/plugin-scheduler（trigger:fired触发源）、@aalis/plugin-cron-engine（cron/interval 调度底座）。