import assert from 'node:assert/strict';
import {EventEmitter} from 'node:events';
import test from 'node:test';
import {tsImport} from 'tsx/esm/api';

const {create_workflow_manager} = await tsImport('../bot/workflow/runtime/states_manager.ts', import.meta.url);
const {init_workflow} = await tsImport('../bot/workflow/init.ts', import.meta.url);
const {parse_message_trigger} = await tsImport('../bot/workflow/node_action/event_node/message.ts', import.meta.url);

function context(ready = true) {
    const events = new EventEmitter();
    const bot = new EventEmitter();
    if (ready) bot.entity = {id: 1};
    return {
        bot_instance: bot, bot_status: 'online', function_manager: {run: async () => true}, info_manager: null,
        workflow_manager: null, bot_config: {}, logger: null, log() {},
        on: events.on.bind(events), off: events.off.bind(events),
        emit: events.emit.bind(events), events,
    };
}

function definition(id = 'one', node = {id: 'entry', type: 'sleep', config: {mode: 'ticks', value: 1}}, extra = {}) {
    return {
        schema_version: 1, id, name: id, enabled: true, start: {type: 'common', auto_start: 'never'},
        entry: 'entry', nodes: [node], edges: [],
        policy: {on_busy: 'queue', max_queue: 10, max_steps: 100, timeout_ms: 5000}, ...extra,
    };
}

function deferred() {
    let resolve;
    const promise = new Promise(finish => { resolve = finish; });
    return {promise, resolve};
}

function function_definition(id) {
    return definition(id, {id: 'entry', type: 'function', config: {path: 'bot', name: id, args: []}});
}

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 state was not reached');
}
const ended = snapshot => snapshot?.ended_at !== null && snapshot?.ended_at !== undefined;

test('manager validates disabled graphs, duplicate ids and recursive workflow event', () => {
    const ctx = context();
    assert.throws(() => create_workflow_manager(ctx, [definition(), definition()]));
    assert.throws(() => create_workflow_manager(ctx, [definition('off', undefined, {enabled: false, entry: 'missing'})]));
    assert.throws(() => create_workflow_manager(ctx, [definition('off', undefined,
        {enabled: false, start: {type: 'event', event: 'workflow_update'}})]));
    assert.equal(ctx.events.eventNames().length, 0);
});

test('manager validates global and definition concurrency limits before installing listeners', async () => {
    const ctx = context();
    for (const max_concurrent_runs of [0, 257, -1, 1.5, NaN, Infinity, '2']) {
        assert.throws(() => create_workflow_manager(ctx, [definition()], {max_concurrent_runs}),
            {code: 'INVALID_CONCURRENCY_LIMIT'});
        assert.equal(ctx.events.eventNames().length, 0);
    }
    for (const max_concurrent_runs of [0, 257, -1, 1.5, '2', null]) {
        const config = definition();
        config.policy.max_concurrent_runs = max_concurrent_runs;
        assert.throws(() => create_workflow_manager(ctx, [config]), /max_concurrent_runs|invalid concurrent run limit/);
        assert.equal(ctx.events.eventNames().length, 0);
    }
    for (const max_concurrent_runs of [1, 256]) {
        const config = definition();
        config.policy.max_concurrent_runs = max_concurrent_runs;
        const manager = create_workflow_manager(ctx, [config], {max_concurrent_runs});
        await manager.dispose();
        assert.equal(ctx.events.eventNames().length, 0);
    }
});

test('manager requires spawn and resets readiness on status disconnect', async () => {
    const ctx = context(false);
    const manager = create_workflow_manager(ctx, [definition()]);
    try {
        assert.throws(() => manager.start('missing'), {code: 'WORKFLOW_NOT_FOUND'});
        assert.throws(() => manager.start('one'), {code: 'BOT_NOT_READY'});
        ctx.bot_instance.entity = {id: 1};
        assert.throws(() => manager.start('one'), {code: 'BOT_NOT_READY'});
        ctx.emit('spawn');
        const run = manager.start('one');
        assert.equal(run.status, 'started');
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 1);
        ctx.bot_status = 'disconnecting';
        ctx.emit('status_update', 'disconnecting');
        await until(() => ended(manager.get_snapshot(run.run_id)));
        assert.equal(manager.get_snapshot(run.run_id).stop_reason, 'disconnect');
        assert.equal(ctx.bot_instance.listenerCount('physicsTick'), 0);
        ctx.bot_status = 'online';
        ctx.emit('status_update', 'online');
        assert.throws(() => manager.start('one'), {code: 'BOT_NOT_READY'});
    } finally { await manager.dispose(); }
    assert.equal(ctx.events.eventNames().length, 0);
});

