

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE version 2.1.1.260726 (26 July 2026)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2026
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  PSYCLONE 2.1 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.1 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.1 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.1.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.1 is distributed in the hope that it will be useful,
//_/_/  but WITHOUT ANY WARRANTY; without even the implied warranty of
//_/_/  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//_/_/  GNU General Public License for more details.
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
// T1.4: Auto-start and lifecycle management of external-module executables.
// See Docs/T1.4-external-autostart-spec.md for the full specification.
//
// POSIX:   full implementation (fork/execvp with pipes, env, workdir, signals).
// WINDOWS: full implementation via utils::NewProcessEx / ReadProcessOutput /
//          SendProcessBreak (CreateProcess with env block, working directory,
//          stdout/stderr capture pipes, CTRL_BREAK graceful stop, and the
//          global Job Object for kill-on-node-exit). See those helpers in Utils.cpp.

#include "ExternalExec.h"
#include "Node.h"

#if !defined(WINDOWS)
	#include <unistd.h>
	#include <fcntl.h>
	#include <signal.h>
	#include <sys/wait.h>
	#include <errno.h>
	#ifndef __APPLE__
		#include <sys/prctl.h>
	#else
		#include <sys/event.h>
		#include <sys/types.h>
	#endif
#endif

namespace	cmlabs{

#define EXTEXEC_POLL_MS			100		/* monitor thread poll interval */
#define EXTEXEC_HEARTBEAT_MS	10000	/* space heartbeat staleness => hung */
#define EXTEXEC_HEALTHY_MS		30000	/* uptime after which backoff resets */

THREAD_RET THREAD_FUNCTION_CALL ExternalExecRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((ExternalExecManager*)arg)->run() ? 0 : 1));
}

ExternalExecManager::ExternalExecManager(Node* node) {
	this->node = node;
	threadID = 0;
	threadRunning = false;
	shouldContinue = true;
}

ExternalExecManager::~ExternalExecManager() {
	stopAll();
	std::lock_guard<std::mutex> lock(mutex);
	while (entries.size()) {
		delete(entries.front());
		entries.pop_front();
	}
}

int ExternalExecManager::ParseOutputMode(const char* str) {
	if (!str) return -1;
	if (stricmp(str, "ignore") == 0) return EXTEXEC_OUTPUT_IGNORE;
	if (stricmp(str, "console") == 0) return EXTEXEC_OUTPUT_CONSOLE;
	if (stricmp(str, "logfile") == 0) return EXTEXEC_OUTPUT_LOGFILE;
	if (stricmp(str, "psyclonelog") == 0) return EXTEXEC_OUTPUT_PSYCLONELOG;
	if (stricmp(str, "messages") == 0) return EXTEXEC_OUTPUT_MESSAGES;
	return -1;
}

int ExternalExecManager::ParseSignal(const char* str) {
	if (!str || !strlen(str)) return -1;
	if (isdigit(str[0])) return (int)utils::Ascii2Uint32(str);
	const char* s = str;
	if (strnicmp(s, "SIG", 3) == 0) s += 3;
#if !defined(WINDOWS)
	if (stricmp(s, "TERM") == 0) return SIGTERM;
	if (stricmp(s, "INT") == 0) return SIGINT;
	if (stricmp(s, "KILL") == 0) return SIGKILL;
	if (stricmp(s, "HUP") == 0) return SIGHUP;
	if (stricmp(s, "QUIT") == 0) return SIGQUIT;
	if (stricmp(s, "USR1") == 0) return SIGUSR1;
	if (stricmp(s, "USR2") == 0) return SIGUSR2;
#else
	// Windows: signal numbers are advisory only. Graceful stop is delivered as a
	// console CTRL_BREAK (see SendProcessBreak); the number is preserved for parity.
	if (stricmp(s, "TERM") == 0) return 15;
	if (stricmp(s, "KILL") == 0) return 9;
#endif
	return -1;
}

