import assert from 'node:assert/strict';
import test from 'node:test';
import {setImmediate, setTimeout as delay} from 'node:timers/promises';
import {tsImport} from 'tsx/esm/api';

const {compile_workflow} = await tsImport('../bot/workflow/runtime/compile.ts', import.meta.url);
const {create_workflow_run} = await tsImport('../bot/workflow/runtime/machine.ts', import.meta.url);
const {assert_json, resolve_value} = await tsImport('../bot/workflow/runtime/value.ts', import.meta.url);

const literal = value => ({kind: 'literal', value});
const ref = (root, ...path) => ({kind: 'ref', root, path});
const edge = (id, source, target, port = 'success') => ({id, source, port, target});
const query = (id, args = {}, extra = {}) => ({id, type: 'query', config: {source: id, args, ...extra}});
const fn = (id, args = []) => ({id, type: 'function', config: {path: 'test', name: id, args}});
const merge = (id, mode = 'all') => ({id, type: 'merge', config: {mode}});
const condition = (id, left, right) => ({id, type: 'condition', config: {condition: {op: 'lt', left, right}}});
const deps = overrides => ({
    run_function: async () => true,
    query: async (_source, args) => args,
    sleep: async () => {},
    log() {},
    ...overrides,
});
function definition(nodes, edges = [], extra = {}) {
    return {
        schema_version: 1, id: 'test_workflow', name: 'Workflow test', enabled: true,
        start: {type: 'common', auto_start: 'never'}, entry: nodes[0].id, nodes, edges,
        initial_vars: {}, policy: {on_busy: 'skip', max_queue: 0, max_steps: 100, timeout_ms: 5000},
        ...extra,
    };
}
function start(t, config, dependencies = deps(), trigger = {}, notify) {
    const run = create_workflow_run(compile_workflow(config), dependencies, trigger, notify);
    t.after(() => run.stop());
    return run;
}
function deferred() {
    let resolve;
    const promise = new Promise(finish => {resolve = finish;});
    return {promise, resolve};
}
function until_aborted(signal) {
    return new Promise((resolve, reject) => {
        if (signal.aborted) reject(signal.reason);
        else signal.addEventListener('abort', () => reject(signal.reason), {once: true});
    });
}
function completed(snapshot, branches = 1) {
    assert.equal(snapshot.status, 'completed', JSON.stringify(snapshot.error));
    assert.equal(snapshot.error, null);
    assert.equal(snapshot.branches.length, branches);
    assert.deepEqual(snapshot.active_nodes, []);
    assert.ok(snapshot.ended_at >= snapshot.started_at);
    return snapshot.branches[0];
}

test('workflow preserves the first failure while siblings drain past its deadline', {timeout: 5000}, async t => {
    const abort_seen = deferred();
    const slow_started = deferred();
    const release = deferred();
    const run = start(t, definition([query('fork'), query('bad'), query('slow')], [
        edge('fork_bad', 'fork', 'bad'), edge('fork_slow', 'fork', 'slow'),
    ], {policy: {on_busy: 'skip', max_queue: 0, max_steps: 100, timeout_ms: 100}}), deps({
        query: async (source, _args, signal) => {
            if (source === 'fork') return null;
            if (source === 'bad') {
                await slow_started.promise;
                throw Object.assign(new Error('first failure'), {code: 'FIRST_FAILURE'});
            }
            slow_started.resolve();
            try {await until_aborted(signal);}
            finally {abort_seen.resolve(); await release.promise;}
            return null;
        },
    }));
    t.after(() => release.resolve());
    await abort_seen.promise;
    await delay(125);
    release.resolve();
    const snapshot = await run.done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.stop_reason, null);
    assert.equal(snapshot.error.code, 'FIRST_FAILURE');
});

test('workflow compiles JSON configs by value and resolves trigger, vars, outputs, last_result and inputs', {timeout: 5000}, async t => {
    const config = definition([
        query('read', {value: ref('trigger', 'nested', 'values', 1)}, {save_as: 'saved'}),
        fn('use', [ref('vars', 'saved'), ref('outputs', 'read', 'output'), ref('last_result', 'output'),
            ref('inputs', 'read_use', 'result', 'output'), ref('vars', 'initial')]),
        {id: 'log', type: 'log', config: {level: 'info', message: ref('last_result', 'output')}},
    ], [edge('read_use', 'read', 'use'), edge('use_log', 'use', 'log')], {initial_vars: {initial: 7}});
    const serialized = JSON.parse(JSON.stringify(config));
    const graph = compile_workflow(serialized);
    serialized.nodes[0].config.source = 'changed';
    assert.equal(graph.nodes.get('read').config.source, 'read');
    const calls = [], logs = [], events = [];
    const trigger = {nested: {values: ['first', {answer: 42}]}};
    const run = create_workflow_run(graph, deps({
        query: async (_source, args, signal) => {assert.equal(signal.aborted, false); return args.value;},
        run_function: async (path, name, args, signal) => {
            assert.equal(signal.aborted, false); calls.push({path, name, args}); return true;
        },
        log: (level, message) => logs.push({level, message}),
    }), trigger, event => events.push(event));
    t.after(() => run.stop());
    trigger.nested.values[1].answer = 0;
    const snapshot = await run.done;
    const branch = completed(snapshot);
    assert.deepEqual(calls, [{path: 'test', name: 'use', args: Array(4).fill({answer: 42}).concat(7)}]);
    assert.deepEqual(logs, [{level: 'info', message: 'true'}]);
    assert.deepEqual(branch.vars, {initial: 7, saved: {answer: 42}});
    assert.deepEqual(Object.keys(branch.outputs), ['read', 'use', 'log']);
    assert.equal(snapshot.run_id, run.run_id);
    assert.equal(snapshot.steps, 3);
    assert.deepEqual(events.filter(event => event.type === 'node_finished').map(event => event.node_id), ['read', 'use', 'log']);
    snapshot.branches[0].vars.initial = 999;
    const copy = run.get_snapshot();
    assert.equal(copy.branches[0].vars.initial, 7);
    copy.branches.length = 0;
    assert.equal(run.get_snapshot().branches.length, 1);
});

