import {randomUUID} from 'node:crypto';
import {isDeepStrictEqual} from 'node:util';
import type {Bot} from '@wp2508/mineflayer';
import type {bot_event_handler, bot_event_name, bot_instance_context} from '../../instance/bot_instance_init.ts';
import type {Json, workflow_config} from '../../../types/bot/workflow/workflow.ts';
import type {CompiledWorkflow, RunSnapshot, StartResult, StopReason, WorkflowDependencies,
    WorkflowManager as WorkflowManagerContract, WorkflowRunEvent, WorkflowControl, WorkflowCommand,
    WorkflowRuntimeState, WorkflowRuntimeSnapshot} from '../../../types/bot/workflow/runtime.ts';
import {compile_workflow} from './compile.ts';
import {create_workflow_run, type WorkflowRun} from './machine.ts';
import {assert_json} from './value.ts';
import {query_registry} from '../registry.ts';
import {resolve_fields} from './template.ts';
import {register_event_start} from '../start_node/event_node.ts';
import {schedule_timer_start} from '../start_node/timer_node.ts';

const max_queue = 256;
const max_trigger_bytes = 65_536;
const max_history_bytes = 5 * 1024 * 1024;
const max_history_runs = 20;
type Request = {graph: CompiledWorkflow; generation: number; trigger: Record<string, Json>; request_id: string};
type Entry = {graph: CompiledWorkflow; state: WorkflowRuntimeState; cleanup?: () => void;
    firing: boolean; common_fired: boolean; manually_stopped: boolean; stopping: boolean};
export interface WorkflowManagerOptions {
    allow_code?: boolean;
    /** Total concurrent runs for this bot, including runs draining cancellation; defaults to 16. */
    max_concurrent_runs?: number;
    /** Initial persisted configuration revision; zero is the unversioned baseline. */
    source_config_revision?: number;
}

function fail(code: string, message: string): never { throw Object.assign(new Error(message), {code}); }

export class WorkflowManager implements WorkflowManagerContract {
    private readonly graphs = new Map<string, CompiledWorkflow>();
    private readonly entries = new Map<string, Entry>();
    private readonly generations = new Map<string, number>();
    private readonly commands = new Map<string, {request: WorkflowControl; command: WorkflowCommand}>();
    private readonly reload_times = new Map<string, number[]>();
    private operation: Promise<void> = Promise.resolve();
    private pending_controls = 0;
    private source_revision: number;
    private source_definitions: workflow_config[];
    private runtime_pending = false;
    private readonly queue: Request[] = [];
    private readonly history = new Map<string, {snapshot: RunSnapshot; bytes: number}>();
    private readonly listeners = new Set<(event: WorkflowRunEvent) => void>();
    private readonly cleanups: (() => void)[] = [];
    private history_bytes = 0;
    private readonly active = new Map<string, {run: WorkflowRun; workflow_id: string}>();
    private readonly max_concurrent_runs: number;
    private spawned_bot: Bot | null = null;
    private connection_bot: Bot | null = null;
    private launching = false;
    private launch_done?: Promise<void>;
    private initialized = false;
    private disposed = false;
    private dispose_task?: Promise<void>;

    constructor(private readonly context: bot_instance_context<any>, configs: workflow_config[],
        private readonly options: WorkflowManagerOptions = {}) {
        this.max_concurrent_runs = options.max_concurrent_runs ?? 16;
        if (!Number.isSafeInteger(this.max_concurrent_runs) || this.max_concurrent_runs < 1 || this.max_concurrent_runs > 256) {
            fail('INVALID_CONCURRENCY_LIMIT', 'Concurrent run limit must be an integer between 1 and 256');
        }
        this.source_revision = options.source_config_revision ?? 0;
        this.check_revision(this.source_revision);
        const compiled = this.compile(configs);
        this.source_definitions = structuredClone(configs);
        for (const graph of compiled.values()) this.install(graph);
    }

