import assert from 'node:assert/strict'; import {EventEmitter} from 'node:events'; import {createRequire, isBuiltin} from 'node:module'; import {fileURLToPath} from 'node:url'; import test from 'node:test'; import {build} from 'esbuild'; const root = fileURLToPath(new URL('../', import.meta.url)); const compiled = await build({ stdin: {contents: `export * from './bot/workflow/runtime/states_manager.ts'; export * from './bot/workflow/runtime/summary.ts'; export * from './bot/forward/stream.ts';`, resolveDir: root, loader: 'ts'}, bundle: true, platform: 'node', format: 'cjs', packages: 'external', write: false, plugins: [{name: 'isolated-registry', setup(build) { build.onResolve({filter: /^\.\.\/instance\/index\.ts$/}, () => ({path: 'registry', namespace: 'mock'})); build.onLoad({filter: /.*/, namespace: 'mock'}, () => ({loader: 'js', contents: 'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};'})); }}], }); const module = {exports: {}}; const require = createRequire(import.meta.url); new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(specifier => { assert.ok(isBuiltin(specifier) || specifier === 'xstate', `Unexpected workflow history dependency: ${specifier}`); return require(specifier); }, module, module.exports); const {create_workflow_manager, summarize_run, BotStream} = module.exports; const turn = () => new Promise(resolve => setImmediate(resolve)); async function until(check) { for (let index = 0; index < 100; index++) { if (check()) return; await turn(); } assert.fail('Expected workflow history state was not reached'); } const ended = snapshot => snapshot?.ended_at !== null && snapshot?.ended_at !== undefined; const log = id => ({id, type: 'log', config: {level: 'info', message: {kind: 'literal', value: 'done'}}}); const sleep = id => ({id, type: 'sleep', config: {mode: 'ticks', value: 1}}); function definition(id = 'one', extra = {}) { return {schema_version: 1, id, name: id, enabled: true, start: {type: 'common', auto_start: 'never'}, entry: 'entry', nodes: [log('entry')], edges: [], policy: {on_busy: 'queue', max_queue: 10, max_steps: 100, timeout_ms: 5000}, ...extra}; } function fixture(t, configs = [definition()]) { const bot = Object.assign(new EventEmitter(), {entity: {id: 1}, chat() {}}); const context = Object.assign(new EventEmitter(), {bot_instance: bot, bot_status: 'online', info_manager: null, function_manager: {run: async () => true}, logger: null, log() {}}); const manager = create_workflow_manager(context, configs); context.workflow_manager = manager; const stream = new BotStream(1, {source: {get: () => context, observe: () => () => {}}}); t.after(async () => {stream.dispose(); await manager.dispose();}); const detail = run_id => stream.workflow_snapshot({run_id}, Date.now() + 5000); return {bot, context, manager, stream, detail}; } test('completed chat and message runs retain their actual normalized trigger values', async t => { const app = fixture(t, ['chat', 'message', 'custom'].map(event => definition(event, {start: {type: 'event', event}}))); const chat_args = ['Alice', 'hello', {nested: [false, 0, null, {value: 'original'}]}]; class ChatMessage {constructor() {this.json = {text: ' message body'};}} app.context.emit('chat', ...chat_args); app.context.emit('message', new ChatMessage(), 'chat', 'sender-id', false); app.context.emit('custom', false, 0, null, {nested: [false, 0, null]}); await until(() => app.manager.get_snapshots().length === 3 && app.manager.get_snapshots().every(ended)); for (const snapshot of app.manager.get_snapshots()) { const expected = { chat: {event: 'chat', args: chat_args, username: 'Alice', message: 'hello'}, message: {event: 'message', username: 'Bob', message: 'message body', plainText: ' message body', position: 'chat', sender: 'sender-id', verified: false}, custom: {event: 'custom', args: [false, 0, null, {nested: [false, 0, null]}]}, }[snapshot.workflow_id]; assert.equal(snapshot.status, 'completed'); const detail = app.detail(snapshot.run_id); assert.deepEqual(detail.trigger, expected); assert.deepEqual(detail.branches[0].trigger, expected); assert.equal(detail.trigger_available, true); assert.deepEqual(JSON.parse(JSON.stringify(detail.trigger)), expected); } const summaries = app.stream.workflow_snapshot({}, Date.now() + 5000); assert.equal(summaries.length, 3); for (const summary of summaries) { assert.equal(summary.trigger_available, true); assert.equal(Object.hasOwn(summary, 'trigger'), false); assert.equal(Object.hasOwn(summary, 'branches'), false); } }); test('queued chat events keep their original argument objects after the emitter mutates them', async t => { const app = fixture(t, [definition('chat', {start: {type: 'event', event: 'chat'}, nodes: [sleep('entry')]})]); app.context.emit('chat', 'Alice', 'first', {nested: {value: 1}}); const first = app.manager.get_snapshots()[0]; const args = ['Bob', 'queued', {nested: {value: 0}, values: [false, null]}]; const original = structuredClone(args); app.context.emit('chat', ...args); assert.equal(app.manager.get_runtime_snapshot().workflows[0].queued_count, 1); args[0] = 'mutated'; args[1] = 'changed'; args[2].nested.value = 99; args[2].values.push(true); await app.manager.stop(first.run_id); await until(() => app.manager.get_snapshots().length === 2 && app.bot.listenerCount('physicsTick') === 1); const queued = app.manager.get_snapshots().find(snapshot => snapshot.run_id !== first.run_id); const expected = {event: 'chat', args: original, username: 'Bob', message: 'queued'}; assert.deepEqual(app.detail(queued.run_id).trigger, expected); app.bot.emit('physicsTick'); await until(() => ended(app.manager.get_snapshot(queued.run_id))); assert.deepEqual(app.detail(queued.run_id).trigger, expected); assert.deepEqual(app.detail(queued.run_id).branches[0].trigger, expected); }); test('failed and cancelled runs retain triggers even with no terminal branches', async t => { for (const status of ['failed', 'cancelled']) { await t.test(status, async t => { const app = fixture(t, [definition('one', {nodes: [sleep('entry'), log('next')], edges: [{id: 'next', source: 'entry', port: 'success', target: 'next'}], policy: {on_busy: 'queue', max_queue: 10, max_steps: 1, timeout_ms: 5000}})]); const trigger = {event: 'custom', args: [false, 0, null, {nested: ['original']}]}; const run = app.manager.start('one', trigger); const running = app.detail(run.run_id); assert.deepEqual(running.branches, []); assert.deepEqual(running.trigger, trigger); assert.equal(running.trigger_available, true); await until(() => app.bot.listenerCount('physicsTick') === 1); if (status === 'cancelled') await app.manager.stop(run.run_id); else app.bot.emit('physicsTick'); await until(() => ended(app.manager.get_snapshot(run.run_id))); const detail = app.detail(run.run_id); assert.equal(detail.status, status); assert.deepEqual(detail.branches, []); assert.deepEqual(detail.trigger, trigger); assert.equal(detail.trigger_available, true); if (status === 'failed') assert.equal(detail.error.code, 'STEP_LIMIT_EXCEEDED'); }); } }); test('trigger inputs, observer events, run details and branch triggers never share mutations', async t => { const app = fixture(t, [definition('one', {nodes: [sleep('entry')]})]); const trigger = {nested: {values: [false, 0, null]}}; const original = structuredClone(trigger); const observed = []; app.manager.on_update(event => { if (event.type === 'run_updated') event.snapshot.trigger.nested.values.push('observer mutation'); }); app.manager.on_update(event => {if (event.type === 'run_updated') observed.push(event.snapshot);}); const run = app.manager.start('one', trigger); trigger.nested.values.push('input mutation'); const running = app.manager.get_snapshot(run.run_id); running.trigger.nested.values.push('snapshot mutation'); assert.deepEqual(app.manager.get_snapshot(run.run_id).trigger, original); await until(() => app.bot.listenerCount('physicsTick') === 1); app.bot.emit('physicsTick'); await until(() => ended(app.manager.get_snapshot(run.run_id))); await turn(); assert.ok(observed.length > 0); assert.ok(observed.every(snapshot => JSON.stringify(snapshot.trigger) === JSON.stringify(original))); const detail = app.detail(run.run_id); detail.trigger.nested.values.push('detail mutation'); assert.deepEqual(detail.branches[0].trigger, original); detail.branches[0].trigger.nested.values.push('branch mutation'); const list = app.manager.get_snapshots(); list[0].trigger.nested.values.push('list mutation'); assert.deepEqual(app.detail(run.run_id).trigger, original); assert.deepEqual(app.detail(run.run_id).branches[0].trigger, original); }); test('summaries mark actual trigger presence including empty input and legacy missing triggers', () => { const legacy = {run_id: 'legacy', branches: [{trigger: {old: true}}], trigger_available: true}; assert.deepEqual(summarize_run(legacy), {run_id: 'legacy', trigger_available: false}); assert.deepEqual(summarize_run({...legacy, trigger: {}, trigger_available: false}), {run_id: 'legacy', trigger_available: true}); }); test('history keeps the latest 20 completed details while evicted run ids return null', async t => { const app = fixture(t, [definition(), definition('active', {nodes: [sleep('entry')]})]); const active = app.manager.start('active', {active: true}); const ids = []; for (let index = 0; index < 25; index++) { const run = app.manager.start('one', {index, values: [false, 0, null]}); ids.push(run.run_id); await until(() => ended(app.manager.get_snapshot(run.run_id))); await turn(); } const completed = app.manager.get_snapshots().filter(ended); assert.deepEqual(completed.map(snapshot => snapshot.run_id), ids.slice(-20)); assert.equal(app.detail(active.run_id).status, 'running'); for (const id of ids.slice(0, 5)) assert.equal(app.detail(id), null); for (let index = 5; index < ids.length; index++) assert.deepEqual(app.detail(ids[index]).trigger, {index, values: [false, 0, null]}); await app.manager.stop(active.run_id); assert.equal(app.manager.get_snapshots().length, 20); assert.equal(app.detail(ids[5]), null); assert.deepEqual(app.detail(active.run_id).trigger, {active: true}); }); test('the 5 MiB history budget evicts older details before reaching the run-count limit', async t => { const app = fixture(t, [definition('one', {initial_vars: {payload: 'x'.repeat(350_000)}})]); const ids = []; for (let index = 0; index < 19; index++) { const run = app.manager.start('one', {index}); ids.push(run.run_id); await until(() => ended(app.manager.get_snapshot(run.run_id))); await turn(); } const retained = app.manager.get_snapshots(); assert.ok(retained.length > 0 && retained.length < 19); assert.ok(retained.reduce((bytes, snapshot) => bytes + Buffer.byteLength(JSON.stringify(snapshot)), 0) <= 5 * 1024 * 1024); assert.deepEqual(retained.map(snapshot => snapshot.run_id), ids.slice(-retained.length)); for (const id of ids.slice(0, -retained.length)) assert.equal(app.detail(id), null); assert.deepEqual(app.detail(ids.at(-1)).trigger, {index: 18}); }); test('oversized terminal history strips branches but preserves its full trigger detail', async t => { const leaves = Array.from({length: 6}, (_, index) => log(`leaf_${index}`)); const app = fixture(t, [definition('one', {initial_vars: {payload: 'x'.repeat(900_000)}, nodes: [log('entry'), ...leaves], edges: leaves.map(node => ({id: node.id, source: 'entry', port: 'success', target: node.id}))})]); const trigger = {event: 'custom', args: [false, 0, null, {nested: ['original']}]}; let terminal; app.manager.on_update(event => { if (event.type === 'run_updated' && ended(event.snapshot)) terminal = event.snapshot; }); const run = app.manager.start('one', trigger); await until(() => terminal); await turn(); assert.equal(terminal.branches.length, 6); assert.ok(Buffer.byteLength(JSON.stringify(terminal)) > 5 * 1024 * 1024); const detail = app.detail(run.run_id); assert.equal(detail.status, 'completed'); assert.deepEqual(detail.branches, []); assert.deepEqual(detail.trigger, trigger); assert.equal(detail.trigger_available, true); });