test('workflow forks every edge on a success port into independent variable and output state', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        query('fork'), query('left', {value: literal('L')}, {save_as: 'left_value'}),
        query('right', {value: literal('R')}, {save_as: 'right_value'}),
    ], [edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right')]),
    deps({query: async (_source, args) => args.value ?? 'root'})).done;
    completed(snapshot, 2);
    const left = snapshot.branches.find(branch => branch.outputs.left);
    const right = snapshot.branches.find(branch => branch.outputs.right);
    assert.deepEqual(left.vars, {left_value: 'L'});
    assert.deepEqual(right.vars, {right_value: 'R'});
    assert.equal(left.outputs.right, undefined);
    assert.equal(right.outputs.left, undefined);
    assert.equal(left.outputs.fork.output, 'root');
    assert.equal(right.outputs.fork.output, 'root');
});

test('workflow cannot resolve a sibling output even after that sibling has finished', {timeout: 5000}, async t => {
    const sibling_finished = deferred();
    const config = definition([
        query('fork'), query('sibling'), query('wait'),
        query('illegal', {value: ref('outputs', 'sibling', 'output')}),
    ], [edge('to_sibling', 'fork', 'sibling'), edge('to_wait', 'fork', 'wait'), edge('to_illegal', 'wait', 'illegal')]);
    const snapshot = await start(t, config, deps({query: async source => {
        if (source === 'wait') await sibling_finished.promise;
        return source;
    }}), {}, event => {
        if (event.type === 'node_finished' && event.node_id === 'sibling') sibling_finished.resolve();
    }).done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'VALUE_NOT_FOUND');
    assert.equal(snapshot.error.node_id, 'illegal');
});

test('workflow all merge exposes inputs by incoming edge id and unions branch outputs and vars', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        query('fork'), query('left', {}, {save_as: 'left_value'}), query('right', {}, {save_as: 'right_value'}),
        merge('join'), query('inspect', {left: ref('outputs', 'left', 'output'), right: ref('outputs', 'right', 'output'),
            inputs: ref('outputs', 'join', 'output')}),
    ], [edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join'), edge('joined', 'join', 'inspect')]),
    deps({query: async (source, args) => source === 'inspect' ? args : source})).done;
    const branch = completed(snapshot);
    assert.deepEqual(branch.vars, {left_value: 'left', right_value: 'right'});
    assert.deepEqual(branch.outputs.join.output, {
        left_join: {source: 'left', port: 'success', result: {status: 'success', output: 'left'}},
        right_join: {source: 'right', port: 'success', result: {status: 'success', output: 'right'}},
    });
    assert.deepEqual(branch.outputs.inspect.output, {left: 'left', right: 'right', inputs: branch.outputs.join.output});
    assert.equal(snapshot.steps, 5);
});

test('workflow any merge emits once per arrival without sharing sibling state', {timeout: 5000}, async t => {
    const events = [];
    const snapshot = await start(t, definition([
        query('fork'), query('left'), query('right'), merge('join', 'any'),
    ], [edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join')]),
    deps({query: async source => source}), {}, event => events.push(event)).done;
    completed(snapshot, 2);
    assert.equal(events.filter(event => event.type === 'node_finished' && event.node_id === 'join').length, 2);
    assert.deepEqual(snapshot.branches.map(branch => Object.keys(branch.outputs.join.output)[0]).sort(), ['left_join', 'right_join']);
    for (const branch of snapshot.branches) assert.notEqual(Boolean(branch.outputs.left), Boolean(branch.outputs.right));
});

test('workflow routes false results and thrown dependency errors through error nodes', {timeout: 5000}, async t => {
    for (const throwing of [false, true]) {
        const calls = [];
        const code = throwing ? 'TEST_FAILURE' : 'FUNCTION_RETURNED_FALSE';
        const snapshot = await start(t, definition([
            fn('fail'), {id: 'catch', type: 'error', config: {codes: [code]}},
            query('recovered', {code: ref('last_result', 'error', 'code')}), query('unexpected'),
        ], [edge('failure', 'fail', 'catch', 'failure'), edge('caught', 'catch', 'recovered', 'error'),
            edge('uncaught', 'catch', 'unexpected', 'unmatched')]), deps({
            run_function: async () => {
                if (throwing) throw Object.assign(new Error('expected dependency error'), {code});
                return false;
            },
            query: async (source, args) => {calls.push(source); return args;},
        })).done;
        const branch = completed(snapshot);
        assert.deepEqual(calls, ['recovered']);
        assert.equal(branch.outputs.fail.status, 'failure');
        assert.deepEqual(branch.outputs.recovered.output, {code});
    }
});

test('workflow goto retries are bounded and exhaustion can be explicitly handled', {timeout: 5000}, async t => {
    let attempts = 0;
    const sleeps = [];
    const snapshot = await start(t, definition([
        fn('attempt'), {id: 'retry', type: 'goto', config: {max_retries: 2, delay_ms: 7}},
        query('exhausted', {code: ref('last_result', 'error', 'code')}),
    ], [edge('failed', 'attempt', 'retry', 'failure'), edge('again', 'retry', 'attempt', 'retry'),
        edge('finished', 'retry', 'exhausted', 'exhausted')]), deps({
        run_function: async () => {attempts++; return false;},
        sleep: async (mode, value, signal) => {assert.equal(signal.aborted, false); sleeps.push([mode, value]);},
    })).done;
    const branch = completed(snapshot);
    assert.equal(attempts, 3);
    assert.deepEqual(sleeps, [['milliseconds', 7], ['milliseconds', 7]]);
    assert.deepEqual(branch.retry_counts, {retry: 2});
    assert.deepEqual(branch.outputs.exhausted.output, {code: 'RETRY_EXHAUSTED'});
    assert.equal(snapshot.steps, 7);
});

test('workflow all merge fails with MERGE_INCOMPLETE for mutually exclusive branches', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        condition('choose', literal(1), literal(2)), query('left'), query('right'), merge('join'),
    ], [edge('yes', 'choose', 'left', 'true'), edge('no', 'choose', 'right', 'false'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join')])).done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'MERGE_INCOMPLETE');
    assert.equal(snapshot.steps, 2);
    assert.deepEqual(snapshot.active_nodes, []);
});

test('workflow conflicting branch variables fail with MERGE_CONFLICT', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        query('fork'), query('left', {value: literal(1)}, {save_as: 'shared'}),
        query('right', {value: literal(2)}, {save_as: 'shared'}), merge('join'),
    ], [edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join')]),
    deps({query: async (_source, args) => args.value ?? null})).done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'MERGE_CONFLICT');
    assert.equal(snapshot.error.node_id, 'join');
    assert.equal(snapshot.steps, 4);
});