    private compile(configs: workflow_config[]): Map<string, CompiledWorkflow> {
        if (!Array.isArray(configs)) fail('INVALID_WORKFLOW_CONFIG', 'Workflow definitions must be an array');
        const compiled = new Map<string, CompiledWorkflow>();
        for (const config of configs) {
            const graph = compile_workflow(config);
            if (compiled.has(graph.definition.id)) fail('DUPLICATE_WORKFLOW_ID', 'Duplicate workflow id');
            if (graph.definition.start.type === 'event' && graph.definition.start.event === 'workflow_update') {
                fail('INVALID_WORKFLOW_EVENT', 'workflow_update cannot trigger a workflow');
            }
            compiled.set(graph.definition.id, graph);
        }
        return compiled;
    }

    private install(graph: CompiledWorkflow): Entry {
        const id = graph.definition.id;
        const generation = (this.generations.get(id) ?? 0) + 1;
        this.generations.set(id, generation);
        const start = graph.definition.start;
        const entry: Entry = {graph, firing: false, common_fired: false, manually_stopped: false, stopping: false, state: {
            workflow_id: id, generation, enabled: graph.definition.enabled, start_type: start.type,
            trigger_state: 'stopped', accepted_count: 0,
            max_runs: start.type === 'common' || start.mode === 'once' ? 1
                : start.type === 'timer' ? start.max_runs ?? null : null,
            next_fire_at: null, active_run_ids: [], queued_count: 0, last_error: null,
        }};
        this.graphs.set(id, graph);
        this.entries.set(id, entry);
        return entry;
    }

    initialize(): void {
        if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
        if (this.initialized) return;
        this.initialized = true;
        this.spawned_bot = this.context.bot_status === 'online' && this.context.bot_instance?.entity
            ? this.context.bot_instance : null;
        this.connection_bot = this.spawned_bot;
        try {
            this.listen('spawn', () => {
                const bot = this.context.bot_instance;
                if (!bot) return;
                if (this.spawned_bot && this.spawned_bot !== bot) this.disconnect();
                this.spawned_bot = bot;
                if (this.connection_bot !== bot) {
                    this.connection_bot = bot;
                    for (const entry of this.entries.values()) {
                        if (entry.stopping) continue;
                        this.stop_trigger(entry);
                        entry.manually_stopped = false;
                        entry.common_fired = false;
                        entry.state.accepted_count = 0;
                        entry.state.last_error = null;
                        this.auto_start(entry);
                    }
                }
                this.runtime_changed();
            });
            this.listen('status_update', status => {
                if (status !== 'online') this.disconnect();
            });
            this.listen('end', () => this.disconnect());
            this.listen('kicked', () => this.disconnect());
            this.listen('error', () => this.disconnect());
            this.listen('connection_lost', () => this.disconnect());
            this.listen('instance_created', () => this.disconnect());
            for (const entry of this.entries.values()) this.auto_start(entry, true);
        } catch (error) {
            void this.dispose().catch(() => {});
            throw error;
        }
    }

    private listen<E extends bot_event_name>(event: E, callback: bot_event_handler<E>): void {
        this.cleanups.push(() => this.context.off(event, callback));
        this.context.on(event, callback);
    }

    private ready(): boolean {
        return this.context.bot_status === 'online' && this.spawned_bot !== null
            && this.context.bot_instance === this.spawned_bot && !!this.spawned_bot.entity;
    }

    private disconnect(): void {
        this.spawned_bot = null;
        this.queue.length = 0;
        for (const entry of this.entries.values()) this.stop_trigger(entry);
        for (const {run} of this.active.values()) void run.stop('disconnect').catch(() => {});
        this.runtime_changed();
    }

    private auto_start(entry: Entry, defer_common = false): void {
        if (!this.ready() || this.disposed || entry.stopping || !entry.graph.definition.enabled || entry.manually_stopped) return;
        const config = entry.graph.definition.start;
        if (config.type === 'common') {
            if (config.auto_start !== 'spawn' || entry.common_fired) return;
            const fire = () => {
                if (!this.ready() || this.disposed || this.entries.get(entry.state.workflow_id) !== entry
                    || entry.stopping || entry.common_fired || entry.manually_stopped) return;
                entry.common_fired = true;
                this.fire(entry, {event: 'spawn'});
            };
            if (defer_common) queueMicrotask(fire); else fire();
        } else if (config.auto_start !== false) this.start_trigger(entry);
    }