std::string ExternalExecManager::substitute(Entry* entry, const std::string& text) {
	std::string result = text;
	std::string port = utils::StringFormat("%u", node->networkPort);
	utils::StringSingleReplace(result, "%nodeid%", port.c_str(), false);
	utils::StringSingleReplace(result, "%port%", port.c_str(), false);
	utils::StringSingleReplace(result, "%spacename%", entry->spec.spaceName.c_str(), false);
	utils::StringSingleReplace(result, "%spaceid%", utils::StringFormat("%u", entry->spec.spaceID).c_str(), false);
	utils::StringSingleReplace(result, "%moduleid%", utils::StringFormat("%u", entry->spec.moduleID).c_str(), false);
	utils::StringSingleReplace(result, "%modulename%", entry->spec.moduleName.c_str(), false);
	std::string specdir;
	if (node->configfile.size()) {
		size_t pos = node->configfile.find_last_of("/\\");
		if (pos != std::string::npos)
			specdir = node->configfile.substr(0, pos);
	}
	utils::StringSingleReplace(result, "%specdir%", specdir.c_str(), false);
	// Directory of the running Psyclone executable (with trailing separator).
	// Handy when a helper binary is shipped next to Psyclone (e.g. the
	// psytest ExternalModules helper): <executable>%execdir%ExternalModules ...
	// resolves regardless of the current working directory, whereas a bare
	// name would rely on PATH and fail when launched as ./bin/<plat>/Psyclone.
	utils::StringSingleReplace(result, "%execdir%", utils::GetCommandLinePath().c_str(), false);
	return result;
}

bool ExternalExecManager::registerExecutable(const ExternalExecSpec& spec) {
	std::lock_guard<std::mutex> lock(mutex);
	std::list<Entry*>::iterator i = entries.begin(), e = entries.end();
	while (i != e) {
		if (stricmp((*i)->spec.spaceName.c_str(), spec.spaceName.c_str()) == 0) {
			LogPrint(node->nodeID, LOG_NODE, 0,
				"External space '%s' already has an executable (declared by module '%s'); ignoring duplicate <executable> from module '%s'...",
				spec.spaceName.c_str(), (*i)->spec.moduleName.c_str(), spec.moduleName.c_str());
			return false;
		}
		i++;
	}

	Entry* entry = new Entry();
	entry->spec = spec;
	entry->expandedCmdline = substitute(entry, spec.cmdline);
	entry->expandedWorkdir = substitute(entry, spec.workdir);
	entry->expandedLogfile = substitute(entry, spec.logfile);
	std::map<std::string, std::string>::iterator ei = entry->spec.env.begin();
	while (ei != entry->spec.env.end()) {
		ei->second = substitute(entry, ei->second);
		ei++;
	}
	entries.push_back(entry);

	if (!spec.enabled) {
		entry->state = EXTEXEC_STATE_DISABLED;
		LogPrint(node->nodeID, LOG_NODE, 1,
			"External executable for space '%s' registered but disabled (enabled=\"no\"): %s",
			spec.spaceName.c_str(), entry->expandedCmdline.c_str());
	}
	else if (!launch(entry))
		handleFailure(entry, "could not start");

	if (!threadRunning) {
		threadRunning = true;
		if (!ThreadManager::CreateThread(ExternalExecRun, this, threadID)) {
			threadRunning = false;
			LogPrint(node->nodeID, LOG_NODE, 0, "Could not start ExternalExec monitor thread...");
			return false;
		}
	}
	return true;
}