test('workflow nested forks join inner branches before the outer all merge', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        query('root'), query('inner_fork'), query('outer_right'), query('inner_left'), query('inner_right'),
        merge('inner_join'), merge('outer_join'),
    ], [edge('to_inner', 'root', 'inner_fork'), edge('to_outer_right', 'root', 'outer_right'),
        edge('to_inner_left', 'inner_fork', 'inner_left'), edge('to_inner_right', 'inner_fork', 'inner_right'),
        edge('inner_left_join', 'inner_left', 'inner_join'), edge('inner_right_join', 'inner_right', 'inner_join'),
        edge('inner_outer', 'inner_join', 'outer_join'), edge('right_outer', 'outer_right', 'outer_join')]),
    deps({query: async source => source})).done;
    const branch = completed(snapshot);
    assert.deepEqual(Object.keys(branch.outputs.inner_join.output).sort(), ['inner_left_join', 'inner_right_join']);
    assert.deepEqual(Object.keys(branch.outputs.outer_join.output).sort(), ['inner_outer', 'right_outer']);
    assert.deepEqual(Object.keys(branch.outputs).sort(), ['inner_fork', 'inner_join', 'inner_left', 'inner_right', 'outer_join', 'outer_right', 'root']);
});

test('workflow repeated fork loops use a fresh all-merge group on every iteration', {timeout: 5000}, async t => {
    const pairs = [];
    const snapshot = await start(t, definition([
        query('fork', {iteration: ref('vars', 'iteration')}),
        query('left', {iteration: ref('outputs', 'fork', 'output')}),
        query('right', {iteration: ref('outputs', 'fork', 'output')}), merge('join'),
        query('advance', {inputs: ref('outputs', 'join', 'output')}, {save_as: 'iteration'}),
        condition('again', ref('vars', 'iteration'), literal(3)), query('done'),
    ], [edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join'), edge('advance', 'join', 'advance'),
        edge('check', 'advance', 'again'), edge('loop', 'again', 'fork', 'true'), edge('finish', 'again', 'done', 'false')],
    {initial_vars: {iteration: 0}}), deps({query: async (source, args) => {
        if (source === 'advance') {
            const pair = [args.inputs.left_join.result.output, args.inputs.right_join.result.output];
            pairs.push(pair); return pair[0] + 1;
        }
        return args.iteration ?? null;
    }})).done;
    const branch = completed(snapshot);
    assert.deepEqual(pairs, [[0, 0], [1, 1], [2, 2]]);
    assert.equal(branch.vars.iteration, 3);
    assert.equal(snapshot.steps, 19);
});

test('workflow overlapping executions of the same fork never cross-pair all-merge inputs', {timeout: 5000}, async t => {
    const arrivals = new Map();
    const all_started = deferred();
    const gates = new Map(['left:1', 'left:2', 'right:1', 'right:2'].map(key => [key, deferred()]));
    const pairs = [];
    const run = start(t, definition([
        query('root'), query('seed_a', {value: literal(1)}), query('seed_b', {value: literal(2)}),
        query('fork', {value: ref('last_result', 'output')}),
        query('left', {value: ref('outputs', 'fork', 'output')}),
        query('right', {value: ref('outputs', 'fork', 'output')}), merge('join'),
        query('collect', {inputs: ref('outputs', 'join', 'output')}),
    ], [edge('seed_a', 'root', 'seed_a'), edge('seed_b', 'root', 'seed_b'),
        edge('a_fork', 'seed_a', 'fork'), edge('b_fork', 'seed_b', 'fork'),
        edge('fork_left', 'fork', 'left'), edge('fork_right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join'), edge('collect', 'join', 'collect')]),
    deps({query: async (source, args, signal) => {
        if (source === 'left' || source === 'right') {
            const key = `${source}:${args.value}`;
            arrivals.set(key, true);
            if (arrivals.size === 4) all_started.resolve();
            await Promise.race([gates.get(key).promise, until_aborted(signal)]);
        }
        if (source === 'collect') {
            const pair = [args.inputs.left_join.result.output, args.inputs.right_join.result.output];
            pairs.push(pair); return pair;
        }
        return args.value ?? null;
    }}));
    await all_started.promise;
    // Release opposite generations first to force a tempting but invalid join.
    for (const key of ['left:1', 'right:2', 'right:1', 'left:2']) {
        gates.get(key).resolve();
        await setImmediate();
    }
    completed(await run.done, 2);
    assert.deepEqual(pairs.sort((a, b) => a[0] - b[0]), [[1, 1], [2, 2]]);
});