    private stop_trigger(entry: Entry, state: WorkflowRuntimeState['trigger_state'] = 'stopped'): void {
        const cleanup = entry.cleanup;
        entry.cleanup = undefined;
        entry.state.trigger_state = state;
        entry.state.next_fire_at = null;
        cleanup?.();
        this.runtime_changed();
    }

    private start_trigger(entry: Entry): void {
        if (entry.stopping) fail('WORKFLOW_STOPPING', 'Workflow is stopping');
        if (entry.cleanup || entry.firing) return;
        if (!entry.graph.definition.enabled) fail('WORKFLOW_DISABLED', 'Workflow is disabled');
        if (!this.ready()) fail('BOT_NOT_READY', 'Bot must be online and spawned');
        entry.manually_stopped = false;
        entry.state.last_error = null;
        const config = entry.graph.definition.start;
        if (config.type === 'event') {
            entry.cleanup = register_event_start(this.context, entry.graph.definition, trigger => this.fire(entry, trigger));
            entry.state.trigger_state = 'listening';
        } else if (config.type === 'timer') {
            entry.state.next_fire_at = Date.now() + config.interval_ms;
            entry.cleanup = schedule_timer_start(config.interval_ms, trigger => {
                entry.cleanup = undefined;
                entry.state.next_fire_at = null;
                this.fire(entry, trigger);
                if (entry.state.trigger_state === 'scheduled' && !entry.cleanup && this.ready() && !this.disposed) {
                    this.start_trigger(entry);
                }
            });
            entry.state.trigger_state = 'scheduled';
        }
        this.runtime_changed();
    }

    private fire(entry: Entry, trigger: Record<string, Json>): void {
        if (entry.firing || this.entries.get(entry.state.workflow_id) !== entry || this.disposed || !this.ready()
            || entry.stopping || entry.manually_stopped || entry.state.trigger_state === 'reloading') return;
        if (entry.state.start_type !== 'common'
            && !['listening', 'scheduled'].includes(entry.state.trigger_state)) return;
        entry.firing = true;
        const previous = entry.state.trigger_state;
        const last = entry.state.max_runs !== null && entry.state.accepted_count + 1 >= entry.state.max_runs;
        // Detach finite triggers before start() publishes any reentrant run notifications.
        if (last) this.stop_trigger(entry, 'exhausted');
        try {
            const result = this.start(entry.state.workflow_id, trigger);
            if (result.status !== 'skipped') entry.state.accepted_count++;
            else if (last) entry.state.trigger_state = previous;
        } catch (error) {
            this.stop_trigger(entry, 'error');
            entry.state.last_error = this.error(error);
        } finally {
            entry.firing = false;
            if (last && entry.state.trigger_state === 'listening' && !entry.cleanup && this.ready()) this.start_trigger(entry);
            this.runtime_changed();
        }
    }

    private available(graph: CompiledWorkflow): boolean {
        const entry = this.entries.get(graph.definition.id);
        return entry?.graph === graph && !entry.stopping && this.active.size < this.max_concurrent_runs
            && [...this.active.values()].filter(item => item.workflow_id === graph.definition.id).length
                < (graph.definition.policy.max_concurrent_runs ?? 1);
    }