bool ExternalExecManager::launch(Entry* entry) {
	entry->launchTime = GetTimeNow();
	entry->outBuf.clear();
	entry->errBuf.clear();

#if !defined(WINDOWS)
	bool capture = (entry->spec.outputMode == EXTEXEC_OUTPUT_LOGFILE) ||
				   (entry->spec.outputMode == EXTEXEC_OUTPUT_PSYCLONELOG) ||
				   (entry->spec.outputMode == EXTEXEC_OUTPUT_MESSAGES);

	int outPipe[2] = {-1, -1}, errPipe[2] = {-1, -1};
	if (capture) {
		if ( (pipe(outPipe) != 0) || (pipe(errPipe) != 0) ) {
			LogPrint(node->nodeID, LOG_NODE, 0, "ExternalExec: could not create pipes for space '%s'...", entry->spec.spaceName.c_str());
			if (outPipe[0] >= 0) { close(outPipe[0]); close(outPipe[1]); }
			return false;
		}
	}

	int argc = 0;
	char** argv = utils::SplitCommandline(entry->expandedCmdline.c_str(), argc);
	if (!argc) {
		LogPrint(node->nodeID, LOG_NODE, 0, "ExternalExec: empty command line for space '%s'...", entry->spec.spaceName.c_str());
		if (capture) { close(outPipe[0]); close(outPipe[1]); close(errPipe[0]); close(errPipe[1]); }
		return false;
	}

	pid_t ppid_before_fork = getpid();
	pid_t pid = fork();
	if (pid < 0) {
		utils::DeleteCommandline(argv, argc);
		if (capture) { close(outPipe[0]); close(outPipe[1]); close(errPipe[0]); close(errPipe[1]); }
		return false;
	}
	else if (pid == 0) {
		// Child
		if (capture) {
			dup2(outPipe[1], STDOUT_FILENO);
			dup2(errPipe[1], STDERR_FILENO);
			close(outPipe[0]); close(outPipe[1]);
			close(errPipe[0]); close(errPipe[1]);
		}
		else if (entry->spec.outputMode == EXTEXEC_OUTPUT_IGNORE) {
			int devnull = open("/dev/null", O_WRONLY);
			if (devnull >= 0) {
				dup2(devnull, STDOUT_FILENO);
				dup2(devnull, STDERR_FILENO);
				close(devnull);
			}
		}
		// EXTEXEC_OUTPUT_CONSOLE: inherit stdout/stderr as-is

		if (entry->expandedWorkdir.size() && (chdir(entry->expandedWorkdir.c_str()) != 0)) {
			fprintf(stderr, "ExternalExec: could not chdir to '%s'\n", entry->expandedWorkdir.c_str());
			_exit(127);
		}

		std::map<std::string, std::string>::iterator ei = entry->spec.env.begin();
		while (ei != entry->spec.env.end()) {
			setenv(ei->first.c_str(), ei->second.c_str(), 1);
			ei++;
		}

		#ifndef __APPLE__
		if (entry->spec.shutdownOnDelete) {
			// Die with the node if it crashes hard (normal shutdown is graceful).
			prctl(PR_SET_PDEATHSIG, SIGKILL);
			if (getppid() != ppid_before_fork)
				_exit(1);
		}
		#else
		if (entry->spec.shutdownOnDelete) {
			// macOS has no PR_SET_PDEATHSIG. Put this child in its own process group
			// and spawn a tiny detached "nanny" grandchild that kqueue-watches the
			// node (our parent) for NOTE_EXIT and kills this child's group when the
			// node dies -- covering even a hard node crash (SIGKILL), which the
			// monitor thread cannot (it dies with the node).
			pid_t nodePid = ppid_before_fork;
			setpgid(0, 0);					// child leads its own group (== its pid)
			pid_t childPid = getpid();
			if (getppid() != nodePid)		// node already gone before we got here
				_exit(1);
			pid_t nanny = fork();
			(void)nanny;
			if (nanny == 0) {
				// Grandchild nanny: watch nodePid; on its exit, kill the child group.
				int kq = kqueue();
				if (kq >= 0) {
					struct kevent ke;
					EV_SET(&ke, nodePid, EVFILT_PROC, EV_ADD | EV_ONESHOT, NOTE_EXIT, 0, NULL);
					if (kevent(kq, &ke, 1, NULL, 0, NULL) == 0) {
						struct kevent out;
						kevent(kq, NULL, 0, &out, 1, NULL);	// blocks until node exits
					}
				}
				kill(-childPid, SIGKILL);		// kill the whole child group
				_exit(0);
			}
			// child continues to exec below; nanny is detached (reaped by launchd)
		}
		#endif

		execvp(argv[0], argv);
		fprintf(stderr, "ExternalExec: could not exec '%s' (%d)\n", entry->expandedCmdline.c_str(), errno);
		_exit(127);
	}

	// Parent
	utils::DeleteCommandline(argv, argc);
	entry->pid = (uint32)pid;
	if (capture) {
		close(outPipe[1]);
		close(errPipe[1]);
		fcntl(outPipe[0], F_SETFL, O_NONBLOCK);
		fcntl(errPipe[0], F_SETFL, O_NONBLOCK);
		entry->outFD = outPipe[0];
		entry->errFD = errPipe[0];
	}
#else
	// Windows: env, workdir, output capture and graceful stop handled via the
	// extended launch helper (utils::NewProcessEx) and its process-map bookkeeping.
	bool capture = (entry->spec.outputMode == EXTEXEC_OUTPUT_LOGFILE) ||
				   (entry->spec.outputMode == EXTEXEC_OUTPUT_PSYCLONELOG) ||
				   (entry->spec.outputMode == EXTEXEC_OUTPUT_MESSAGES);
	bool ignore = (entry->spec.outputMode == EXTEXEC_OUTPUT_IGNORE);
	// New process group so we can deliver a graceful CTRL_BREAK on stop.
	entry->pid = utils::NewProcessEx(entry->expandedCmdline.c_str(),
		entry->expandedWorkdir.size() ? entry->expandedWorkdir.c_str() : NULL,
		entry->spec.spaceName.c_str(),
		entry->spec.env.size() ? &entry->spec.env : NULL,
		capture, ignore, true);
	if (!entry->pid)
		return false;
#endif

	entry->state = EXTEXEC_STATE_LAUNCHING;
	if (entry->spec.shutdownOnDelete && entry->spec.spaceID) {
		node->manager->processMemory->setProcessOSID(entry->spec.spaceID, entry->pid);
		node->manager->processMemory->setProcessCommandLine(entry->spec.spaceID, entry->expandedCmdline.c_str());
	}
	LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: started '%s' for space '%s' (pid %u)...",
		entry->expandedCmdline.c_str(), entry->spec.spaceName.c_str(), entry->pid);
	return true;
}

