nextcloud-nodered-agent/nodes/agent-process.js

'use strict';

const { spawn } = require('child_process');

module.exports = function (RED) {
    const agentSessions = new Map();

    function AgentProcessNode(config) {
        RED.nodes.createNode(this, config);
        const node = this;
        this.configNode = RED.nodes.getNode(config.agentConfig);
        this.processId = config.processId || 'default';
        this.action = config.action || 'start';
        this.initialInput = config.initialInput || '';
        this.workingDir = config.workingDir || '';
        this.autoRestart = config.autoRestart !== false;

        if (!this.configNode) {
            node.error('No agent config node selected');
            return;
        }

        node.on('input', async function (msg) {
            const cfg = node.configNode;
            const action = msg.action || node.action;
            const processId = msg.processId || node.processId;
            const workspaceDir = msg.workingDir || node.workingDir || cfg.workspaceDir || '/data/agent-workspace';

            try {
                switch (action) {
                    case 'start':
                        await startProcess(node, cfg, processId, workspaceDir, msg);
                        break;
                    case 'stop':
                        await stopProcess(node, processId);
                        break;
                    case 'restart':
                        await stopProcess(node, processId);
                        await startProcess(node, cfg, processId, workspaceDir, msg);
                        break;
                    case 'send':
                        await sendToProcess(node, processId, msg.payload || msg.initialInput);
                        break;
                    case 'status':
                        await getProcessStatus(node, processId);
                        break;
                    default:
                        throw new Error(`Unknown action: ${action}`);
                }
            } catch (err) {
                node.error(`Process ${action} failed: ${err.message}`, msg);
            }
        });

        node.on('close', function (removed, done) {
            if (removed) {
                stopProcess(node, node.processId).finally(done);
            } else {
                done();
            }
        });
    }

    async function startProcess(node, cfg, processId, workspaceDir, msg) {
        if (agentSessions.has(processId)) {
            node.warn(`Process ${processId} already running`);
            msg.payload = { status: 'running', processId };
            node.send(msg);
            return;
        }

        const cliCmd = cfg.cliPath || 'agent';
        const initialInput = msg.initialInput || msg.payload || node.initialInput;
        const defaultArgs = cfg.defaultArgs ? cfg.defaultArgs.split(' ').filter(Boolean) : [];

        const args = ['run', ...defaultArgs];
        if (initialInput) args.push('--input', initialInput);

        node.status({ fill: 'yellow', shape: 'dot', text: `Starting ${processId}...` });

        const child = spawn(cliCmd, args, {
            cwd: workspaceDir,
            env: {
                ...process.env,
                AGENT_WORKSPACE: workspaceDir,
                AGENT_API_ENDPOINT: cfg.apiEndpoint,
                AGENT_API_KEY: cfg.apiKey || ''
            },
            windowsHide: true
        });

        let stdout = '';
        let stderr = '';

        child.stdout.on('data', (data) => { stdout += data.toString(); });
        child.stderr.on('data', (data) => { stderr += data.toString(); });

        const session = {
            id: processId,
            process: child,
            workspaceDir,
            startTime: Date.now(),
            stdout,
            stderr
        };

        agentSessions.set(processId, session);

        child.on('close', (code) => {
            agentSessions.delete(processId);
            node.status({ fill: 'red', shape: 'ring', text: `Stopped (${code})` });
            node.send({
                payload: { status: 'stopped', processId, exitCode: code, stdout, stderr },
                processId
            });
        });

        child.on('error', (err) => {
            agentSessions.delete(processId);
            node.status({ fill: 'red', shape: 'ring', text: 'Error' });
            node.error(`Process ${processId} error: ${err.message}`);
        });

        await new Promise(r => setTimeout(r, 2000));

        node.status({ fill: 'green', shape: 'dot', text: `Running: ${processId}` });
        msg.payload = { status: 'started', processId, pid: child.pid };
        msg.processId = processId;
        node.send(msg);
    }

    async function stopProcess(node, processId) {
        const session = agentSessions.get(processId);
        if (!session) {
            node.warn(`Process ${processId} not found`);
            return;
        }

        session.process.kill('SIGTERM');
        await new Promise(r => setTimeout(r, 1000));
        if (!session.process.killed) {
            session.process.kill('SIGKILL');
        }
    }

    async function sendToProcess(node, processId, input) {
        const session = agentSessions.get(processId);
        if (!session) {
            throw new Error(`Process ${processId} not running`);
        }

        if (session.process.stdin.writable) {
            session.process.stdin.write(input + '\n');
        }

        node.status({ fill: 'blue', shape: 'dot', text: `Sent to ${processId}` });
        node.send({ payload: { sent: true, processId, input }, processId });
    }

    async function getProcessStatus(node, processId) {
        const session = agentSessions.get(processId);
        if (!session) {
            node.send({ payload: { status: 'not running', processId }, processId });
            return;
        }

        node.send({
            payload: {
                status: 'running',
                processId,
                pid: session.process.pid,
                uptime: Date.now() - session.startTime,
                workspaceDir: session.workspaceDir
            },
            processId
        });
    }

    RED.nodes.registerType('agent-process', AgentProcessNode);
};