    start(workflow_id: string, trigger: Record<string, Json> = {}): StartResult {
        if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
        const graph = this.graphs.get(workflow_id);
        if (!graph) fail('WORKFLOW_NOT_FOUND', `Unknown workflow: ${workflow_id}`);
        const entry = this.entries.get(workflow_id)!;
        if (entry.state.trigger_state === 'reloading') fail('WORKFLOW_RELOADING', 'Workflow is being replaced');
        if (entry.stopping) fail('WORKFLOW_STOPPING', 'Workflow is stopping');
        if (!graph.definition.enabled) fail('WORKFLOW_DISABLED', `Workflow is disabled: ${workflow_id}`);
        if (!this.initialized || !this.ready()) fail('BOT_NOT_READY', 'Bot must be online and spawned');
        assert_json(trigger);
        if (trigger === null || typeof trigger !== 'object' || Array.isArray(trigger)
            || Buffer.byteLength(JSON.stringify(trigger)) > max_trigger_bytes) {
            fail('INVALID_WORKFLOW_TRIGGER', 'Workflow trigger must be a bounded JSON object');
        }
        const copy = structuredClone(trigger);
        if (!this.available(graph) || this.launching) {
            if (graph.definition.policy.on_busy === 'skip' || this.queue.length >= max_queue
                || this.queue.filter(item => item.graph === graph).length >= graph.definition.policy.max_queue) {
                return {status: 'skipped'};
            }
            const request_id = randomUUID();
            this.queue.push({graph, generation: entry.state.generation, trigger: copy, request_id});
            this.runtime_changed();
            queueMicrotask(() => this.pump());
            return {status: 'queued', request_id};
        }
        return {status: 'started', run_id: this.launch(graph, copy)};
    }

    private launch(graph: CompiledWorkflow, trigger: Record<string, Json>): string {
        this.launching = true;
        let finish_launch!: () => void;
        this.launch_done = new Promise(resolve => { finish_launch = resolve; });
        const notifications: WorkflowRunEvent[] = [];
        let assigned = false;
        try {
            const run = create_workflow_run(graph, this.dependencies(), trigger, event => {
                if (!assigned) notifications.push(event);
                else this.publish(event);
            });
            this.active.set(run.run_id, {run, workflow_id: graph.definition.id});
            this.runtime_changed();
            assigned = true;
            void run.done.then(snapshot => {
                this.remember(snapshot);
                this.active.delete(run.run_id);
                this.runtime_changed();
                this.pump();
            }).catch(() => {
                this.active.delete(run.run_id);
                this.runtime_changed();
                this.pump();
            });
            if (this.disposed || !this.ready()) {
                void run.stop(this.disposed ? 'disposed' : 'disconnect').catch(() => {});
            }
            for (const event of notifications) this.publish(event);
            return run.run_id;
        } finally {
            this.launching = false;
            finish_launch();
        }
    }

    private pump(): void {
        if (this.disposed || this.launching || !this.ready()) return;
        for (let index = this.queue.length - 1; index >= 0; index--) {
            const request = this.queue[index];
            const entry = this.entries.get(request.graph.definition.id);
            if (!entry || entry.graph !== request.graph || entry.state.generation !== request.generation
                || entry.stopping || entry.state.trigger_state === 'reloading') this.queue.splice(index, 1);
        }
        while (this.active.size < this.max_concurrent_runs && this.ready() && !this.disposed) {
            // A busy definition must not block other workflows with free capacity.
            const index = this.queue.findIndex(request => this.available(request.graph));
            if (index < 0) break;
            const request = this.queue.splice(index, 1)[0];
            this.runtime_changed();
            try { this.launch(request.graph, request.trigger); }
            catch { /* A failed launch must not strand later requests. */ }
        }
    }