void ExternalExecManager::closePipes(Entry* entry) {
#if !defined(WINDOWS)
	if (entry->outFD >= 0) { close(entry->outFD); entry->outFD = -1; }
	if (entry->errFD >= 0) { close(entry->errFD); entry->errFD = -1; }
#endif
}

bool ExternalExecManager::stopChild(Entry* entry, bool graceful) {
	if (!entry->pid)
		return true;
	uint32 pid = entry->pid;
#if !defined(WINDOWS)
	if (graceful) {
		if (kill((pid_t)pid, entry->spec.stopSignal) != 0) {
			// Already gone (e.g. reaped by the monitor)
			entry->pid = 0;
			closePipes(entry);
			return true;
		}
		uint64 start = GetTimeNow();
		int status;
		pid_t res;
		while (GetTimeAgeMS(start) < (int32)entry->spec.stopTimeout) {
			pumpOutput(entry);
			if ((res = waitpid((pid_t)pid, &status, WNOHANG)) != 0) {
				entry->pid = 0;
				closePipes(entry);
				return true;
			}
			utils::Sleep(20);
		}
		LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: '%s' (pid %u) did not stop within %u ms, sending SIGKILL...",
			entry->spec.spaceName.c_str(), pid, entry->spec.stopTimeout);
	}
	if (kill((pid_t)pid, SIGKILL) == 0) {
		int status;
		pid_t res;
		uint64 start = GetTimeNow();
		while ( ((res = waitpid((pid_t)pid, &status, WNOHANG)) == 0) && (GetTimeAgeMS(start) < 2000) )
			utils::Sleep(10);
	}