test('workflow retry bypassing the join owner preserves generations across repeated all merges', {timeout: 5000}, async t => {
    const gates = new Map(['a:1', 'b:2', 'b:1', 'a:2'].map(key => [key, deferred()]));
    const finished = new Map([...gates.keys()].map(key => [key, deferred()]));
    const retry_started = deferred(), first_retry_join = deferred();
    const joins = [], logs = [];
    let arrivals = 0;
    const config = definition([
        query('root'), query('seed1', {value: literal(1)}), query('seed2', {value: literal(2)}),
        query('s', {value: ref('last_result', 'output')}),
        ...['a', 'b'].map(id => query(id, {value: ref('outputs', 's', 'output'), outputs: ref('outputs')})),
        merge('join'), {id: 'retry', type: 'goto', config: {max_retries: 1, delay_ms: 0}},
        {id: 'end', type: 'log', config: {level: 'info', message: ref('outputs', 's', 'output')}},
    ], [edge('seed1', 'root', 'seed1'), edge('seed2', 'root', 'seed2'),
        edge('seed1_s', 'seed1', 's'), edge('seed2_s', 'seed2', 's'),
        edge('s_a', 's', 'a'), edge('s_b', 's', 'b'), edge('a_join', 'a', 'join'), edge('b_join', 'b', 'join'),
        edge('join_retry', 'join', 'retry'), edge('retry_a', 'retry', 'a', 'retry'),
        edge('retry_b', 'retry', 'b', 'retry'), edge('exhausted', 'retry', 'end', 'exhausted')]);
    assert.equal(compile_workflow(config).join_owners.get('join'), 's');
    const run = start(t, config, deps({query: async (source, args, signal) => {
        if (source !== 'a' && source !== 'b') return args.value ?? null;
        const round = Object.hasOwn(args.outputs, 'retry') ? 2 : 1;
        if (round === 2) {
            if (++arrivals === 4) retry_started.resolve();
            await Promise.race([gates.get(`${source}:${args.value}`).promise, until_aborted(signal)]);
        }
        return {generation: args.value, round};
    }, log: (level, message) => logs.push({level, message})}), {}, event => {
        if (event.type !== 'node_finished') return;
        if (event.node_id === 'join') {
            const inputs = event.result.output;
            joins.push([inputs.a_join.result.output, inputs.b_join.result.output]);
            if (joins.length === 3) first_retry_join.resolve();
        } else if (event.node_id === 'a' || event.node_id === 'b') {
            if (event.result.output.round === 2) {
                finished.get(`${event.node_id}:${event.result.output.generation}`).resolve();
            }
        }
    });
    await retry_started.promise;
    const expected = (generation, round) => Array.from({length: 2}, () => ({generation, round}));
    assert.deepEqual(joins.slice().sort((a, b) => a[0].generation - b[0].generation), [expected(1, 1), expected(2, 1)]);
    // These arrivals would cross-pair if the first join discarded its lineage.
    for (const key of ['a:1', 'b:2']) {
        gates.get(key).resolve(); await finished.get(key).promise;
    }
    await setImmediate();
    assert.equal(joins.length, 2);
    gates.get('b:1').resolve();
    await first_retry_join.promise;
    assert.deepEqual(joins[2], expected(1, 2));
    gates.get('a:2').resolve();
    const snapshot = await run.done;
    completed(snapshot, 2);
    assert.deepEqual(joins[3], expected(2, 2));
    assert.equal(snapshot.steps, 23);
    assert.deepEqual(logs.sort((a, b) => a.message.localeCompare(b.message)),
        [{level: 'info', message: '1'}, {level: 'info', message: '2'}]);
    for (const branch of snapshot.branches) {
        assert.deepEqual(branch.retry_counts, {retry: 1});
        assert.equal(branch.outputs.retry.error.code, 'RETRY_EXHAUSTED');
        assert.equal(branch.outputs.join.output.a_join.result.output.generation, branch.outputs.s.output);
        assert.equal(branch.outputs.join.output.b_join.result.output.generation, branch.outputs.s.output);
    }
});

test('workflow node timeout aborts and drains its dependency before routing failure', {timeout: 5000}, async t => {
    let aborted = false, drained = false;
    const snapshot = await start(t, definition([
        {...query('slow'), timeout_ms: 100}, query('recover', {code: ref('last_result', 'error', 'code')}),
    ], [edge('timeout', 'slow', 'recover', 'failure')]), deps({query: async (source, args, signal) => {
        if (source !== 'slow') {assert.equal(drained, true); return args;}
        try {await until_aborted(signal);} finally {aborted = signal.aborted; await setImmediate(); drained = true;}
    }})).done;
    const branch = completed(snapshot);
    assert.equal(aborted, true);
    assert.equal(drained, true);
    assert.equal(branch.outputs.slow.error.code, 'NODE_TIMEOUT');
    assert.deepEqual(branch.outputs.recover.output, {code: 'NODE_TIMEOUT'});
});