    private dependencies(): WorkflowDependencies {
        const bot = this.context.bot_instance!;
        const functions = this.context.function_manager;
        const info = this.context.info_manager;
        const check = (signal: AbortSignal) => {
            signal.throwIfAborted();
            if (this.disposed || !this.ready() || this.context.bot_instance !== bot
                || this.context.function_manager !== functions || this.context.info_manager !== info) {
                fail('WORKFLOW_CONTEXT_CHANGED', 'Bot runtime changed during workflow execution');
            }
        };
        return {
            ...(!this.options.allow_code ? {execute_code: async () =>
                fail('CODE_EXECUTION_DISABLED', 'Workflow code execution is disabled')} : {}),
            run_function: async (path, name, args, signal) => {
                check(signal);
                if (!functions) fail('FUNCTION_MANAGER_UNAVAILABLE', 'Bot function manager is unavailable');
                return functions.run(path, name, args, () => check(signal), {signal});
            },
            query: async (source, args, signal) => {
                check(signal);
                if (!Object.hasOwn(query_registry, source)) fail('UNKNOWN_QUERY_SOURCE', `Unknown workflow query source: ${source}`);
                const provider = query_registry[source as keyof typeof query_registry];
                let config: Record<string, Json>;
                try {config = resolve_fields(provider.config_template.config, args);}
                catch {fail('INVALID_QUERY_ARGUMENTS', 'Invalid workflow query arguments');}
                return provider.query_info({info_manager: info}, config);
            },
            sleep: (mode, value, signal) => {
                check(signal);
                return new Promise<void>((resolve, reject) => {
                    let timer: ReturnType<typeof setTimeout> | undefined;
                    let ticks = 0;
                    let settled = false;
                    const cleanup = () => {
                        clearTimeout(timer);
                        signal.removeEventListener('abort', abort);
                        bot.off('physicsTick', tick);
                        bot.off('end', end);
                    };
                    const finish = (error?: unknown) => {
                        if (settled) return;
                        settled = true;
                        cleanup();
                        if (error !== undefined) reject(error); else resolve();
                    };
                    const abort = () => finish(signal.reason ?? new Error('Workflow sleep cancelled'));
                    const end = () => finish(new Error('Bot disconnected during workflow sleep'));
                    const tick = () => {
                        try { check(signal); if (++ticks >= value) finish(); }
                        catch (error) { finish(error); }
                    };
                    try {
                        signal.addEventListener('abort', abort, {once: true});
                        if (signal.aborted) { abort(); return; }
                        if (value === 0) { finish(); return; }
                        bot.on('end', end);
                        if (mode === 'milliseconds') timer = setTimeout(() => {
                            try { check(signal); finish(); } catch (error) { finish(error); }
                        }, value);
                        else bot.on('physicsTick', tick);
                    } catch (error) { finish(error); }
                });
            },
            log: (level, message) => {
                const logger = this.context.logger;
                if (logger) this.context.log([logger[level](message)]);
            },
        };
    }

    private remember(snapshot: RunSnapshot): void {
        const previous = this.history.get(snapshot.run_id);
        if (previous) { this.history.delete(snapshot.run_id); this.history_bytes -= previous.bytes; }
        let bytes = Buffer.byteLength(JSON.stringify(snapshot));
        if (bytes > max_history_bytes) {
            snapshot = {...snapshot, branches: []};
            bytes = Buffer.byteLength(JSON.stringify(snapshot));
        }
        this.history.set(snapshot.run_id, {snapshot: structuredClone(snapshot), bytes});
        this.history_bytes += bytes;
        while (this.history.size > max_history_runs || this.history_bytes > max_history_bytes) {
            const oldest = this.history.entries().next().value!;
            this.history.delete(oldest[0]);
            this.history_bytes -= oldest[1].bytes;
        }
    }

    private publish(event: WorkflowRunEvent): void {
        if (event.type === 'run_updated' && event.snapshot.ended_at !== null) this.remember(event.snapshot);
        try { this.context.emit('workflow_update', structuredClone(event)); } catch {}
        for (const listener of Array.from(this.listeners)) {
            if (!this.listeners.has(listener)) continue;
            try { listener(structuredClone(event)); } catch {}
        }
    }

    async stop(run_id: string, reason: StopReason = 'manual'): Promise<void> {
        const active = this.active.get(run_id);
        if (active) await active.run.stop(reason);
        else if (!this.history.has(run_id)) fail('WORKFLOW_RUN_NOT_FOUND', `Unknown workflow run: ${run_id}`);
    }

    get_snapshot(run_id: string): RunSnapshot | undefined {
        const active = this.active.get(run_id);
        if (active) return active.run.get_snapshot();
        const snapshot = this.history.get(run_id)?.snapshot;
        return snapshot ? structuredClone(snapshot) : undefined;
    }