#else
	if (graceful) {
		// Deliver a graceful CTRL_BREAK to the child's process group, then wait
		// up to stopTimeout for it to exit before forcing termination.
		int returncode;
		if (utils::GetProcessStatus(pid, returncode) != PROC_RUNNING) {
			entry->pid = 0;
			closePipes(entry);
			return true;
		}
		utils::SendProcessBreak(pid, entry->spec.stopSignal);
		uint64 start = GetTimeNow();
		while (GetTimeAgeMS(start) < (int32)entry->spec.stopTimeout) {
			pumpOutput(entry);
			if (utils::GetProcessStatus(pid, returncode) != PROC_RUNNING) {
				entry->pid = 0;
				closePipes(entry);
				return true;
			}
			utils::Sleep(20);
		}
		LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: '%s' (pid %u) did not stop within %u ms, terminating...",
			entry->spec.spaceName.c_str(), pid, entry->spec.stopTimeout);
	}
	// Forceful fallback: TerminateProcess via the process map.
	utils::EndProcess(pid);
#endif
	entry->pid = 0;
	closePipes(entry);
	return true;
}

void ExternalExecManager::handleFailure(Entry* entry, const char* reason) {
	// Called with mutex held (or from registerExecutable before thread start)
	stopChild(entry, false);

	if (!entry->spec.autoRestart) {
		entry->state = EXTEXEC_STATE_STOPPED;
		LogPrint(node->nodeID, LOG_NODE, 0, "ExternalExec: '%s' %s; autorestart=\"no\", not restarting...",
			entry->spec.spaceName.c_str(), reason);
		return;
	}

	uint64 now = GetTimeNow();

	// restartlimit within restartwindow
	if (entry->spec.restartLimit) {
		while (entry->restartTimes.size() &&
				(GetTimeAgeMS(entry->restartTimes.front()) > (int64)entry->spec.restartWindow))
			entry->restartTimes.pop_front();
		if (entry->restartTimes.size() >= entry->spec.restartLimit) {
			entry->state = EXTEXEC_STATE_FAILED;
			LogPrint(node->nodeID, LOG_NODE, 0,
				"ExternalExec: '%s' %s; restart limit reached (%u restarts within %u ms), giving up...",
				entry->spec.spaceName.c_str(), reason, entry->spec.restartLimit, entry->spec.restartWindow);
			return;
		}
	}

	// Exponential backoff: doubles per consecutive fast failure, capped
	if (entry->launchTime && (GetTimeAgeMS(entry->launchTime) > EXTEXEC_HEALTHY_MS))
		entry->consecutiveFailures = 0;
	uint64 delay = entry->spec.restartDelay;
	uint32 shift = entry->consecutiveFailures;
	if (shift > 16) shift = 16;
	delay <<= shift;
	if (delay > entry->spec.maxRestartDelay)
		delay = entry->spec.maxRestartDelay;
	entry->consecutiveFailures++;
	entry->restartTimes.push_back(now);
	entry->restartDue = now + delay * 1000; // GetTimeNow() is microseconds
	entry->state = EXTEXEC_STATE_WAITRESTART;
	LogPrint(node->nodeID, LOG_NODE, 0, "ExternalExec: '%s' %s; restarting in %u ms...",
		entry->spec.spaceName.c_str(), reason, (uint32)delay);
}