test('manager queues or skips repeated definitions while distinct workflows run together', async () => {
    const ctx = context();
    const first = definition('one');
    first.policy.max_queue = 1;
    const skip = definition('skip');
    skip.policy.on_busy = 'skip';
    const manager = create_workflow_manager(ctx, [first, skip]);
    try {
        const running = manager.start('one');
        const trigger = {nested: {value: 7}};
        assert.equal(manager.start('one', trigger).status, 'queued');
        trigger.nested.value = 99;
        assert.equal(manager.start('one').status, 'skipped');
        const distinct = manager.start('skip');
        assert.equal(distinct.status, 'started');
        assert.equal(manager.start('skip').status, 'skipped');
        assert.equal(manager.get_snapshots().length, 2);
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 2);
        await manager.stop(running.run_id);
        await until(() => manager.get_snapshots().length === 3);
        const queued = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'one'
            && snapshot.run_id !== running.run_id);
        assert.equal(ended(manager.get_snapshot(distinct.run_id)), false);
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 2);
        ctx.bot_instance.emit('physicsTick');
        await until(() => ended(manager.get_snapshot(queued.run_id)) && ended(manager.get_snapshot(distinct.run_id)));
        assert.equal(manager.get_snapshot(queued.run_id).branches[0].trigger.nested.value, 7);
    } finally { await manager.dispose(); }
});

test('manager defaults to sixteen concurrent distinct definitions', async () => {
    const ctx = context();
    ctx.bot_instance.setMaxListeners(0);
    const configs = Array.from({length: 17}, (_, index) => definition(`workflow_${index}`));
    const manager = create_workflow_manager(ctx, configs);
    try {
        for (const config of configs.slice(0, 16)) assert.equal(manager.start(config.id).status, 'started');
        assert.equal(manager.start(configs[16].id).status, 'queued');
        assert.equal(manager.get_snapshots().length, 16);
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 16);
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().length === 17
            && ctx.bot_instance.listenerCount('physicsTick') === 1);
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().every(ended));
        assert.ok(manager.get_snapshots().every(snapshot => snapshot.status === 'completed'));
    } finally { await manager.dispose(); }
});

test('distinct concurrent workflows keep triggers, variables and outputs independent', async () => {
    const ctx = context();
    const releases = {one: deferred(), two: deferred()};
    const calls = [];
    ctx.info_manager = {get_info: name => ({value: {owner: name}})};
    ctx.function_manager = {async run(_path, name, args, check, {signal}) {
        calls.push({name, args, signal});
        await releases[name].promise;
        check();
        return true;
    }};
    const configs = ['one', 'two'].map(id => definition(id, undefined, {
        initial_vars: {owner: id},
        nodes: [
            {id: 'entry', type: 'query', config: {source: 'info', save_as: 'saved',
                args: {name: {kind: 'ref', root: 'trigger', path: ['nested', 'owner']}}}},
            {id: 'use', type: 'function', config: {path: 'bot', name: id, args: [
                {kind: 'ref', root: 'trigger', path: ['nested']},
                {kind: 'ref', root: 'vars', path: ['owner']},
                {kind: 'ref', root: 'vars', path: ['saved']},
                {kind: 'ref', root: 'outputs', path: ['entry', 'output']},
            ]}},
        ],
        edges: [{id: 'use_result', source: 'entry', port: 'success', target: 'use'}],
    }));
    const manager = create_workflow_manager(ctx, configs);
    try {
        const trigger = {nested: {owner: 'one'}};
        const one = manager.start('one', trigger);
        trigger.nested.owner = 'two';
        const two = manager.start('two', trigger);
        trigger.nested.owner = 'mutated';
        assert.equal(one.status, 'started');
        assert.equal(two.status, 'started');
        assert.notEqual(one.run_id, two.run_id);
        await until(() => calls.length === 2);
        for (const call of calls) {
            assert.deepEqual(call.args, [{owner: call.name}, call.name, {owner: call.name}, {owner: call.name}]);
            assert.equal(call.signal.aborted, false);
        }
        assert.notEqual(calls[0].signal, calls[1].signal);
        releases.two.resolve();
        await until(() => ended(manager.get_snapshot(two.run_id)));
        assert.equal(ended(manager.get_snapshot(one.run_id)), false);
        releases.one.resolve();
        await until(() => ended(manager.get_snapshot(one.run_id)));
        for (const run of [one, two]) {
            const snapshot = manager.get_snapshot(run.run_id);
            assert.equal(snapshot.status, 'completed');
            const branch = snapshot.branches[0];
            assert.deepEqual(branch.trigger, {nested: {owner: snapshot.workflow_id}});
            assert.deepEqual(branch.vars, {owner: snapshot.workflow_id, saved: {owner: snapshot.workflow_id}});
            assert.deepEqual(branch.outputs.entry.output, {owner: snapshot.workflow_id});
            assert.equal(branch.outputs.use.output, true);
        }
    } finally {
        Object.values(releases).forEach(release => release.resolve());
        await manager.dispose();
    }
});

