

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 external-module test child (cross-platform: builds on POSIX and Windows).
// Used by the psy_extexec_* unit tests in Psyclone/src/PsycloneTests.cpp.
//
// Also hosts the T1.6 <triggergroup> test cranks (TGDriver/TGSink). Per the
// "nothing Psyclone in CMSDK" rule these are NOT built-in CMSDK internal
// cranks; the child registers them by name via registerCrankCallback() so the
// psy_triggergroup_* tests' specs (function="TGDriver"/"TGSink") bind to them.
//
// Usage: PsyTestChild port=<port> space=<name> [mode=run|crashafter|noconnect|hang] [ms=<ms>]
//   run        connect PsySpace <name> to <port>, stay alive, print output (default)
//   crashafter connect, then exit(3) after <ms> ms
//   noconnect  never connect the PsySpace (exercises startuptimeout); stays alive
//   hang       connect, then stop servicing the space (sleep forever) so its
//              heartbeat goes stale (portable stand-in for a hung process)
// (TGDriver/TGSink cranks are always registered in run mode; they only fire
//  for modules whose spec declares function="TGDriver"/"TGSink".)

#include "PsySpace.h"

using namespace cmlabs;

static const char* getArg(int argc, char* argv[], const char* key) {
	size_t klen = strlen(key);
	for (int n=1; n<argc; n++)
		if ((strncmp(argv[n], key, klen) == 0) && (argv[n][klen] == '='))
			return argv[n] + klen + 1;
	return NULL;
}

// ============================================================================
//  T1.6 <triggergroup> test cranks (host-side; not part of the CMSDK).
// ----------------------------------------------------------------------------
//  TGDriver: on the "Ready" trigger, posts a scripted sequence of member
//    messages named by its "Steps" parameter (comma-separated post names, each
//    optionally "Name=N" to post that member N times), pausing StepDelayMS
//    between posts. Post names A/B/C map to the module's <post> entries.
//  TGSink: consumes joined sets via waitForNewMessageGroup() and appends one
//    line per join to the file named by its "ResultFile" parameter, in the
//    format  groupName|memberCount|member1,member2,...  (read back by the
//    psy_triggergroup_* tests in PsycloneTests.cpp).
// ----------------------------------------------------------------------------

int8 TGDriver(PsyAPI* api) {
	api->logPrint(1, "TGDriver started");
	const char* triggerName = NULL;
	while (api->shouldContinue()) {
		DataMessage* msg = api->waitForNewMessage(100, triggerName);
		if (!msg) continue;
		// Fire the scripted post sequence once, on the Ready trigger.
		std::string steps = api->getParameterString("Steps");
		int64 stepDelay = 0;
		api->getParameter("StepDelayMS", stepDelay);
		std::vector<std::string> tokens = utils::TextListSplit(steps.c_str(), ",", true, true);
		api->logPrint(1, "Started running, %u steps...", (uint32)tokens.size());
		for (size_t i = 0; i < tokens.size(); i++) {
			// token is a post name, optionally "Name=N" to repeat N times.
			std::string tok = tokens[i];
			std::string name = tok;
			int reps = 1;
			size_t eq = tok.find('=');
			if (eq != std::string::npos) {
				name = tok.substr(0, eq);
				reps = (int)utils::Ascii2Int64(tok.substr(eq + 1).c_str());
				if (reps < 1) reps = 1;
			}
			for (int r = 0; r < reps; r++) {
				api->postOutputMessage(name.c_str());
				if (stepDelay > 0) utils::Sleep((uint32)stepDelay);
			}
		}
	}
	return 0;
}

int8 TGSink(PsyAPI* api) {
	api->logPrint(1, "TGSink started");
	std::string resultFile = api->getParameterString("ResultFile");
	std::string groupName;
	while (api->shouldContinue()) {
		std::map<std::string, DataMessage*> members = api->waitForNewMessageGroup(100, groupName);
		if (members.empty()) continue;
		// Build "member1,member2,..." from the joined-set keys (member trigger names).
		std::string memberList;
		for (std::map<std::string, DataMessage*>::iterator it = members.begin();
				it != members.end(); ++it) {
			if (memberList.length()) memberList += ",";
			memberList += it->first;
		}
		api->logPrint(1, "TGSink '%s' recorded join '%s' with %d members",
			groupName.c_str(), groupName.c_str(), (int)members.size());
		if (resultFile.length()) {
			std::string line = utils::StringFormat("%s|%d|%s\n",
				groupName.c_str(), (int)members.size(), memberList.c_str());
			// Append (one line per join); read back by the test harness.
			std::string existing = utils::ReadAFileString(resultFile.c_str());
			existing += line;
			utils::WriteAFile(resultFile.c_str(), existing.c_str(), (uint32)existing.size());
		}
	}
	return 0;
}

// ----------------------------------------------------------------------------
//  Trigger-filter test cranks (psy_triggerfilter in PsycloneTests.cpp):
//  FiltDriver: on the Ready trigger, posts one "V" message per comma token of
//    its "Values" parameter, each carrying user entry "k" = token (stored as a
//    string; the subscription engine parses it numerically for the numeric
//    filter operations).
//  FiltSink: records every fired trigger to its "ResultFile" parameter as one
//    "triggerName|k" line per delivery (read back by the test harness).
// ----------------------------------------------------------------------------