test('workflow partial nested forks retain outer join generations across overlapping executions', {timeout: 5000}, async t => {
    const leaves = ['inner_left', 'inner_right', 'outer_right'];
    const keys = leaves.flatMap(id => [1, 2].map(value => `${id}:${value}`));
    const gates = new Map(keys.map(key => [key, deferred()]));
    const finished = new Map(keys.map(key => [key, deferred()]));
    const all_started = deferred(), first_join = deferred();
    const triples = [];
    let arrivals = 0;
    const run = start(t, definition([
        query('root'), query('seed_a', {value: literal(1)}), query('seed_b', {value: literal(2)}),
        query('outer_fork', {value: ref('last_result', 'output')}),
        query('partial', {value: ref('last_result', 'output')}),
        ...leaves.map(id => query(id, {value: ref('last_result', 'output')})), merge('join'),
        query('collect', {inputs: ref('outputs', 'join', 'output')}),
    ], [edge('seed_a', 'root', 'seed_a'), edge('seed_b', 'root', 'seed_b'),
        edge('a_outer', 'seed_a', 'outer_fork'), edge('b_outer', 'seed_b', 'outer_fork'),
        edge('partial', 'outer_fork', 'partial'), edge('outer_right', 'outer_fork', 'outer_right'),
        edge('inner_left', 'partial', 'inner_left'), edge('inner_right', 'partial', 'inner_right'),
        ...leaves.map(id => edge(`${id}_join`, id, 'join')), edge('collect', 'join', 'collect')]),
    deps({query: async (source, args, signal) => {
        if (leaves.includes(source)) {
            if (++arrivals === 6) all_started.resolve();
            await Promise.race([gates.get(`${source}:${args.value}`).promise, until_aborted(signal)]);
        }
        if (source === 'collect') {
            const triple = leaves.map(id => args.inputs[`${id}_join`].result.output);
            triples.push(triple); first_join.resolve(); return triple;
        }
        return args.value ?? null;
    }}), {}, event => {
        if (event.type === 'node_finished') finished.get(`${event.node_id}:${event.result.output}`)?.resolve();
    });
    await all_started.promise;
    for (const key of ['inner_left:1', 'inner_right:2', 'outer_right:1']) {
        gates.get(key).resolve(); await finished.get(key).promise;
    }
    await setImmediate();
    assert.deepEqual(triples, []);
    gates.get('inner_right:1').resolve();
    await first_join.promise;
    assert.deepEqual(triples, [[1, 1, 1]]);
    gates.get('inner_left:2').resolve(); gates.get('outer_right:2').resolve();
    completed(await run.done, 2);
    assert.deepEqual(triples, [[1, 1, 1], [2, 2, 2]]);
});

test('workflow partial nested forks open fresh outer join generations over repeated rounds', {timeout: 5000}, async t => {
    const leaves = ['inner_left', 'inner_right', 'outer_right'];
    const triples = [];
    const snapshot = await start(t, definition([
        query('outer_fork', {value: ref('vars', 'round')}), query('partial', {value: ref('last_result', 'output')}),
        ...leaves.map(id => query(id, {value: ref('last_result', 'output')})), merge('join'),
        query('advance', {inputs: ref('outputs', 'join', 'output')}, {save_as: 'round'}),
        condition('again', ref('vars', 'round'), literal(3)), query('done'),
    ], [edge('partial', 'outer_fork', 'partial'), edge('outer_right', 'outer_fork', 'outer_right'),
        edge('inner_left', 'partial', 'inner_left'), edge('inner_right', 'partial', 'inner_right'),
        ...leaves.map(id => edge(`${id}_join`, id, 'join')), edge('advance', 'join', 'advance'),
        edge('check', 'advance', 'again'), edge('loop', 'again', 'outer_fork', 'true'), edge('done', 'again', 'done', 'false')],
    {initial_vars: {round: 0}}), deps({query: async (source, args) => {
        if (source === 'advance') {
            const triple = leaves.map(id => args.inputs[`${id}_join`].result.output);
            triples.push(triple); return triple[0] + 1;
        }
        return args.value ?? null;
    }})).done;
    assert.equal(completed(snapshot).vars.round, 3);
    assert.deepEqual(triples, [[0, 0, 0], [1, 1, 1], [2, 2, 2]]);
    assert.equal(snapshot.steps, 25);
});

test('workflow successive nested all joins preserve overlapping fork generations', {timeout: 5000}, async t => {
    const gates = new Map(['x:1', 'x:2', 'c:1', 'c:2'].map(key => [key, deferred()]));
    const finished = new Map([...gates.keys()].map(key => [key, deferred()]));
    const all_started = deferred(), first_join = deferred();
    let arrivals = 0;
    const pairs = [];
    const run = start(t, definition([
        query('root'), query('seed_a', {value: literal(1)}), query('seed_b', {value: literal(2)}),
        query('fork', {value: ref('last_result', 'output')}),
        ...['a', 'b', 'c'].map(id => query(id, {value: ref('outputs', 'fork', 'output')})), merge('inner'),
        query('x', {value: ref('outputs', 'fork', 'output')}), merge('outer'),
        query('collect', {inputs: ref('outputs', 'outer', 'output')}),
    ], [edge('seed_a', 'root', 'seed_a'), edge('seed_b', 'root', 'seed_b'),
        edge('a_fork', 'seed_a', 'fork'), edge('b_fork', 'seed_b', 'fork'),
        ...['a', 'b', 'c'].map(id => edge(`fork_${id}`, 'fork', id)),
        edge('a_inner', 'a', 'inner'), edge('b_inner', 'b', 'inner'), edge('inner_x', 'inner', 'x'),
        edge('x_outer', 'x', 'outer'), edge('c_outer', 'c', 'outer'), edge('collect', 'outer', 'collect')]),
    deps({query: async (source, args, signal) => {
        if (source === 'x' || source === 'c') {
            if (++arrivals === 4) all_started.resolve();
            await Promise.race([gates.get(`${source}:${args.value}`).promise, until_aborted(signal)]);
        }
        if (source === 'collect') {
            const pair = [args.inputs.x_outer.result.output, args.inputs.c_outer.result.output];
            pairs.push(pair); first_join.resolve(); return pair;
        }
        return args.value ?? null;
    }}), {}, event => {
        if (event.type === 'node_finished') finished.get(`${event.node_id}:${event.result.output}`)?.resolve();
    });
    await all_started.promise;
    for (const key of ['x:1', 'c:2']) {
        gates.get(key).resolve(); await finished.get(key).promise;
    }
    await setImmediate();
    assert.deepEqual(pairs, []);
    gates.get('c:1').resolve();
    await first_join.promise;
    assert.deepEqual(pairs, [[1, 1]]);
    gates.get('x:2').resolve();
    completed(await run.done, 2);
    assert.deepEqual(pairs, [[1, 1], [2, 2]]);
});