    get_snapshots(): RunSnapshot[] {
        const snapshots = Array.from(this.history.values(), entry => structuredClone(entry.snapshot));
        for (const {run} of this.active.values()) if (!this.history.has(run.run_id)) snapshots.push(run.get_snapshot());
        return snapshots;
    }

    on_update(callback: (event: WorkflowRunEvent) => void): () => void {
        if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
        const listener = (event: WorkflowRunEvent) => callback(event);
        this.listeners.add(listener);
        return () => { this.listeners.delete(listener); };
    }

    private error(error: unknown): {code: string; message: string} {
        return {code: typeof (error as any)?.code === 'string' ? (error as any).code : 'WORKFLOW_OPERATION_FAILED',
            message: error instanceof Error ? error.message : 'Workflow operation failed'};
    }

    private check_revision(revision: number): void {
        if (!Number.isSafeInteger(revision) || revision < 0) fail('INVALID_SOURCE_REVISION', 'Invalid workflow source revision');
    }

    private serialize<T>(operation: () => Promise<T> | T): Promise<T> {
        const task = this.operation.then(() => {
            if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
            return operation();
        });
        this.operation = task.then(() => {}, () => {});
        return task;
    }

    apply(configs: workflow_config[], source_revision?: number): Promise<void> {
        // Compilation and cloning finish before any controller or running graph is touched.
        let compiled: Map<string, CompiledWorkflow>;
        let definitions: workflow_config[];
        try {
            compiled = this.compile(configs);
            definitions = structuredClone(configs);
            if (source_revision !== undefined) this.check_revision(source_revision);
        } catch (error) { return Promise.reject(error); }
        return this.serialize(async () => {
            if (source_revision !== undefined) {
                if (source_revision < this.source_revision) fail('STALE_SOURCE_REVISION', 'Workflow source revision is stale');
                if (source_revision === this.source_revision) {
                    if (!isDeepStrictEqual(definitions, this.source_definitions)) {
                        fail('SOURCE_REVISION_CONFLICT', 'Workflow source revision has different content');
                    }
                    return;
                }
            }
            const replaced = [...this.entries.values()].filter(entry =>
                !isDeepStrictEqual(entry.graph.definition, compiled.get(entry.state.workflow_id)?.definition));
            // Connection events may overwrite trigger_state while cancellation is draining.
            for (const entry of replaced) entry.stopping = true;
            const ids = new Set(replaced.map(entry => entry.state.workflow_id));
            try {
                for (const entry of replaced) this.stop_trigger(entry, 'reloading');
                this.remove_queue(ids);
                await Promise.all(replaced.map(entry => this.drain(entry.state.workflow_id, 'updated')));
            } catch (error) {
                for (const entry of replaced) {
                    entry.manually_stopped = true;
                    entry.state.last_error = this.error(error);
                    this.stop_trigger(entry, 'error');
                }
                throw error;
            }
            if (this.disposed) return;
            for (const id of ids) { this.entries.delete(id); this.graphs.delete(id); }
            for (const [id, graph] of compiled) {
                if (!this.entries.has(id)) this.auto_start(this.install(graph));
            }
            this.source_definitions = definitions;
            if (source_revision !== undefined) this.source_revision = source_revision;
            this.runtime_changed();
            this.pump();
        });
    }

    private remove_queue(ids: Set<string>): void {
        for (let index = this.queue.length - 1; index >= 0; index--) {
            if (ids.has(this.queue[index].graph.definition.id)) this.queue.splice(index, 1);
        }
        this.runtime_changed();
    }

    private async drain(workflow_id: string, reason: StopReason): Promise<void> {
        await this.launch_done;
        await Promise.all([...this.active.values()].filter(active => active.workflow_id === workflow_id)
            .map(({run}) => run.stop(reason)));
    }

