Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/subsystems/workflow.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher,
## Entry points

- Endpoint `workflow/state` and `workflow/templates` are reads.
- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/say`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes.
- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes.
- Tool `run_workflow` starts a named run with a brief. Tool `define_workflow` writes the same graph the canvas edits.
- `flybot.tasks.onWrite` calls `ingest`. A new Bot match returns `{ ownerBotId }`. A Workflow match starts the run inside the plugin and leaves the card's owner to the host.

Expand Down
2 changes: 1 addition & 1 deletion plugins/workflow/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ npx -y @deepseek-ai/dsh@0.2.0-rc.2 plugin --profile web add "github:ThinkFlowLab
npm test
```

Endpoints, all under `flybot/workflow/`: `state`, `templates` (read), `create`, `assign`, `concurrency`, `advance`, `reset`, `say`, `graph`, `ingest`.
Endpoints, all under `flybot/workflow/`: `state`, `templates` (read), `create`, `assign`, `concurrency`, `advance`, `reset`, `graph`, `ingest`.

## 中文

Expand Down
1 change: 0 additions & 1 deletion plugins/workflow/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,6 @@ export function apply(ctx, config = {}) {
fb.api.handle('workflow/reset', body => service.reset(body)),
fb.api.handle('workflow/pause', body => service.pause(body)),
fb.api.handle('workflow/resume', body => service.resume(body)),
fb.api.handle('workflow/say', body => service.say(body)),
fb.api.handle('workflow/graph', body => service.saveGraph(body)),
fb.api.handle('workflow/templates', () => ({ ok: true, templates: service.state().templates }), { read: true }),
fb.api.handle('workflow/ingest', body => service.ingest(body)),
Expand Down
2 changes: 1 addition & 1 deletion plugins/workflow/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
"cordis.patch.yml"
],
"scripts": {
"test": "node --test \"test/*.test.js\""
"test": "node --import ./test/resolve.mjs --test \"test/*.test.js\""
},
"peerDependencies": {
"ds-bot": ">=0.1.5-flybot.1"
Expand Down
15 changes: 0 additions & 15 deletions plugins/workflow/src/service.js
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,7 @@
const missing = unmetRequired(item, id, run.produced, run.task)
if (missing.length === 0) { started.push(id); continue }
const preds = item.edges.filter(edge => edge.to === id && edge.loop !== true).map(edge => edge.from)
const stuck = preds.length === 0 || preds.every(pred => status[pred] === 'done' || status[pred] === 'failed' || status[pred] === 'skipped')

Check warning on line 266 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-length-check)

Found a useless array length check
if (stuck) {
status[id] = 'failed'
blocked.push({ id, missing })
Expand All @@ -276,7 +276,7 @@
function applyOp(item, run, op) {
let working = run
if (op.type === 'complete' && op.outputs && typeof op.outputs === 'object') {
working = { ...run, produced: { ...(run.produced ?? {}), [op.nodeId]: op.outputs } }

Check warning on line 279 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
}
let result
if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
Expand Down Expand Up @@ -386,7 +386,7 @@
push(item.id, [{ kind: 'system', tone: 'bad', text: `${node.name} 没有接到这一步(${code})。` }], now())
continue
}
current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }

Check warning on line 389 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
storeRun(item.id, current)
}
write()
Expand Down Expand Up @@ -935,21 +935,6 @@
return present()
})
},
say(body = {}) {
return mutate(String(body.workflowId ?? ''), (item) => {
const text = cleanLine(body.text, 2000)
if (!text) throw new Error('A message needs text')
const at = now()
const lines = [{ kind: 'user', text }]
const run = displayRun(item.id)
const running = run?.live === true ? orderedIds(item, run.nodeStatus, 'running')[0] : undefined
if (running) {
const node = nodeById(item, running)
lines.push({ kind: 'bot', botId: running, step: '进行中', text: `${node?.name ?? running}:收到。我继续处理「${run.task?.title ?? ''}」。` })
}
push(item.id, lines, at)
})
},
// Replace the graph the canvas and the run share. Positions and edges
// persist; a node that remains keeps its run status, a new one waits.
saveGraph(body = {}) {
Expand Down Expand Up @@ -1095,7 +1080,7 @@
const runs = []
let endedLive = false
for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) {
for (const run of [...(lane.active ?? [])]) {

Check warning on line 1083 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (run?.live === true && run.status === 'running') {
endRun(workflow(workflowId), run)
archiveCutoffRun(workflowId, run)
Expand Down
13 changes: 2 additions & 11 deletions plugins/workflow/test/assign.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -122,21 +122,11 @@ test('create seats a local Bot and refuses a remote one', () => {
}).ok, false)
})