test('workflow all merge accepts equal branch vars with different object key insertion order', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        query('fork'), query('left', {}, {save_as: 'shared'}), query('right', {}, {save_as: 'shared'}), merge('join'),
    ], [edge('left', 'fork', 'left'), edge('right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join')]), deps({query: async source => {
        if (source === 'left') return {a: 1, nested: {x: 2, y: [3]}};
        if (source === 'right') return {nested: {y: [3], x: 2}, a: 1};
        return null;
    }})).done;
    assert.deepEqual(completed(snapshot).vars.shared, {a: 1, nested: {x: 2, y: [3]}});
});

test('workflow all merge keeps one branch write when its sibling retains the inherited value', {timeout: 5000}, async t => {
    const branch = completed(await start(t, definition([
        query('fork'), query('write', {}, {save_as: 'shared'}), query('unchanged'), merge('join'),
    ], [edge('write', 'fork', 'write'), edge('unchanged', 'fork', 'unchanged'),
        edge('write_join', 'write', 'join'), edge('unchanged_join', 'unchanged', 'join')],
    {initial_vars: {shared: {value: 0}, preserved: true}}),
    deps({query: async source => source === 'write' ? {value: 1} : null})).done);
    assert.deepEqual(branch.vars, {shared: {value: 1}, preserved: true});
});

test('workflow merge conflicts route through merge failure edges with the merge node identity', {timeout: 5000}, async t => {
    const logs = [];
    const branch = completed(await start(t, definition([
        query('fork'), query('left', {}, {save_as: 'shared'}), query('right', {}, {save_as: 'shared'}), merge('join'),
        {id: 'recover', type: 'log', config: {level: 'warning', message: ref('last_result', 'error', 'code')}},
    ], [edge('left', 'fork', 'left'), edge('right', 'fork', 'right'),
        edge('left_join', 'left', 'join'), edge('right_join', 'right', 'join'), edge('recover', 'join', 'recover', 'failure')]),
    deps({query: async source => source,
        log: (level, message) => logs.push({level, message})})).done);
    assert.equal(branch.outputs.join.status, 'failure');
    assert.equal(branch.outputs.join.error.code, 'MERGE_CONFLICT');
    assert.equal(branch.outputs.join.error.node_id, 'join');
    assert.deepEqual(logs, [{level: 'warning', message: 'MERGE_CONFLICT'}]);
});

test('workflow schedules at most 16 dependencies and resumes queued work as slots free', {timeout: 5000}, async t => {
    const gates = Array.from({length: 32}, deferred);
    const started = Array.from({length: 32}, deferred);
    let active = 0, peak = 0, calls = 0;
    const run = start(t, definition([query('fork'), query('work', {inputs: ref('inputs')})],
        gates.map((_, index) => edge(`work_${index}`, 'fork', 'work'))), deps({query: async (source, args, signal) => {
        if (source === 'fork') return null;
        const index = Number(Object.keys(args.inputs)[0].slice('work_'.length));
        calls++; active++; peak = Math.max(peak, active); started[index].resolve();
        try {await Promise.race([gates[index].promise, until_aborted(signal)]); return index;}
        finally {active--;}
    }}));
    await Promise.all(started.slice(0, 16).map(item => item.promise));
    await setImmediate();
    assert.equal(calls, 16);
    assert.equal(active, 16);
    assert.equal(run.get_snapshot().active_nodes.length, 16);
    gates[0].resolve();
    await started[16].promise;
    assert.equal(calls, 17);
    assert.equal(active, 16);
    gates.forEach(item => item.resolve());
    const snapshot = await run.done;
    completed(snapshot, 32);
    assert.equal(peak, 16);
    assert.equal(active, 0);
    assert.equal(calls, 32);
    assert.equal(snapshot.steps, 33);
});

test('workflow token bound accepts 256 branches and rejects 257 before dispatching children', {timeout: 5000}, async t => {
    for (const count of [256, 257]) {
        let children = 0;
        const config = definition([query('fork'), query('leaf')],
            Array.from({length: count}, (_, index) => edge(`leaf_${index}`, 'fork', 'leaf')));
        config.policy.max_steps = 512;
        const snapshot = await start(t, config, deps({query: async source => {
            if (source === 'leaf') children++;
            return null;
        }})).done;
        if (count === 256) {
            completed(snapshot, 256);
            assert.equal(children, 256);
            assert.equal(snapshot.steps, 257);
        } else {
            assert.equal(snapshot.status, 'failed');
            assert.equal(snapshot.error.code, 'BRANCH_LIMIT_EXCEEDED');
            assert.equal(children, 0);
            assert.equal(snapshot.steps, 1);
        }
    }
});

test('workflow token bound includes completed branches and tokens waiting at all merges', {timeout: 5000}, async t => {
    for (const waiting of [false, true]) {
        const nodes = [query('root'), query('leaf'), query('expand'), ...(waiting ? [merge('join')] : [query('sink')])];
        const target = waiting ? 'join' : 'sink';
        const edges = [
            ...Array.from({length: 255}, (_, index) => edge(`leaf_${index}`, 'root', 'leaf')),
            edge('expand', 'root', 'expand'), edge('expanded_a', 'expand', target), edge('expanded_b', 'expand', target),
            ...(waiting ? [edge('leaf_join', 'leaf', 'join')] : []),
        ];
        const config = definition(nodes, edges);
        config.policy.max_steps = 1024;
        const snapshot = await start(t, config).done;
        assert.equal(snapshot.status, 'failed');
        assert.equal(snapshot.error.code, 'BRANCH_LIMIT_EXCEEDED');
        assert.equal(snapshot.error.node_id, 'expand');
        assert.equal(snapshot.steps, 257);
        assert.equal(snapshot.branches.length, waiting ? 0 : 255);
        assert.deepEqual(snapshot.active_nodes, []);
    }
});