test('definition concurrency two overlaps two runs and queues the next instance', async () => {
    const ctx = context();
    const config = definition();
    config.policy.max_concurrent_runs = 2;
    config.policy.max_queue = 1;
    const manager = create_workflow_manager(ctx, [config]);
    try {
        const first = manager.start('one', {instance: 1});
        const second = manager.start('one', {instance: 2});
        assert.equal(first.status, 'started');
        assert.equal(second.status, 'started');
        assert.notEqual(first.run_id, second.run_id);
        assert.equal(manager.start('one', {instance: 3}).status, 'queued');
        assert.equal(manager.start('one', {instance: 4}).status, 'skipped');
        assert.equal(manager.get_snapshots().length, 2);
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 2);
        await manager.stop(first.run_id);
        await until(() => manager.get_snapshots().length === 3
            && ctx.bot_instance.listenerCount('physicsTick') === 2);
        assert.equal(ended(manager.get_snapshot(second.run_id)), false);
        const queued = manager.get_snapshots().find(snapshot => ![first.run_id, second.run_id].includes(snapshot.run_id));
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().every(ended));
        assert.equal(manager.get_snapshot(second.run_id).status, 'completed');
        assert.equal(manager.get_snapshot(queued.run_id).status, 'completed');
        assert.equal(manager.get_snapshot(queued.run_id).branches[0].trigger.instance, 3);
    } finally { await manager.dispose(); }
});

test('global concurrency two queues or skips distinct definitions at capacity', async () => {
    const ctx = context();
    const skip = definition('skip');
    skip.policy.on_busy = 'skip';
    const manager = create_workflow_manager(ctx, [definition('one'), definition('two'), definition('three'), skip],
        {max_concurrent_runs: 2});
    try {
        const one = manager.start('one');
        const two = manager.start('two');
        assert.equal(one.status, 'started');
        assert.equal(two.status, 'started');
        assert.equal(manager.start('three').status, 'queued');
        assert.equal(manager.start('skip').status, 'skipped');
        assert.equal(manager.get_snapshots().length, 2);
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 2);
        await manager.stop(one.run_id);
        await until(() => manager.get_snapshots().length === 3
            && ctx.bot_instance.listenerCount('physicsTick') === 2);
        assert.equal(ended(manager.get_snapshot(two.run_id)), false);
        assert.ok(manager.get_snapshots().some(snapshot => snapshot.workflow_id === 'three'));
        assert.ok(manager.get_snapshots().every(snapshot => snapshot.workflow_id !== 'skip'));
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().every(ended));
    } finally { await manager.dispose(); }
});