    control(request: WorkflowControl): WorkflowCommand {
        if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
        assert_json(request);
        if (!request || typeof request !== 'object' || Array.isArray(request)
            || typeof request.command_id !== 'string' || !/^[a-zA-Z0-9_-]{1,100}$/.test(request.command_id)
            || !['run_once', 'trigger_start', 'trigger_stop', 'stop', 'stop_run', 'reload'].includes(request.action)
            || Object.keys(request).some(key => !['command_id', 'action', 'workflow_id', 'run_id', 'expected_generation', 'input'].includes(key))
            || request.expected_generation !== undefined && (!Number.isSafeInteger(request.expected_generation) || request.expected_generation < 1)
            || request.workflow_id !== undefined && (typeof request.workflow_id !== 'string' || request.workflow_id.length > 100)
            || request.run_id !== undefined && (typeof request.run_id !== 'string' || request.run_id.length > 100)) {
            fail('INVALID_WORKFLOW_CONTROL', 'Invalid workflow control request');
        }
        if (request.input !== undefined && (request.input === null || typeof request.input !== 'object'
            || Array.isArray(request.input) || Buffer.byteLength(JSON.stringify(request.input)) > max_trigger_bytes)) {
            fail('INVALID_WORKFLOW_TRIGGER', 'Workflow input must be a bounded JSON object');
        }
        const previous = this.commands.get(request.command_id);
        if (previous) {
            if (!isDeepStrictEqual(previous.request, request)) fail('COMMAND_ID_CONFLICT', 'Command id has different content');
            return structuredClone(previous.command);
        }
        if (this.pending_controls >= 32) fail('WORKFLOW_CONTROL_BUSY', 'Too many pending workflow controls');
        const copy = structuredClone(request);
        const command: WorkflowCommand = {command_id: request.command_id, action: request.action,
            workflow_id: request.workflow_id ?? null, state: 'accepted'};
        this.commands.set(command.command_id, {request: copy, command});
        this.pending_controls++;
        this.runtime_changed();
        void this.serialize(async () => {
            if (copy.action === 'stop_run') {
                if (!copy.run_id) fail('INVALID_WORKFLOW_CONTROL', 'stop_run requires run_id');
                const snapshot = this.get_snapshot(copy.run_id);
                if (!snapshot) fail('WORKFLOW_RUN_NOT_FOUND', 'Unknown workflow run');
                if (copy.workflow_id !== undefined && copy.workflow_id !== snapshot.workflow_id) {
                    fail('WORKFLOW_RUN_MISMATCH', 'Run belongs to another workflow');
                }
                command.workflow_id = snapshot.workflow_id;
                this.expect_generation(copy, this.entries.get(snapshot.workflow_id));
                await this.stop(copy.run_id);
                command.result = {stopped: true};
                return;
            }
            const entry = copy.workflow_id ? this.entries.get(copy.workflow_id) : undefined;
            if (!entry) fail('WORKFLOW_NOT_FOUND', 'Unknown workflow');
            this.expect_generation(copy, entry);
            switch (copy.action) {
                case 'run_once': command.result = this.start(entry.state.workflow_id, copy.input ?? {}); break;
                case 'trigger_start':
                    if (entry.state.start_type === 'common') fail('INVALID_TRIGGER_TYPE', 'Common workflows have no recurring trigger');
                    if (!entry.cleanup && !entry.firing) {
                        if (!this.ready()) fail('BOT_NOT_READY', 'Bot must be online and spawned');
                        entry.state.accepted_count = 0;
                        this.start_trigger(entry);
                    }
                    command.result = {generation: entry.state.generation};
                    break;
                case 'trigger_stop':
                    if (entry.state.start_type === 'common') fail('INVALID_TRIGGER_TYPE', 'Common workflows have no recurring trigger');
                    entry.manually_stopped = true;
                    this.stop_trigger(entry);
                    command.result = {stopped: true};
                    break;
                case 'stop':
                    entry.manually_stopped = true;
                    entry.stopping = true;
                    this.stop_trigger(entry);
                    this.remove_queue(new Set([entry.state.workflow_id]));
                    try { await this.drain(entry.state.workflow_id, 'manual'); }
                    finally { entry.stopping = false; }
                    command.result = {stopped: true};
                    break;
                case 'reload': {
                    const now = Date.now();
                    const recent = (this.reload_times.get(entry.state.workflow_id) ?? []).filter(time => now - time < 10_000);
                    if (recent.length >= 10) fail('WORKFLOW_RELOAD_RATE_LIMIT', 'Workflow reload limit exceeded');
                    recent.push(now);
                    this.reload_times.set(entry.state.workflow_id, recent);
                    const graph = this.compile([entry.graph.definition]).get(entry.state.workflow_id)!;
                    entry.stopping = true;
                    try {
                        this.stop_trigger(entry, 'reloading');
                        this.remove_queue(new Set([entry.state.workflow_id]));
                        await this.drain(entry.state.workflow_id, 'reloaded');
                    } catch (error) {
                        entry.manually_stopped = true;
                        entry.state.last_error = this.error(error);
                        this.stop_trigger(entry, 'error');
                        throw error;
                    }
                    if (this.disposed) fail('WORKFLOW_MANAGER_DISPOSED', 'Workflow manager is disposed');
                    const replacement = this.install(graph);
                    this.auto_start(replacement);
                    command.result = {generation: replacement.state.generation};
                    break;
                }
            }
        }).then(() => { command.state = 'completed'; }, error => {
            command.state = 'failed'; command.error = this.error(error);
        }).finally(() => {
            this.pending_controls--;
            while (this.commands.size > 100) {
                const oldest = [...this.commands].find(([, item]) => item.command.state !== 'accepted');
                if (!oldest) break;
                this.commands.delete(oldest[0]);
            }
            this.runtime_changed();
        });
        return structuredClone(command);
    }