test('workflow completion scheduling advances a loop beside a pending sibling and drains at the step limit', {timeout: 5000}, async t => {
    const pending_started = deferred();
    let iterations = 0, drained = false;
    const config = definition([query('fork'), query('pending'), query('loop')],
        [edge('pending', 'fork', 'pending'), edge('loop', 'fork', 'loop'), edge('again', 'loop', 'loop')]);
    config.policy.max_steps = 128;
    const snapshot = await start(t, config, deps({query: async (source, _args, signal) => {
        if (source === 'fork') return null;
        if (source === 'pending') {
            pending_started.resolve();
            try {await until_aborted(signal);} finally {await setImmediate(); drained = true;}
        }
        await pending_started.promise;
        return ++iterations;
    }})).done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'STEP_LIMIT_EXCEEDED');
    assert.equal(snapshot.steps, 128);
    assert.equal(iterations, 126);
    assert.equal(drained, true);
    assert.deepEqual(snapshot.active_nodes, []);
});

test('workflow JSON guards reject prototypes, accessors and cycles without invoking getters', () => {
    let reads = 0;
    const accessor = Object.defineProperty({}, 'secret', {enumerable: true, get() {reads++; return 1;}});
    const cyclic = {}; cyclic.self = cyclic;
    for (const value of [Object.create({inherited: 1}), JSON.parse('{"__proto__":1}'), accessor, cyclic]) {
        assert.throws(() => assert_json(value), {code: 'INVALID_JSON'});
    }
    const state = {trigger: accessor, vars: Object.create({inherited: 1}), outputs: {}, last_result: null, inputs: {}, retry_counts: {}};
    assert.throws(() => resolve_value(ref('trigger', 'secret'), state), {code: 'VALUE_NOT_FOUND'});
    assert.throws(() => resolve_value(ref('vars', 'inherited'), state), {code: 'VALUE_NOT_FOUND'});
    assert.throws(() => resolve_value(ref('vars', '__proto__'), state), {code: 'INVALID_VALUE_SOURCE'});
    assert.throws(() => resolve_value(literal(cyclic), state), {code: 'INVALID_JSON'});
    assert.equal(reads, 0);
    const shared = {value: [null, true, 2, 'json']};
    assert.doesNotThrow(() => assert_json({left: shared, right: shared}));
    assert.deepEqual(resolve_value(ref('trigger', 'value', 2), {...state, trigger: shared}), 2);
});

test('workflow code workers normalize oversized and non-JSON results into routable failures', {timeout: 5000}, async t => {
    for (const source of ['return "x".repeat(1048577);', 'return undefined;', 'return NaN;',
        'const value = {}; value.self = value; return value;']) {
        const snapshot = await start(t, definition([
            {id: 'code', type: 'code', config: {source}}, query('recover', {code: ref('last_result', 'error', 'code')}),
        ], [edge('failure', 'code', 'recover', 'failure')])).done;
        const branch = completed(snapshot);
        assert.equal(branch.outputs.code.status, 'failure');
        assert.equal(branch.outputs.code.output, null);
        assert.equal(branch.outputs.code.error.code, 'INVALID_JSON');
        assert.equal(branch.outputs.code.error.node_id, 'code');
        assert.deepEqual(branch.outputs.recover.output, {code: 'INVALID_JSON'});
    }
});

test('workflow code worker infinite loop times out and routes NODE_TIMEOUT after termination', {timeout: 5000}, async t => {
    const snapshot = await start(t, definition([
        {id: 'code', type: 'code', timeout_ms: 150, config: {source: 'while (true) {}'}},
        query('recover', {code: ref('last_result', 'error', 'code')}),
    ], [edge('timeout', 'code', 'recover', 'failure')])).done;
    const branch = completed(snapshot);
    assert.equal(snapshot.stop_reason, null);
    assert.equal(branch.outputs.code.error.code, 'NODE_TIMEOUT');
    assert.deepEqual(branch.outputs.recover.output, {code: 'NODE_TIMEOUT'});
});

test('workflow run timeout aborts dependencies and finishes as timed_out', {timeout: 5000}, async t => {
    let drained = false;
    const config = definition([query('slow')]);
    config.policy.timeout_ms = 150;
    const snapshot = await start(t, config, deps({query: async (_source, _args, signal) => {
        try {await until_aborted(signal);} finally {await setImmediate(); drained = true;}
    }})).done;
    assert.equal(snapshot.status, 'timed_out');
    assert.equal(snapshot.stop_reason, 'timeout');
    assert.equal(drained, true);
    assert.deepEqual(snapshot.active_nodes, []);
    assert.equal(snapshot.branches.length, 0);
});

test('workflow cancellation drains all cooperative fork dependencies before stop or done resolves', {timeout: 5000}, async t => {
    const started = Array.from({length: 3}, deferred);
    const aborted = Array.from({length: 3}, deferred);
    const cleanup = deferred();
    let active = 0, drained = 0;
    const block = async (index, signal) => {
        active++; started[index].resolve();
        try {await until_aborted(signal);} finally {
            aborted[index].resolve(); await cleanup.promise; active--; drained++;
        }
    };
    t.after(() => cleanup.resolve());
    const run = start(t, definition([
        query('fork'), fn('function_wait'), query('query_wait'),
        {id: 'sleep_wait', type: 'sleep', config: {mode: 'ticks', value: 10}},
    ], [edge('function', 'fork', 'function_wait'), edge('query', 'fork', 'query_wait'), edge('sleep', 'fork', 'sleep_wait')]),
    deps({run_function: (_path, _name, _args, signal) => block(0, signal),
        query: (source, _args, signal) => source === 'fork' ? Promise.resolve(null) : block(1, signal),
        sleep: (_mode, _value, signal) => block(2, signal)}));
    await Promise.all(started.map(item => item.promise));
    let stop_finished = false, done_finished = false;
    const stopping = run.stop('disconnect').then(() => {stop_finished = true;});
    run.done.then(() => {done_finished = true;});
    await Promise.all(aborted.map(item => item.promise));
    await setImmediate();
    assert.equal(run.get_snapshot().status, 'stopping');
    assert.equal(active, 3);
    assert.equal(stop_finished, false);
    assert.equal(done_finished, false);
    cleanup.resolve();
    await stopping;
    const snapshot = await run.done;
    assert.equal(snapshot.status, 'cancelled');
    assert.equal(snapshot.stop_reason, 'disconnect');
    assert.equal(active, 0);
    assert.equal(drained, 3);
    assert.deepEqual(snapshot.active_nodes, []);
    assert.equal(snapshot.branches.length, 0);
    await run.stop('manual');
    assert.equal(run.get_snapshot().stop_reason, 'disconnect');
});