test('queue pump bypasses a busy definition without losing its earlier queued request', async () => {
    const ctx = context();
    const manager = create_workflow_manager(ctx, [definition('one'), definition('occupied'), definition('two')],
        {max_concurrent_runs: 2});
    try {
        const one = manager.start('one', {instance: 1});
        const occupied = manager.start('occupied');
        assert.equal(manager.start('one', {instance: 2}).status, 'queued');
        assert.equal(manager.start('two').status, 'queued');
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 2);
        await manager.stop(occupied.run_id);
        await until(() => manager.get_snapshots().length === 3
            && ctx.bot_instance.listenerCount('physicsTick') === 2);
        const two = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'two');
        assert.ok(two);
        assert.equal(manager.get_snapshots().filter(snapshot => snapshot.workflow_id === 'one').length, 1);
        assert.equal(ended(manager.get_snapshot(one.run_id)), false);
        await manager.stop(two.run_id);
        await turn();
        assert.equal(manager.get_snapshots().length, 3);
        await manager.stop(one.run_id);
        await until(() => manager.get_snapshots().length === 4
            && ctx.bot_instance.listenerCount('physicsTick') === 1);
        const queued = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'one'
            && snapshot.run_id !== one.run_id);
        ctx.bot_instance.emit('physicsTick');
        await until(() => ended(manager.get_snapshot(queued.run_id)));
        assert.equal(manager.get_snapshot(queued.run_id).branches[0].trigger.instance, 2);
    } finally { await manager.dispose(); }
});

test('manager bounds global queue at 256 and disconnect clears queued work', async () => {
    const ctx = context();
    const config = definition();
    config.policy.max_queue = 1000;
    const manager = create_workflow_manager(ctx, [config]);
    try {
        const run = manager.start('one');
        for (let index = 0; index < 256; index++) assert.equal(manager.start('one').status, 'queued');
        assert.equal(manager.start('one').status, 'skipped');
        ctx.bot_status = 'offline';
        ctx.emit('status_update', 'offline');
        await until(() => ended(manager.get_snapshot(run.run_id)));
        ctx.bot_status = 'online';
        ctx.emit('spawn');
        await turn();
        assert.equal(manager.get_snapshots().length, 1);
    } finally { await manager.dispose(); }
});

test('manager validates bounded JSON triggers without invoking accessors', async () => {
    const manager = create_workflow_manager(context(), [definition()]);
    try {
        let reads = 0;
        const accessor = Object.defineProperty({}, 'value', {enumerable: true, get() { reads++; return 1; }});
        const cycle = {}; cycle.self = cycle;
        for (const trigger of [accessor, cycle, [], null, {text: 'x'.repeat(70_000)}, {value: Infinity}]) {
            assert.throws(() => manager.start('one', trigger));
        }
        assert.equal(reads, 0);
        assert.equal(manager.get_snapshots().length, 0);
    } finally { await manager.dispose(); }
});

test('reentrant observer can launch another definition after the launching gate clears', async () => {
    const ctx = context();
    const manager = create_workflow_manager(ctx, [definition(), definition('two')]);
    let result;
    const off = manager.on_update(() => {
        if (!result) result = manager.start('two', {queued: true});
        throw new Error('broken observer');
    });
    ctx.on('workflow_update', () => { throw new Error('broken context observer'); });
    try {
        const run = manager.start('one');
        assert.equal(result.status, 'queued');
        assert.equal(manager.get_snapshots().length, 1);
        await until(() => manager.get_snapshots().length === 2
            && ctx.bot_instance.listenerCount('physicsTick') === 2);
        assert.equal(ended(manager.get_snapshot(run.run_id)), false);
        assert.ok(manager.get_snapshots().some(snapshot => snapshot.workflow_id === 'two'));
        off();
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().every(ended));
        const queued = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'two');
        assert.deepEqual(queued.branches[0].trigger, {queued: true});
    } finally { await manager.dispose(); }
});

test('terminal snapshots are readable in the final update and retained snapshots are cloned', async () => {
    const ctx = context();
    const manager = create_workflow_manager(ctx, [definition('one', {id: 'entry', type: 'log',
        config: {level: 'info', message: {kind: 'literal', value: 'done'}}})]);
    let terminal;
    manager.on_update(event => {
        if (event.type === 'run_updated' && event.snapshot.ended_at !== null) {
            terminal = manager.get_snapshots().find(snapshot => snapshot.run_id === event.snapshot.run_id);
        }
    });
    try {
        const run = manager.start('one', {value: 1});
        await until(() => terminal);
        assert.equal(terminal.status, 'completed');
        terminal.branches[0].trigger.value = 2;
        assert.equal(manager.get_snapshot(run.run_id).branches[0].trigger.value, 1);
        await manager.stop(run.run_id);
        await assert.rejects(manager.stop('unknown'), {code: 'WORKFLOW_RUN_NOT_FOUND'});
    } finally { await manager.dispose(); }
});