void ExternalExecManager::routeLine(Entry* entry, const char* line, bool isStderr) {
	switch (entry->spec.outputMode) {
	case EXTEXEC_OUTPUT_LOGFILE:
		if (entry->expandedLogfile.size()) {
			FILE* f = fopen(entry->expandedLogfile.c_str(), "a");
			if (f) {
				fprintf(f, "%s [%s] %s\n", PrintTimeString(GetTimeNow()).c_str(), isStderr ? "err" : "out", line);
				fclose(f);
			}
		}
		break;
	case EXTEXEC_OUTPUT_PSYCLONELOG:
		LogPrint(node->nodeID, LOG_NODE, isStderr ? 0 : 1, "[%s %s] %s",
			entry->spec.moduleName.c_str(), isStderr ? "stderr" : "stdout", line);
		break;
	case EXTEXEC_OUTPUT_MESSAGES:
		{
			PsyType type = node->getPsyType(utils::StringFormat("%s.%s",
				entry->spec.moduleName.c_str(), isStderr ? "Stderr" : "Stdout").c_str());
			if (type.isValid()) {
				DataMessage* msg = new DataMessage();
				msg->setType(type);
				msg->setString("line", line);
				node->processMessage(msg);
				LogPrint(node->nodeID, LOG_NODE, 3, "ExternalExec: posted %s.%s message: %s",
					entry->spec.moduleName.c_str(), isStderr ? "Stderr" : "Stdout", line);
				delete(msg);
			}
			else
				LogPrint(node->nodeID, LOG_NODE, 0, "ExternalExec: could not create message type %s.%s for output line...",
					entry->spec.moduleName.c_str(), isStderr ? "Stderr" : "Stdout");
		}
		break;
	default:
		break;
	}
}

void ExternalExecManager::pumpOutput(Entry* entry) {
#if !defined(WINDOWS)
	char buf[4096];
	int fds[2] = { entry->outFD, entry->errFD };
	for (int n = 0; n < 2; n++) {
		if (fds[n] < 0)
			continue;
		bool isStderr = (n == 1);
		std::string& lineBuf = isStderr ? entry->errBuf : entry->outBuf;
		ssize_t len;
		while ((len = read(fds[n], buf, sizeof(buf) - 1)) > 0) {
			buf[len] = 0;
			lineBuf += buf;
			size_t pos;
			while ((pos = lineBuf.find('\n')) != std::string::npos) {
				std::string line = lineBuf.substr(0, pos);
				if (line.size() && (line[line.size()-1] == '\r'))
					line.resize(line.size()-1);
				lineBuf.erase(0, pos + 1);
				if (line.size())
					routeLine(entry, line.c_str(), isStderr);
			}
		}
	}
#else
	if (!entry->pid)
		return;
	// Drain any captured bytes from the child's stdout/stderr pipes, then split
	// into lines and route exactly as the POSIX path does.
	std::string out, err;
	if (!utils::ReadProcessOutput(entry->pid, out, err))
		return;
	std::string* chunks[2] = { &out, &err };
	for (int n = 0; n < 2; n++) {
		if (chunks[n]->empty())
			continue;
		bool isStderr = (n == 1);
		std::string& lineBuf = isStderr ? entry->errBuf : entry->outBuf;
		lineBuf += *chunks[n];
		size_t pos;
		while ((pos = lineBuf.find('\n')) != std::string::npos) {
			std::string line = lineBuf.substr(0, pos);
			if (line.size() && (line[line.size()-1] == '\r'))
				line.resize(line.size()-1);
			lineBuf.erase(0, pos + 1);
			if (line.size())
				routeLine(entry, line.c_str(), isStderr);
		}
	}
#endif
}

bool ExternalExecManager::removeByModule(uint32 moduleID) {
	std::lock_guard<std::mutex> lock(mutex);
	std::list<Entry*>::iterator i = entries.begin(), e = entries.end();
	while (i != e) {
		if ((*i)->spec.moduleID == moduleID) {
			Entry* entry = *i;
			if (entry->spec.shutdownOnDelete) {
				LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: module %u deleted, stopping '%s' (pid %u)...",
					moduleID, entry->spec.spaceName.c_str(), entry->pid);
				stopChild(entry, true);
			}
			else
				LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: module %u deleted, leaving '%s' running (shutdownondelete=\"no\")...",
					moduleID, entry->spec.spaceName.c_str());
			entry->state = EXTEXEC_STATE_STOPPED;
			entries.erase(i);
			delete(entry);
			return true;
		}
		i++;
	}
	return false;
}