test('a read-only host refuses assign, and a say lands in the workflow chat', () => {
test('a read-only host refuses assign', () => {
const svc = service({ readOnly: () => true })
const refused = svc.assign({ workflowId: DELIVERY_ID, taskId: '#1', title: 'No' })
assert.equal(refused.ok, false)
assert.equal(refused.code, 'read-only')

const live = service()
const said = live.say({ workflowId: DELIVERY_ID, text: '请把 notice 再短一点' })
const messages = delivery(said).messages
assert.ok(messages.some(message => message.kind === 'user' && message.text.includes('请把 notice 再短一点')))
live.assign({ workflowId: DELIVERY_ID, taskId: '#say', title: 'Notice tweak' })
const withRun = live.say({ workflowId: DELIVERY_ID, text: '继续' })
const tail = delivery(withRun).messages
assert.equal(tail.at(-2).kind, 'user')
assert.match(tail.at(-1).text, /Planner|收到/)
})

test('the store reloads a workflow file and drops a file that does not parse', () => {
Expand Down Expand Up @@ -298,6 +288,7 @@ test('apply registers the workflow endpoints and stays quiet when disabled', ()
assert.ok(names.some(([name]) => name === 'workflow/assign'))
assert.ok(names.some(([name]) => name === 'workflow/graph'))
assert.ok(names.some(([name, read]) => name === 'workflow/templates' && read))
assert.ok(!names.some(([name]) => name === 'workflow/say'), 'the removed say endpoint is not registered')
assert.equal(writes.length, 1)
assert.deepEqual(writes[0]({ id: '#9', title: 'please', status: 'open', assignee: 'Coder' }), { ownerBotId: 'bot-coder' })
assert.equal(writes[0]({ id: '#10', title: 'nothing', status: 'open' }), null)
Expand Down
58 changes: 58 additions & 0 deletions plugins/workflow/test/fake-cordis.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
// A minimal Cordis-like context for @flybot/workflow's host test, shipped with the plugin (#34):
// enough of the services ds-bot's apply() needs to boot, plus ctx.plugin(plugin, config),
// which runs a plugin in its own scope (its ctx.effect cleanups run when that scope is disposed)
// and only after the services in plugin.inject are provided. No ds-bot test helpers are used.
export function fakeCordis() {
const services = {}
const disposers = []
const registeredTools = new Map()
const listeners = new Map()
const scope = (cleanups) => ({
logger: { warn() {}, info() {}, debug() {}, error() {} },
effect(execute) {
const dispose = execute()
if (typeof dispose === 'function') cleanups.push(dispose)
return dispose
},
})
const ctx = {
...scope(disposers),
registeredTools,
get: name => services[name],
set(name, value) { services[name] = value; ctx[name] = value },
provide(name, value) { ctx.set(name, value) },
on(name, listener) {
listeners.set(name, [...(listeners.get(name) ?? []), listener])
return () => listeners.set(name, (listeners.get(name) ?? []).filter(l => l !== listener))
},
emit(name, payload) { for (const l of listeners.get(name) ?? []) if (l.length < 2) l(payload) },
tools: {
register(tool) { registeredTools.set(tool.name, tool); return () => registeredTools.delete(tool.name) },
guard() { return () => {} },
schemas: () => [],
},
systemPrompt: { section() { return () => {} } },
connection: { fetch: { register() { return () => {} } } },
llm: { listProviders: () => [] },
// Loads a plugin as Cordis would: in a child scope that sees the parent's services.
plugin(plugin, config) {
const missing = (plugin.inject ?? []).filter(name => services[name] === undefined)
if (missing.length) throw new Error(`${plugin.name}: missing services ${missing.join(', ')}`)
const cleanups = []
const child = Object.assign(Object.create(ctx), scope(cleanups))
let active = true
const dispose = () => {
if (!active) return
active = false
for (const d of cleanups.splice(0).reverse()) d()
}
plugin.apply(child, config)
disposers.push(dispose)
return { dispose }
},
dispose() {
return Promise.allSettled(disposers.splice(0).reverse().map(dispose => dispose()))
},
}
return ctx
}
57 changes: 57 additions & 0 deletions plugins/workflow/test/host.test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
// @flybot/workflow against the real ctx.flybot host (ds-bot's hooks.js), not the mock
// (ThinkFlowLab/FlyBot#34). #1475 removed the dead `workflow/say` surface: the bot route strips
// its `flybot/` prefix and asks the host for `workflow/say` (core/src/host/channel.js:701), and a
// name no plugin registered is answered by the host itself (core/src/host/hooks.js:1044). So the
// removed endpoint must come back as `Unknown FlyBot endpoint workflow/say` with code
// `bot/not-found` — never `{ ok: true }`, never an empty object.
import assert from 'node:assert/strict'
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { test } from 'node:test'
import { apply } from 'ds-bot'
import * as workflow from '../index.js'
import { fakeCordis } from './fake-cordis.js'

const team = () => ({
version: 1,
mainBotIds: ['a'],
bots: {
a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 },
b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 },
},
rooms: {}, agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, revision: 0,
})

async function start() {
const home = mkdtempSync(join(tmpdir(), 'flybot-workflow-host-'))
writeFileSync(join(home, 'state.json'), JSON.stringify(team()))
const ctx = fakeCordis()
Object.assign(ctx, {
llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} },
sessionController: { create: async () => ({ sessionId: 's1' }), resolveAgent: async () => undefined, selectModel: async () => {}, page: async () => ({ records: [], hasMore: false }) },
sessionTitle: { rename: async () => {} },
sessions: { flush: async () => {} },
})
let route
const register = ctx.connection.fetch.register
ctx.connection.fetch.register = (options) => { if (options.path.endsWith('bot')) route = options; return register(options) }
apply(ctx, { home })
const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', { method: 'POST', body: JSON.stringify({ endpoint, payload }) }))).json()
const stop = async () => { await ctx.dispose(); rmSync(home, { recursive: true, force: true, maxRetries: 5 }) }
return { ctx, call, stop }
}

test('workflow plugin on the real host: the removed say endpoint is unknown, not an empty answer', async (t) => {
const h = await start()
t.after(h.stop)
h.ctx.plugin(workflow, { persist: false })
const state = await h.call('flybot/workflow/state')
assert.equal(state.ok, true, 'the plugin is loaded, so the name is the only thing missing')
assert.ok(state.value.workflows.length > 0)
assert.deepEqual(await h.call('flybot/workflow/say', { workflowId: 'delivery', text: 'hi' }), {
ok: false,
error: { code: 'bot/not-found', message: 'Unknown FlyBot endpoint workflow/say', details: {} },
})
})
34 changes: 34 additions & 0 deletions plugins/workflow/test/resolve.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
// Test-only module resolution for @flybot/workflow (ThinkFlowLab/FlyBot#34).
// - `ds-bot` resolves to the FlyBot core checkout, the way a DSH profile supplies the peer:
// FLYBOT_CORE, else the repo-root `core/` (pristine ds-bot/ + patches/ds-bot/, generated by
// scripts/ds-bot-patches.mjs). Never `ds-bot/`: that is unpatched upstream, with no flybot API.
// - `@deepseek-ai/*` imported from inside that core resolves from a DeepSeek Harness:
// DSH_HARNESS, else <core>/dev/harness (`npm run harness` in the core).
// Workflow's own code never imports ds-bot; only test/host.test.js does, by its public entry.
import { existsSync } from 'node:fs'
import { registerHooks } from 'node:module'
import { join, resolve } from 'node:path'
import { pathToFileURL } from 'node:url'

const repo = resolve(import.meta.dirname, '../../..')
const core = process.env.FLYBOT_CORE
? resolve(process.env.FLYBOT_CORE)
: join(repo, 'core')
if (!existsSync(join(core, 'package.json'))) throw new Error(`@flybot/workflow tests: no FlyBot core at ${core} (set FLYBOT_CORE)`)
const harness = process.env.DSH_HARNESS ?? join(core, 'dev', 'harness')
process.env.DSH_HARNESS = harness
process.env.FLYBOT_CORE = core // tests read interface-layer files (flybot-registry.js) from here
const base = resolve(harness, 'packages/bundle/base/package.json')
const anchor = pathToFileURL(existsSync(base) ? base : resolve(harness, 'package.json')).href
const coreUrl = `${pathToFileURL(core).href}/`
const entry = pathToFileURL(join(core, 'index.js')).href

registerHooks({
resolve(specifier, context, nextResolve) {
if (specifier === 'ds-bot') return { url: entry, format: 'module', shortCircuit: true }
if (specifier.startsWith('@deepseek-ai/') && context.parentURL?.startsWith(coreUrl)) {
return nextResolve(specifier, { ...context, parentURL: anchor })
}
return nextResolve(specifier, context)
},
})
Loading