test('function dependencies bind at actual queue start and pass scoped cancellation signal', async () => {
    const ctx = context();
    const calls = [];
    const manager = create_workflow_manager(ctx, [definition(), definition('function', {id: 'entry', type: 'function',
        config: {path: 'bot', name: 'chat', args: [{kind: 'literal', value: 'hello'}]}})], {max_concurrent_runs: 1});
    try {
        const first = manager.start('one');
        manager.start('function');
        ctx.function_manager = {async run(path, name, args, check, options) {
            check(); calls.push({path, name, args, signal: options.signal}); return true;
        }};
        await manager.stop(first.run_id);
        await until(() => calls.length === 1);
        assert.equal(calls[0].path, 'bot');
        assert.equal(calls[0].name, 'chat');
        assert.deepEqual(calls[0].args, ['hello']);
        assert.ok(calls[0].signal instanceof AbortSignal);
    } finally { await manager.dispose(); }
});

test('function cancellation drains before the queued workflow starts', async () => {
    const ctx = context();
    let aborted = false;
    let drained;
    const drain = new Promise(resolve => { drained = resolve; });
    ctx.function_manager = {async run(_path, _name, _args, _check, {signal}) {
        await new Promise(resolve => signal.addEventListener('abort', () => { aborted = true; resolve(); }, {once: true}));
        await drain;
        signal.throwIfAborted();
    }};
    const manager = create_workflow_manager(ctx, [definition(), definition('function', {id: 'entry', type: 'function',
        config: {path: 'bot', name: 'chat', args: []}})], {max_concurrent_runs: 1});
    try {
        const run = manager.start('function');
        await turn();
        manager.start('one');
        let stopped = false;
        const stop = manager.stop(run.run_id).then(() => { stopped = true; });
        await until(() => aborted);
        assert.equal(stopped, false);
        assert.equal(manager.get_snapshots().length, 1);
        drained();
        await stop;
        await until(() => manager.get_snapshots().length === 2);
    } finally { drained(); await manager.dispose(); }
});

test('stopping one concurrent run preserves other signals and holds its slot until drain', async () => {
    const ctx = context();
    const releases = {one: deferred(), two: deferred()};
    const calls = [];
    ctx.function_manager = {async run(_path, name, _args, check, {signal}) {
        calls.push({name, signal});
        await releases[name].promise;
        check();
        return true;
    }};
    const manager = create_workflow_manager(ctx, [function_definition('one'), function_definition('two'),
        definition('queued')], {max_concurrent_runs: 2});
    try {
        const one = manager.start('one');
        const two = manager.start('two');
        assert.equal(manager.start('queued').status, 'queued');
        await until(() => calls.length === 2);
        const first = calls.find(call => call.name === 'one');
        const second = calls.find(call => call.name === 'two');
        assert.notEqual(first.signal, second.signal);
        let stopped = false;
        const stop = manager.stop(one.run_id).then(() => { stopped = true; });
        await until(() => first.signal.aborted);
        await turn();
        assert.equal(stopped, false);
        assert.equal(second.signal.aborted, false);
        assert.equal(ended(manager.get_snapshot(two.run_id)), false);
        assert.equal(manager.get_snapshots().length, 2);
        assert.equal(ctx.bot_instance.listenerCount('physicsTick'), 0);
        releases.one.resolve();
        await stop;
        await until(() => manager.get_snapshots().length === 3
            && ctx.bot_instance.listenerCount('physicsTick') === 1);
        assert.equal(manager.get_snapshot(one.run_id).status, 'cancelled');
        assert.equal(manager.get_snapshot(one.run_id).stop_reason, 'manual');
        assert.equal(second.signal.aborted, false);
        releases.two.resolve();
        await until(() => ended(manager.get_snapshot(two.run_id)));
        assert.equal(manager.get_snapshot(two.run_id).status, 'completed');
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().every(ended));
        assert.equal(manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'queued').status, 'completed');
    } finally {
        Object.values(releases).forEach(release => release.resolve());
        await manager.dispose();
    }
});