test('workflow unhandled failure aborts and drains sibling work before done resolves', {timeout: 5000}, async t => {
    const started = deferred(), aborted = deferred(), cleanup = deferred();
    let drained = false, settled = false;
    t.after(() => cleanup.resolve());
    const run = start(t, definition([query('fork'), fn('fail'), query('wait')],
        [edge('fail', 'fork', 'fail'), edge('wait', 'fork', 'wait')]), deps({
        run_function: async () => {await started.promise; return false;},
        query: async (source, _args, signal) => {
            if (source === 'fork') return null;
            started.resolve();
            try {await until_aborted(signal);} finally {
                aborted.resolve(); await cleanup.promise; drained = true;
            }
        },
    }));
    run.done.then(() => {settled = true;});
    await aborted.promise;
    await setImmediate();
    assert.equal(settled, false);
    assert.equal(drained, false);
    cleanup.resolve();
    const snapshot = await run.done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'FUNCTION_RETURNED_FALSE');
    assert.equal(drained, true);
    assert.deepEqual(snapshot.active_nodes, []);
    assert.equal(snapshot.branches.length, 1);
    assert.equal(snapshot.branches[0].outputs.wait, undefined);
});

test('workflow max_steps terminates cycles at the configured execution budget', {timeout: 5000}, async t => {
    const config = definition([query('loop')], [edge('again', 'loop', 'loop')]);
    config.policy.max_steps = 5;
    let calls = 0;
    const snapshot = await start(t, config, deps({query: async () => ++calls})).done;
    assert.equal(snapshot.status, 'failed');
    assert.equal(snapshot.error.code, 'STEP_LIMIT_EXCEEDED');
    assert.equal(snapshot.steps, 5);
    assert.equal(calls, 5);
    assert.deepEqual(snapshot.active_nodes, []);
});

test('workflow code worker returns JSON from trigger state', {timeout: 5000}, async t => {
    const branch = completed(await start(t, definition([
        {id: 'code', type: 'code', config: {source: 'return {value: state.trigger.value, initial: state.vars.initial};'}},
    ], [], {initial_vars: {initial: 8}}), deps(), {value: ['worker', 42]}).done);
    assert.deepEqual(branch.outputs.code.output, {value: ['worker', 42], initial: 8});
});

test('workflow cancelling an infinite code loop terminates and drains its worker', {timeout: 5000}, async t => {
    const entered = deferred();
    const run = start(t, definition([
        {id: 'loop', type: 'code', config: {source: 'while (true) {}'}},
    ]), deps(), {}, event => {
        if (event.type === 'node_started' && event.node_id === 'loop') entered.resolve();
    });
    await entered.promise;
    await delay(75);
    assert.equal(run.get_snapshot().status, 'running');
    await run.stop('manual');
    const snapshot = await run.done;
    assert.equal(snapshot.status, 'cancelled');
    assert.deepEqual(snapshot.active_nodes, []);
    assert.equal(snapshot.branches.length, 0);
});

test('workflow schema rejects malformed definitions, references, policies and graph topology', async t => {
    const valid = () => definition([query('first'), query('last')], [edge('next', 'first', 'last')]);
    const cases = [
        ['schema version', value => {value.schema_version = 2;}],
        ['duplicate node', value => {value.nodes.push(structuredClone(value.nodes[0]));}],
        ['duplicate edge', value => {value.edges.push(structuredClone(value.edges[0]));}],
        ['dangling source', value => {value.edges[0].source = 'missing';}],
        ['dangling target', value => {value.edges[0].target = 'missing';}],
        ['missing entry', value => {value.entry = 'missing';}],
        ['unreachable node', value => {value.nodes.push(query('unreachable'));}],
        ['invalid port', value => {value.edges[0].port = 'true';}],
        ['unknown node type', value => {value.nodes[0].type = 'unknown';}],
        ['zero step budget', value => {value.policy.max_steps = 0;}],
        ['fractional step budget', value => {value.policy.max_steps = 1.5;}],
        ['negative queue', value => {value.policy.max_queue = -1;}],
        ['invalid busy policy', value => {value.policy.on_busy = 'parallel';}],
        ['zero run timeout', value => {value.policy.timeout_ms = 0;}],
        ['zero node timeout', value => {value.nodes[0].timeout_ms = 0;}],
        ['invalid reference root', value => {value.nodes[0].config.args.x = ref('unknown');}],
        ['unsafe reference path', value => {value.nodes[0].config.args.x = ref('vars', '__proto__');}],
        ['negative reference index', value => {value.nodes[0].config.args.x = ref('trigger', -1);}],
        ['non-JSON literal', value => {value.nodes[0].config.args.x = literal(NaN);}],
        ['goto without retry target', value => {value.nodes[1] = {id: 'last', type: 'goto', config: {max_retries: 2, delay_ms: 0}};}],
        ['all merge with one input', value => {value.nodes[1] = merge('last');}],
        ['unsafe initial var', value => {value.initial_vars = JSON.parse('{"__proto__":1}');}],
    ];
    for (const [name, mutate] of cases) await t.test(name, () => {
        const value = valid(); mutate(value);
        assert.throws(() => compile_workflow(value));
    });
    assert.doesNotThrow(() => compile_workflow(valid()));
});