int8 FiltDriver(PsyAPI* api) {
	api->logPrint(1, "FiltDriver started");
	const char* triggerName = NULL;
	while (api->shouldContinue()) {
		DataMessage* msg = api->waitForNewMessage(100, triggerName);
		if (!msg) continue;
		std::string values = api->getParameterString("Values");
		std::string posts = api->getParameterString("PostNames");
		std::string dbgFile = api->getParameterString("DebugFile");
		std::vector<std::string> toks = utils::TextListSplit(values.c_str(), ",", true, true);
		std::vector<std::string> postNames = utils::TextListSplit(posts.c_str(), ",", true, true);
		api->logPrint(1, "FiltDriver posting %u values on %u posts...",
			(uint32)toks.size(), (uint32)postNames.size());
		for (size_t i = 0; i < toks.size(); i++) {
			for (size_t p = 0; p < postNames.size(); p++) {
				DataMessage* out = new DataMessage();
				out->setString("k", toks[i].c_str());
				int32 rc = api->postOutputMessage(postNames[p].c_str(), out);
				if (dbgFile.length()) {
					std::string existing = utils::ReadAFileString(dbgFile.c_str());
					existing += utils::StringFormat("POST|%s|%s|%d\n",
						postNames[p].c_str(), toks[i].c_str(), rc);
					utils::WriteAFile(dbgFile.c_str(), existing.c_str(), (uint32)existing.size());
				}
			}
		}
	}
	return 0;
}

int8 FiltSink(PsyAPI* api) {
	api->logPrint(1, "FiltSink started");
	std::string resultFile = api->getParameterString("ResultFile");
	const char* triggerName = NULL;
	while (api->shouldContinue()) {
		DataMessage* msg = api->waitForNewMessage(100, triggerName);
		if (!msg) continue;
		std::string k = msg->getAsString("k");
		api->logPrint(1, "FiltSink fired trigger '%s' k='%s'",
			triggerName ? triggerName : "", k.c_str());
		if (resultFile.length()) {
			std::string line = utils::StringFormat("%s|%s\n",
				triggerName ? triggerName : "", k.c_str());
			std::string existing = utils::ReadAFileString(resultFile.c_str());
			existing += line;
			utils::WriteAFile(resultFile.c_str(), existing.c_str(), (uint32)existing.size());
		}
	}
	return 0;
}

int main(int argc, char* argv[]) {
	const char* sport = getArg(argc, argv, "port");
	const char* sname = getArg(argc, argv, "space");
	const char* smode = getArg(argc, argv, "mode");
	const char* sms = getArg(argc, argv, "ms");
	uint16 port = sport ? (uint16)atoi(sport) : 10000;
	std::string spaceName = sname ? sname : "Ext1";
	std::string mode = smode ? smode : "run";
	int ms = sms ? atoi(sms) : 2000;

	std::string cwd = utils::GetCurrentDir();
	const char* myvar = getenv("T14_VAR");
	printf("T14CHILD started space=%s port=%u mode=%s cwd=%s T14_VAR=%s pid=%u\n",
		spaceName.c_str(), port, mode.c_str(), cwd.c_str(), myvar ? myvar : "(unset)",
		utils::GetLocalProcessID());
	fprintf(stderr, "T14CHILD stderr-check space=%s\n", spaceName.c_str());
	fflush(stdout); fflush(stderr);

	if (mode == "noconnect") {
		while (true) utils::Sleep(500);
		return 0;
	}

	PsySpace* space = new PsySpace(spaceName.c_str());
	uint64 start = GetTimeNow();
	while (!space->connect(port)) {
		printf("T14CHILD waiting to connect to %u...\n", port);
		fflush(stdout);
		utils::Sleep(250);
		if (GetTimeAgeMS(start) > 30000) {
			printf("T14CHILD giving up connecting\n");
			return 2;
		}
	}
	// T1.6: register the <triggergroup> test cranks by their spec name (which
	// equals the function name) so the module's <crank name="TGDriver"/> and
	// <crank name="TGSink"/> bind to these host-side callbacks. The crank map
	// entries only exist once the node has pushed module config to this space,
	// so retry briefly until registration succeeds.
	{
		uint64 regStart = GetTimeNow();
		bool regDriver = false, regSink = false, regFDriver = false, regFSink = false;
		while (GetTimeAgeMS(regStart) < 5000) {
			if (!regDriver)  regDriver  = space->registerCrankCallback("TGDriver", TGDriver);
			if (!regSink)    regSink    = space->registerCrankCallback("TGSink", TGSink);
			if (!regFDriver) regFDriver = space->registerCrankCallback("FiltDriver", FiltDriver);
			if (!regFSink)   regFSink   = space->registerCrankCallback("FiltSink", FiltSink);
			// A spec only declares one crank pair, so stop as soon as a full
			// pair is bound (the other pair's names are absent and would keep
			// failing for the whole 5s window, delaying space->start()).
			if ((regDriver && regSink) || (regFDriver && regFSink)) break;
			utils::Sleep(50);
		}
	}
	if (!space->start()) {
		printf("T14CHILD could not start space\n");
		return 2;
	}
	printf("T14CHILD connected space=%s\n", spaceName.c_str());
	fflush(stdout);

	if (mode == "crashafter") {
		utils::Sleep(ms);
		printf("T14CHILD crashing now (exit 3)\n");
		fflush(stdout);
		exit(3);
	}

	if (mode == "hang") {
		// Drop the space below ACTIVE while keeping the OS process alive: this
		// hits the node's "process alive but space no longer active" hang path.
		// (Portable stand-in for a hung/SIGSTOP process that stops heart-beating.)
		space->shutdown();
		while (true) utils::Sleep(1000);
	}

	int count = 0;
	while (true) {
		utils::Sleep(1000);
		printf("T14CHILD alive %d\n", ++count);
		fflush(stdout);
	}
	return 0;
}