test('dispose aborts all concurrent runs and awaits every deferred callback drain', async () => {
    const ctx = context();
    const releases = {one: deferred(), two: deferred()};
    const calls = [];
    ctx.function_manager = {async run(_path, name, _args, check, {signal}) {
        calls.push({name, signal});
        await releases[name].promise;
        check();
        return true;
    }};
    const manager = create_workflow_manager(ctx, [function_definition('one'), function_definition('two'),
        definition('queued')], {max_concurrent_runs: 2});
    try {
        const one = manager.start('one');
        const two = manager.start('two');
        assert.equal(manager.start('queued').status, 'queued');
        await until(() => calls.length === 2);
        let disposed = false;
        const dispose = manager.dispose();
        assert.equal(manager.dispose(), dispose);
        const done = dispose.then(() => { disposed = true; });
        await until(() => calls.every(call => call.signal.aborted));
        assert.equal(disposed, false);
        assert.equal(ctx.events.eventNames().length, 0);
        assert.throws(() => manager.start('one'), {code: 'WORKFLOW_MANAGER_DISPOSED'});
        releases.one.resolve();
        await until(() => ended(manager.get_snapshot(one.run_id)));
        assert.equal(disposed, false);
        assert.equal(ended(manager.get_snapshot(two.run_id)), false);
        releases.two.resolve();
        await done;
        assert.equal(disposed, true);
        assert.equal(manager.get_snapshots().length, 2);
        for (const run of [one, two]) {
            assert.ok(ended(manager.get_snapshot(run.run_id)));
            assert.equal(manager.get_snapshot(run.run_id).status, 'cancelled');
            assert.equal(manager.get_snapshot(run.run_id).stop_reason, 'disposed');
        }
        assert.equal(ctx.bot_instance.listenerCount('physicsTick'), 0);
        assert.equal(ctx.bot_instance.listenerCount('end'), 0);
    } finally {
        Object.values(releases).forEach(release => release.resolve());
        await manager.dispose();
    }
});

test('disconnect cancels every concurrent run and clears queued definitions across respawn', async () => {
    const ctx = context();
    const releases = {one: deferred(), two: deferred()};
    const calls = [];
    ctx.function_manager = {async run(_path, name, _args, check, {signal}) {
        calls.push({name, signal});
        await releases[name].promise;
        check();
        return true;
    }};
    const manager = create_workflow_manager(ctx, [function_definition('one'), function_definition('two'),
        definition('queued')], {max_concurrent_runs: 2});
    try {
        const one = manager.start('one');
        const two = manager.start('two');
        assert.equal(manager.start('queued').status, 'queued');
        assert.equal(manager.start('one').status, 'queued');
        await until(() => calls.length === 2);
        ctx.bot_status = 'offline';
        ctx.emit('status_update', 'offline');
        await until(() => calls.every(call => call.signal.aborted));
        assert.equal(manager.get_snapshots().length, 2);
        assert.ok(manager.get_snapshots().every(snapshot => !ended(snapshot)));
        ctx.bot_status = 'online';
        ctx.emit('status_update', 'online');
        assert.throws(() => manager.start('queued'), {code: 'BOT_NOT_READY'});
        ctx.emit('spawn');
        releases.one.resolve();
        releases.two.resolve();
        await until(() => manager.get_snapshots().every(ended));
        await turn();
        assert.equal(manager.get_snapshots().length, 2);
        assert.equal(calls.length, 2);
        for (const run of [one, two]) {
            assert.equal(manager.get_snapshot(run.run_id).status, 'cancelled');
            assert.equal(manager.get_snapshot(run.run_id).stop_reason, 'disconnect');
        }
        assert.equal(ctx.bot_instance.listenerCount('physicsTick'), 0);
        const fresh = manager.start('queued');
        assert.equal(fresh.status, 'started');
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 1);
        ctx.bot_instance.emit('physicsTick');
        await until(() => ended(manager.get_snapshot(fresh.run_id)));
        assert.equal(manager.get_snapshot(fresh.run_id).status, 'completed');
    } finally {
        Object.values(releases).forEach(release => release.resolve());
        await manager.dispose();
    }
});