bool ExternalExecManager::stopAll() {
	{
		std::lock_guard<std::mutex> lock(mutex);
		if (!shouldContinue)
			return true;
		shouldContinue = false;
		std::list<Entry*>::iterator i = entries.begin(), e = entries.end();
		while (i != e) {
			Entry* entry = *i;
			if (entry->pid && entry->spec.shutdownOnDelete) {
				LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: node shutdown, stopping '%s' (pid %u)...",
					entry->spec.spaceName.c_str(), entry->pid);
				stopChild(entry, true);
			}
			else if (entry->pid)
				LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: node shutdown, leaving '%s' (pid %u) running (shutdownondelete=\"no\")...",
					entry->spec.spaceName.c_str(), entry->pid);
			entry->state = EXTEXEC_STATE_STOPPED;
			i++;
		}
	}
	uint32 timeout = 2000;
	while (threadRunning && timeout) {
		utils::Sleep(10);
		timeout -= 10;
	}
	return true;
}

uint32 ExternalExecManager::count() {
	std::lock_guard<std::mutex> lock(mutex);
	return (uint32)entries.size();
}

int ExternalExecManager::getState(const char* spaceName) {
	std::lock_guard<std::mutex> lock(mutex);
	std::list<Entry*>::iterator i = entries.begin(), e = entries.end();
	while (i != e) {
		if (stricmp((*i)->spec.spaceName.c_str(), spaceName) == 0)
			return (*i)->state;
		i++;
	}
	return -1;
}

bool ExternalExecManager::run() {
	int returncode;
	uint64 lastseen;
	while (shouldContinue) {
		{
			std::lock_guard<std::mutex> lock(mutex);
			std::list<Entry*>::iterator i = entries.begin(), e = entries.end();
			while (i != e) {
				Entry* entry = *i;
				switch (entry->state) {
				case EXTEXEC_STATE_LAUNCHING:
					pumpOutput(entry);
					if (utils::GetProcessStatus(entry->pid, returncode) != PROC_RUNNING) {
						pumpOutput(entry);
						handleFailure(entry, utils::StringFormat("exited during startup (code %d)", returncode).c_str());
					}
					else if (entry->spec.spaceID &&
							(node->manager->processMemory->getProcessStatus(entry->spec.spaceID, lastseen) >= PSYPROC_ACTIVE)) {
						entry->state = EXTEXEC_STATE_RUNNING;
						LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: space '%s' is active (pid %u)...",
							entry->spec.spaceName.c_str(), entry->pid);
					}
					else if (GetTimeAgeMS(entry->launchTime) > (int64)entry->spec.startupTimeout) {
						handleFailure(entry, utils::StringFormat("did not connect its space within %u ms", entry->spec.startupTimeout).c_str());
					}
					break;
				case EXTEXEC_STATE_RUNNING:
					pumpOutput(entry);
					if (utils::GetProcessStatus(entry->pid, returncode) != PROC_RUNNING) {
						pumpOutput(entry);
						handleFailure(entry, utils::StringFormat("exited (code %d)", returncode).c_str());
					}
					else if (entry->spec.spaceID) {
						uint8 status = node->manager->processMemory->getProcessStatus(entry->spec.spaceID, lastseen);
						if (status < PSYPROC_ACTIVE)
							// The node dropped the space (heartbeat timeout / disconnect) while the OS process is still alive
							handleFailure(entry, "lost its space connection (process alive but space no longer active; hung?)");
						else if (lastseen && (GetTimeAgeMS(lastseen) > EXTEXEC_HEARTBEAT_MS))
							handleFailure(entry, utils::StringFormat("appears hung (no heartbeat for %d ms)", (int)GetTimeAgeMS(lastseen)).c_str());
					}
					break;
				case EXTEXEC_STATE_WAITRESTART:
					if (GetTimeNow() >= entry->restartDue) {
						LogPrint(node->nodeID, LOG_NODE, 1, "ExternalExec: restarting '%s'...", entry->spec.spaceName.c_str());
						if (!launch(entry))
							handleFailure(entry, "could not restart");
					}
					break;
				default:
					break;
				}
				i++;
			}
		}
		utils::Sleep(EXTEXEC_POLL_MS);
	}
	threadRunning = false;
	return true;
}

}