    private expect_generation(request: WorkflowControl, entry?: Entry): void {
        if (request.expected_generation !== undefined && entry?.state.generation !== request.expected_generation) {
            fail('WORKFLOW_GENERATION_CONFLICT', 'Workflow generation has changed');
        }
    }

    get_command(command_id: string): WorkflowCommand | undefined {
        const command = this.commands.get(command_id)?.command;
        return command ? structuredClone(command) : undefined;
    }

    get_runtime_snapshot(): WorkflowRuntimeSnapshot {
        return {source_config_revision: this.source_revision,
            workflows: [...this.entries.values()].map(entry => ({...structuredClone(entry.state),
                active_run_ids: [...this.active].filter(([, active]) => active.workflow_id === entry.state.workflow_id)
                    .map(([id]) => id),
                queued_count: this.queue.filter(request => request.graph === entry.graph
                    && request.generation === entry.state.generation).length})),
            commands: [...this.commands.values()].map(item => structuredClone(item.command))};
    }

    private runtime_changed(): void {
        if (!this.initialized || this.runtime_pending || this.disposed) return;
        this.runtime_pending = true;
        queueMicrotask(() => {
            this.runtime_pending = false;
            if (!this.disposed) this.publish({type: 'runtime_updated', snapshot: this.get_runtime_snapshot()});
        });
    }

    dispose(): Promise<void> {
        if (this.dispose_task) return this.dispose_task;
        this.disposed = true;
        this.spawned_bot = null;
        this.queue.length = 0;
        for (const entry of this.entries.values()) this.stop_trigger(entry);
        for (const cleanup of this.cleanups.splice(0).reverse()) {
            try { cleanup(); } catch {}
        }
        const launching = this.launch_done;
        this.dispose_task = Promise.resolve().then(async () => {
            try { await launching; await Promise.all([...this.active.values()].map(({run}) => run.stop('disposed'))); }
            finally { await this.operation; this.listeners.clear(); }
        });
        return this.dispose_task;
    }
}

export function create_workflow_manager(context: bot_instance_context<any>, configs: workflow_config[],
    options: WorkflowManagerOptions = {}): WorkflowManager {
    const manager = new WorkflowManager(context, configs, options);
    manager.initialize();
    return manager;
}