test('info query uses the info adapter and unknown sources fail predictably', async () => {
    const ctx = context();
    ctx.info_manager = {get_info: name => name === 'health' ? {value: {health: 20}} : undefined};
    const manager = create_workflow_manager(ctx, ['info', 'missing'].map(source => definition(source,
        {id: 'entry', type: 'query', config: {source, args: {name: {kind: 'literal', value: 'health'}}}})));
    try {
        const run = manager.start('info');
        await until(() => ended(manager.get_snapshot(run.run_id)));
        assert.deepEqual(manager.get_snapshot(run.run_id).branches[0].last_result.output, {health: 20});
        const bad = manager.start('missing');
        await until(() => ended(manager.get_snapshot(bad.run_id)));
        assert.equal(manager.get_snapshot(bad.run_id).error.code, 'UNKNOWN_QUERY_SOURCE');
    } finally { await manager.dispose(); }
});

test('info query metadata validation preserves the argument error code', async () => {
    const ctx = context();
    ctx.info_manager = {get_info: () => ({value: 20})};
    const configs = [{}, {name: {kind: 'literal', value: ''}}, {name: {kind: 'literal', value: 1}}]
        .map((args, index) => definition(`invalid_${index}`, {id: 'entry', type: 'query', config: {source: 'info', args}}));
    const manager = create_workflow_manager(ctx, configs);
    try {
        const runs = configs.map(config => manager.start(config.id));
        await until(() => runs.every(run => ended(manager.get_snapshot(run.run_id))));
        for (const run of runs) assert.equal(manager.get_snapshot(run.run_id).error.code, 'INVALID_QUERY_ARGUMENTS');
    } finally {await manager.dispose();}
});

test('common spawn, message conditions, arbitrary events and any recursion filter work', async () => {
    const ctx = context(false);
    const auto = definition('auto', undefined, {start: {type: 'common', auto_start: 'spawn'}});
    const event = definition('message', undefined, {start: {type: 'event', event: 'message', condition: {
        op: 'eq', left: {kind: 'ref', root: 'trigger', path: ['username']}, right: {kind: 'literal', value: 'Alice'},
    }}});
    const custom = definition('custom', undefined, {start: {type: 'event', event: 'someCustomEvent'}});
    const any = definition('any', undefined, {start: {type: 'event', event: 'any'}});
    const manager = create_workflow_manager(ctx, [auto, event, custom, any]);
    try {
        ctx.bot_instance.entity = {id: 1};
        ctx.emit('spawn');
        assert.equal(manager.get_snapshots()[0].workflow_id, 'auto');
        ctx.emit('message', {text: '<Bob> no'}, 'chat');
        ctx.emit('message', {text: '<Alice> hello'}, 'chat');
        ctx.emit('someCustomEvent', 1, {safe: true}, ctx.bot_instance);
        ctx.emit('any', 'workflow_update', {loop: true});
        ctx.emit('any', 'chat', 'Alice', 'hello');
        await until(() => ctx.bot_instance.listenerCount('physicsTick') === 4);
        assert.equal(manager.get_snapshots().length, 4);
        assert.ok(manager.get_snapshots().every(snapshot => !ended(snapshot)));
        ctx.bot_instance.emit('physicsTick');
        await until(() => manager.get_snapshots().length === 4 && manager.get_snapshots().every(ended));
        const message = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'message');
        assert.equal(message.branches[0].trigger.message, 'hello');
        const custom_trigger = manager.get_snapshots().find(snapshot => snapshot.workflow_id === 'custom').branches[0].trigger;
        assert.deepEqual(custom_trigger.args, [1, {safe: true}, null]);
    } finally { await manager.dispose(); }
});

test('message adapter reads bounded ChatMessage JSON and rejects malformed objects', () => {
    class ChatMessage { constructor() { this.json = {text: '<Alice> hello'}; } }
    assert.deepEqual(parse_message_trigger([new ChatMessage(), 'chat', 'uuid', true]), {
        event: 'message', username: 'Alice', message: 'hello', plainText: '<Alice> hello',
        position: 'chat', sender: 'uuid', verified: true,
    });
    assert.equal(parse_message_trigger([{text: 'x'.repeat(70_000)}]), undefined);
    const cyclic = {}; cyclic.text = cyclic;
    assert.equal(parse_message_trigger([cyclic]), undefined);
});

test('initially ready common start and timer lifetime are disposed', async t => {
    t.mock.timers.enable({apis: ['setTimeout']});
    const ctx = context();
    const config = definition('auto', undefined, {start: {type: 'common', auto_start: 'spawn'}});
    const manager = create_workflow_manager(ctx, [config,
        definition('timer', undefined, {start: {type: 'timer', interval_ms: 10}})]);
    await turn();
    assert.equal(manager.get_snapshots()[0].workflow_id, 'auto');
    t.mock.timers.tick(25);
    await turn();
    await manager.dispose();
    const count = manager.get_snapshots().length;
    t.mock.timers.tick(25);
    assert.equal(manager.get_snapshots().length, count);
    assert.equal(ctx.events.eventNames().length, 0);
    assert.equal(ctx.bot_instance.listenerCount('physicsTick'), 0);
    assert.equal(ctx.bot_instance.listenerCount('end'), 0);
});

test('init assigns before subscriptions and rolls back all listeners on setup failure', async () => {
    const ctx = context();
    ctx.bot_config = {config_revision: 1};
    const original_on = ctx.on;
    let observed = false;
    ctx.on = (event, callback) => {
        assert.ok(ctx.workflow_manager);
        observed = true;
        original_on(event, callback);
        if (event === 'someCustomEvent') throw new Error('listener install failed');
    };
    await assert.rejects(init_workflow(ctx, [definition('bad', undefined,
        {start: {type: 'event', event: 'someCustomEvent'}})]), /listener install failed/);
    assert.equal(observed, true);
    assert.equal(ctx.workflow_manager, null);
    assert.equal(ctx.events.eventNames().length, 0);
});

test('dispose during first-node logging awaits launched run cleanup', async () => {
    const ctx = context();
    let dispose;
    let manager;
    ctx.logger = {info: message => ({message})};
    ctx.log = () => { dispose = manager.dispose(); };
    manager = create_workflow_manager(ctx, [definition('one', {id: 'entry', type: 'log',
        config: {level: 'info', message: {kind: 'literal', value: 'dispose'}}})]);
    const run = manager.start('one');
    await until(() => dispose);
    await dispose;
    assert.ok(ended(manager.get_snapshot(run.run_id)));
    assert.equal(ctx.events.eventNames().length, 0);
    assert.throws(() => manager.start('one'), {code: 'WORKFLOW_MANAGER_DISPOSED'});
});

test('manager disables code by default and trusted host can explicitly enable it', async () => {
    const config = definition('code', {id: 'entry', type: 'code', config: {source: 'return 42;'}});
    const blocked = create_workflow_manager(context(), [config]);
    const trusted = create_workflow_manager(context(), [config], {allow_code: true});
    try {
        const first = blocked.start('code');
        const second = trusted.start('code');
        await until(() => ended(blocked.get_snapshot(first.run_id)));
        assert.equal(blocked.get_snapshot(first.run_id).error.code, 'CODE_EXECUTION_DISABLED');
        await new Promise(resolve => trusted.on_update(event => {
            if (event.type === 'run_updated' && event.snapshot.ended_at !== null) resolve();
        }));
        assert.equal(trusted.get_snapshot(second.run_id).branches[0].last_result.output, 42);
    } finally { await blocked.dispose(); await trusted.dispose(); }
});

test('manager retains only the latest 20 completed runs without evicting active runs', async () => {
    const manager = create_workflow_manager(context(), [definition('one', {id: 'entry', type: 'log',
        config: {level: 'info', message: {kind: 'literal', value: 'done'}}}),
        definition('active', {id: 'entry', type: 'sleep', config: {mode: 'ticks', value: 1000}})]);
    const completed_ids = [];
    try {
        const active = manager.start('active');
        for (let index = 0; index < 25; index++) {
            const run = manager.start('one');
            completed_ids.push(run.run_id);
            await until(() => ended(manager.get_snapshot(run.run_id)));
            await turn();
        }
        const snapshots = manager.get_snapshots();
        assert.equal(snapshots.length, 21);
        assert.deepEqual(snapshots.filter(ended).map(run => run.run_id), completed_ids.slice(-20));
        for (const id of completed_ids.slice(0, 5)) assert.equal(manager.get_snapshot(id), undefined);
        assert.equal(manager.get_snapshot(active.run_id).status, 'running');
        await manager.stop(active.run_id);
        assert.equal(manager.get_snapshots().length, 20);
        assert.equal(manager.get_snapshot(completed_ids[5]), undefined);
        assert.equal(manager.get_snapshot(active.run_id).status, 'cancelled');
    } finally { await manager.dispose(); }
});