

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "PsyTest.h"

#include <vector>
#include <string>
#include <cstdio>
#include <cstdarg>
#include <algorithm>

namespace cmlabs {

// ***************************************************************
// Test result aggregation
//
// Every stage master records a one-line PASS/FAIL result here and
// also logs it. All test modules run as threads in the same process
// (single node), so a process-global table works; the PsyTest_Summary
// crank (wired to the Shutdown module) prints the final verdict.
// ***************************************************************

// Use the CMSDK cross-platform mutex (utils::Mutex) rather than pthreads
// directly, so this test builds and runs on Windows as well as POSIX.
static utils::Mutex g_psytestMutex;
static std::vector<std::string> g_psytestLines;
static int  g_psytestPass = 0;
static int  g_psytestFail = 0;
static bool g_psytestSummaryDone = false;

// Structured per-test report. Each stage master reports one of these (via
// psytestReport) with a headline, PASS/FAIL status, a short description of
// what the test contained, and — on failure — a detailed failure report.
// PsyTest_Summary (the last module in the chain) aggregates them and prints
// the formatted summary. The `order` field lets the summary present stages
// in chain order regardless of the order the reports arrive in.
struct PsyTestReport {
	int         order;
	bool        pass;
	std::string headline;      // half a line: what was tested
	std::string description;   // 1-3 lines: what the test contained
	std::string detail;        // 2-5 lines (FAIL only): what failed
};
static std::vector<PsyTestReport> g_psytestReports;

static void psytestRecord(bool pass, const char* line) {
	g_psytestMutex.enter(1000, __FUNCTION__);
	g_psytestLines.push_back(line ? line : "");
	if (pass) g_psytestPass++; else g_psytestFail++;
	g_psytestMutex.leave();
}

// Record a structured report from a stage. `order` fixes chain position;
// `headline`/`description` are always shown; `detail` is shown only on FAIL.
static void psytestReport(int order, bool pass, const char* headline,
                          const char* description, const char* detail) {
	PsyTestReport r;
	r.order       = order;
	r.pass        = pass;
	r.headline    = headline     ? headline     : "";
	r.description = description   ? description   : "";
	r.detail      = (!pass && detail) ? detail   : "";
	g_psytestMutex.enter(1000, __FUNCTION__);
	g_psytestReports.push_back(r);
	g_psytestMutex.leave();
}

// Return the q-th percentile (0..100) of the recorded latency samples (us).
// Uses the nearest-rank method on a sorted copy; caller sorts once beforehand.
static double psytestPercentile(const std::vector<double>& sorted, double q) {
	if (sorted.empty()) return 0;
	size_t idx = (size_t)((q / 100.0) * (double)sorted.size());
	if (idx >= sorted.size()) idx = sorted.size() - 1;
	return sorted[idx];
}

// Build a ping-pong "ball" DataMessage carrying a payload of `payloadSize`
// bytes (0 = empty, original behaviour). The payload is attached as message
// content via setData (data-map / shared-memory content path), so it exercises
// the real transport at the requested size without changing any struct layout.
static DataMessage* psytestMakeBall(uint32 payloadSize) {
	DataMessage* m = new DataMessage();
	if (payloadSize > 0) {
		static std::vector<char> g_payloadBuf;
		if (g_payloadBuf.size() < payloadSize)
			g_payloadBuf.assign(payloadSize, 'x');
		m->setData("p", g_payloadBuf.data(), payloadSize);
	}
	return m;
}

// Format "p50=... p90=... p95=... p99=... p99.9=..." from raw per-message
// latency samples (microseconds). Sorts the vector in place.
static std::string psytestTailStats(std::vector<double>& samples) {
	if (samples.empty()) return "(no samples)";
	std::sort(samples.begin(), samples.end());
	char buf[256];
	snprintf(buf, sizeof(buf), "p50=%.1fus p90=%.1fus p95=%.1fus p99=%.1fus p99.9=%.1fus",
		psytestPercentile(samples, 50), psytestPercentile(samples, 90),
		psytestPercentile(samples, 95), psytestPercentile(samples, 99),
		psytestPercentile(samples, 99.9));
	return buf;
}

// Emit one PASS/FAIL result line (always shown) and record it for the summary.
static void psytestResult(PsyAPI* api, bool pass, const char* fmt, ...) {
	char detail[512];
	va_list ap;
	va_start(ap, fmt);
	vsnprintf(detail, sizeof(detail), fmt, ap);
	va_end(ap);

	char line[640];
	snprintf(line, sizeof(line), "[ %s ] %s", pass ? "PASS" : "FAIL", detail);
	api->logPrint(0, "%s", line);
	psytestRecord(pass, line);
}

// Combined helper: emit the existing one-line PASS/FAIL log line (via the
// printf-style `fmt`, as before) AND record a structured report for the
// final summary. `order` is the stage's chain position; `headline` is a
// half-line of what was tested; `description` is 1-3 lines of what the test
// contained; `detail` is a 2-5 line failure report (shown only on FAIL, may
// be NULL). This keeps the live per-stage line unchanged while giving
// PsyTest_Summary the richer per-test blocks.
static void psytestResultR(PsyAPI* api, int order, bool pass,
                           const char* headline, const char* description,
                           const char* detail, const char* fmt, ...) {
	char oneline[512];
	va_list ap;
	va_start(ap, fmt);
	vsnprintf(oneline, sizeof(oneline), fmt, ap);
	va_end(ap);
	psytestResult(api, pass, "%s", oneline);
	psytestReport(order, pass, headline, description, detail);
}


// ***************************************
// ************** Intro ******************
// ***************************************
//
// First module to run (triggered by Psyclone.Ready). Prints a short banner
// telling the user which test stages they are about to see and that the run
// takes a little while, then posts "Done" to activate the first stage context
// and kick off the chain.

int8 PsyTest_Intro(PsyAPI* api) {
	DataMessage* inMsg;
	const char* triggerName;
	bool started = false;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			if (!started && ((stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0))) {
				started = true;
				api->logPrint(0, "============================================================");
				api->logPrint(0, " Psyclone function + performance test starting");
				api->logPrint(0, " ");
				api->logPrint(0, " You will see one [ PASS ]/[ FAIL ] line per stage, with");
				api->logPrint(0, " performance data, in this order:");
				api->logPrint(0, "   1. TimeMaster     - clock round-trip timing");
				api->logPrint(0, "   2. PingMaster     - reliable ping/pong throughput");
				api->logPrint(0, "   3. PingUDPMaster  - best-effort (UDP) ping/pong throughput");
				api->logPrint(0, "   4. SignalMaster   - signal emit/receive round-trips");
				api->logPrint(0, "   5. ContextMaster  - dynamic context switching + timer triggers");
				api->logPrint(0, "   6. RetrieveTest   - whiteboard / file / data catalog retrieval");
				api->logPrint(0, " ");
				api->logPrint(0, " Please be patient: the run takes ~35s (the timing stage alone");
				api->logPrint(0, " paces itself over ~30s), then PsyTest_Summary prints the verdict.");
				api->logPrint(0, "============================================================");
				api->postOutputMessage("Done");	// -> activates the first stage context
			}
		}
	}
	return 0;
}


// ***************************************
// ************** Time Test **************
// ***************************************

int8 PsyTest_TimeMaster(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);

	api->logPrint(2,"[%s] started running...", myName);

	DataMessage* inMsg;
	int64 counter = 0;
	uint64 start = 0, tLocalSent, tRemoteSent, tLocalReceived;
	const char* triggerName;
	int64 count = 10000;
	int64 nodes = 1;

	int64 val;
	if (api->hasParameter("Count"))
		if (api->getParameter("Count", val))
			count = val;
	if (api->hasParameter("Nodes"))
		if (api->getParameter("Nodes", val))
			nodes = val;

	uint64 replies = 0, anomalies = 0;
	double sumRtt = 0;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
				api->logPrint(2,"[%s] starting %lld count test...", myName, (long long)count);
				start = GetTimeNow();
				api->postOutputMessage("OutputMsg");
				counter = 1;
				replies = 0; anomalies = 0; sumRtt = 0;
			}
			else if (stricmp(triggerName, "InputMsg") == 0) {

				tLocalSent = start;
				tLocalReceived = api->getCurrentMessageReceiveTime();
				tRemoteSent = inMsg->getSendTime();

				if (tLocalSent >= tRemoteSent) {
					anomalies++;
					api->logPrint(2, "Module %u time is behind by at least %s", inMsg->getFrom(), PrintTimeDifString(tLocalSent-tRemoteSent).c_str() );
				}
				else if (tRemoteSent >= tLocalReceived) {
					anomalies++;
					api->logPrint(2, "Module %u time is ahead by at least %s", inMsg->getFrom(), PrintTimeDifString(tRemoteSent-tLocalReceived).c_str());
				}
				else {
					replies++;
					sumRtt += (double)(tLocalReceived - tLocalSent);
					api->logPrint(2, "Module %u time is ok, msg time %s = out: %s + back: %s", inMsg->getFrom(),
						PrintTimeDifString(tLocalReceived-tLocalSent).c_str(),
						PrintTimeDifString(tRemoteSent-tLocalSent).c_str(),
						PrintTimeDifString(tLocalReceived-tRemoteSent).c_str());
				}
			}
			else {
				api->logPrint(2,"[%s] got other message type '%s'...", myName, api->typeToText(inMsg->getType()).c_str());
			}
		}
		else if (counter > count) {
			// we are done, do nothing
		}
		else if (GetTimeAgeMS(start) > 3000) {
			if (counter == count) {
				if (replies > 0)
					psytestResultR(api, 1, true,
						"Clock round-trip timing (master/slave)",
						"Master posts time.2 to two slaves; each replies time.1.\n"
						"The master times each round trip and flags clock skew.",
						NULL,
						"%-13s rounds=%lld replies=%llu avg_rtt=%.1fus skew_warnings=%llu",
						myName, (long long)count, (unsigned long long)replies, sumRtt/(double)replies, (unsigned long long)anomalies);
				else
					psytestResultR(api, 1, false,
						"Clock round-trip timing (master/slave)",
						"Master posts time.2 to two slaves; each replies time.1.\n"
						"The master times each round trip and flags clock skew.",
						"No time.1 replies were received from either slave within\n"
						"the per-round 3s window across all rounds — the timing ring\n"
						"never completed a single round trip (slaves not triggering\n"
						"on time.2, or replies not routed back to the master).",
						"%-13s no replies from time slaves (rounds=%lld)",
						myName, (long long)count);
				api->postOutputMessage("Done");
				counter++;
			}
			else {
				api->postOutputMessage("OutputMsg");
				start = GetTimeNow();
				counter++;
			}
		}
	}

	delete [] myName;
	return 0;
}

int8 PsyTest_TimeSlave(PsyAPI* api) {
	DataMessage* inMsg;
	const char* triggerName;
	while (api->shouldContinue()) {
		if (inMsg = api->waitForNewMessage(100, triggerName))
			api->postOutputMessage();
	}
	return 0;
}


// ***************************************
// *********** Ping Pong Test ************
// ***************************************

int8 PsyTest_PingMaster(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);

	// This crank backs both the guaranteed stage (module "PingMaster", chain
	// order 2) and the best-effort/UDP stage (module "PingUDPMaster", order 3).
	// Distinguish by module name so each reports its own headline/description.
	bool isUDP = (strstr(myName, "UDP") != NULL) || (strstr(myName, "udp") != NULL);
	int  reportOrder = isUDP ? 3 : 2;
	const char* reportHeadline = isUDP
		? "Best-effort (UDP) ping/pong throughput"
		: "Reliable (guaranteed) ping/pong throughput";
	const char* reportDesc = isUDP
		? "Same 3-module ring but posts use guaranteed=\"no\" (best-\n"
		  "effort); measures unguaranteed-transport throughput."
		: "Bounces a ball round a ring of 3 modules for Count*Cycles\n"
		  "round trips; measures reliable-transport throughput.";

	api->logPrint(2,"[%s] started running...", myName);

	DataMessage* inMsg;
	int64 counter = 0;
	const char* triggerName;
	int64 count = 10000;
	int64 cycles = 10;
	int64 nodes = 2;

	int64 val;
	if (api->hasParameter("Cycles"))
		if (api->getParameter("Cycles", val))
			cycles = val;
	if (api->hasParameter("Count"))
		if (api->getParameter("Count", val))
			count = val;
	if (api->hasParameter("Nodes"))
		if (api->getParameter("Nodes", val))
			nodes = val;

	int64 payloadSize = 0;   // bytes of message payload per ball (0 = empty)
	if (api->hasParameter("PayloadSize"))
		if (api->getParameter("PayloadSize", val))
			payloadSize = val;

	int64 totalMessageCount = cycles*count;

	uint64 stageStart = 0, lastMsg = 0;
	double sumAge = 0;
	std::vector<double> ageSamples;          // per-message latency (us) for tail stats
	ageSamples.reserve((size_t)totalMessageCount);
	bool running = false, done = false;

	while (api->shouldContinue()) {
		inMsg = api->waitForNewMessage(100, triggerName);
		if (!inMsg) {
			// Watchdog: a ping-pong has continuous sub-ms traffic, so a multi-second
			// gap means the ring stalled (e.g. a dropped UDP message). Report and
			// release the chain instead of hanging forever.
			if (running && !done && GetTimeAgeMS(lastMsg) > 3000) {
				psytestResultR(api, reportOrder, false, reportHeadline, reportDesc,
					"The ring stalled: continuous sub-ms ball traffic stopped for\n"
					"over 3s before all round trips completed. A ball was dropped\n"
					"or a slave stopped forwarding, so the ring never closed — see\n"
					"got-vs-expected counts below (UDP is best-effort, so a stall\n"
					"here means messages went missing beyond tolerance).",
					"%-13s stalled: got %lld of %lld msgs",
					myName, (long long)counter, (long long)totalMessageCount);
				api->postOutputMessage("Done");
				done = true; running = false;
			}
			continue;
		}

		if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
			api->logPrint(2,"[%s] starting %lld cycles test...", myName, (long long)cycles);
			running = true; done = false;
			counter = 0; sumAge = 0; ageSamples.clear();
			stageStart = lastMsg = GetTimeNow();
			api->postOutputMessage("OutputMsg", psytestMakeBall((uint32)payloadSize));	// first ball
		}
		else if (!done && (stricmp(triggerName, "InputMsg") == 0)) {
			lastMsg = GetTimeNow();
			double age = (double)GetTimeAge(inMsg->getCreatedTime());
			sumAge += age;
			ageSamples.push_back(age);
			counter++;
			if (counter == 1)
				stageStart = GetTimeNow();
			if (counter >= totalMessageCount) {
				double elapsedUs = (double)(GetTimeNow() - stageStart);
				double thr = (elapsedUs > 0) ? ((double)totalMessageCount * 1000000.0 / elapsedUs) : 0;
				psytestResultR(api, reportOrder, true, reportHeadline, reportDesc, NULL,
					"%-13s msgs=%lld throughput=%.0f msg/s avg_age=%.1fus %s",
					myName, (long long)totalMessageCount, thr, sumAge/(double)counter,
					psytestTailStats(ageSamples).c_str());
				api->postOutputMessage("Done", new DataMessage());
				done = true; running = false;
			}
			else {
				api->postOutputMessage("OutputMsg", psytestMakeBall((uint32)payloadSize));	// bounce the ball
			}
		}
		// else: ignore (already done, or unexpected type)
	}

	delete [] myName;
	return 0;
}

int8 PsyTest_PingSlave(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);

	api->logPrint(2,"[%s] started running...", myName);

	const char* triggerName;
	DataMessage* inMsg;
	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			// Forward the ball, preserving any payload attached by the master so
			// the full ring carries the payload (fair like-for-like transport).
			uint32 psz = 0;
			const char* pd = inMsg->getData("p", psz);
			if (pd && psz > 0) {
				DataMessage* out = new DataMessage();
				out->setData("p", pd, psz);
				api->postOutputMessage("OutputMsg", out);
			} else {
				api->postOutputMessage("OutputMsg");
			}
		}
	}

	delete [] myName;
	return 0;
}


// ***************************************
// ************* Signal Test *************
// ***************************************

int8 PsyTest_SignalMaster(PsyAPI* api) {
	DataMessage* inMsg;
	const char* triggerName;

	char* myName = new char[256];
	api->getModuleName(myName, 256);
	api->logPrint(2,"[%s] started running...", myName);

	int64 counter = 0;
	uint64 count = 10000;
	uint64 cycles = 10;
	uint64 nodes = 1;

	int64 val;
	if (api->hasParameter("Cycles"))
		if (api->getParameter("Cycles", val))
			cycles = (uint64)val;
	if (api->hasParameter("Count"))
		if (api->getParameter("Count", val))
			count = (uint64)val;
	if (api->hasParameter("Nodes"))
		if (api->getParameter("Nodes", val))
			nodes = (uint64)val;

	int64 totalMessageCount = cycles*count;

	uint64 lastTime = GetTimeNow() - 1000000;
	uint64 stageStart = 0, lastSig = 0;
	double sumAge = 0;
	std::vector<double> ageSamples;          // per-signal latency (us) for tail stats
	ageSamples.reserve((size_t)totalMessageCount);
	bool running = false, done = false;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(0, triggerName)) {
			if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
				api->logPrint(2,"[%s] emitting signals (count: %llu cycles: %llu)...", myName,
					(unsigned long long)count, (unsigned long long)cycles);
				running = true; done = false;
				counter = 0; sumAge = 0; ageSamples.clear();
				utils::Sleep(10);
				stageStart = lastSig = GetTimeNow();
				api->emitSignal("Output");
			}
		}
		else if (inMsg = api->waitForSignal("Input", 100, lastTime)) {
			lastSig = GetTimeNow();
			double age = (double)GetTimeAge(lastTime = inMsg->getCreatedTime());
			sumAge += age;
			ageSamples.push_back(age);
			counter++;
			if (counter == 1)
				stageStart = GetTimeNow();
			if (!done && counter >= totalMessageCount) {
				double elapsedUs = (double)(GetTimeNow() - stageStart);
				double thr = (elapsedUs > 0) ? ((double)totalMessageCount * 1000000.0 / elapsedUs) : 0;
				psytestResultR(api, 4, true,
					"Signal emit/receive round-trips",
					"Signals (lighter than messages) ring round 3 modules\n"
					"(My.Signal.1->2->3->1); master counts the round trips.",
					NULL,
					"%-13s signals=%lld throughput=%.0f sig/s avg_age=%.1fus %s",
					myName, (long long)totalMessageCount, thr, sumAge/(double)counter,
					psytestTailStats(ageSamples).c_str());
				api->postOutputMessage("Done");
				done = true; running = false;
			}
			else if (!done) {
				api->emitSignal("Output");
			}
		}
		else {
			// Watchdog: signals bounce continuously; a multi-second gap means a stall.
			if (running && !done && GetTimeAgeMS(lastSig) > 3000) {
				psytestResultR(api, 4, false,
					"Signal emit/receive round-trips",
					"Signals (lighter than messages) ring round 3 modules\n"
					"(My.Signal.1->2->3->1); master counts the round trips.",
					"The signal ring stalled: continuous signal traffic stopped\n"
					"for over 3s before all round trips completed. A signal was\n"
					"not received or not re-emitted by one of the slaves, so the\n"
					"ring broke — see got-vs-expected counts below.",
					"%-13s stalled: got %lld of %lld signals",
					myName, (long long)counter, (long long)totalMessageCount);
				api->postOutputMessage("Done");
				done = true; running = false;
			}
		}
	}

	delete [] myName;
	return 0;
}

int8 PsyTest_SignalSlave(PsyAPI* api) {
	DataMessage* inMsg;
	const char* triggerName;

	char* myName = new char[256];
	api->getModuleName(myName, 256);
	api->logPrint(2,"[%s] started running...", myName);

	uint64 lastTime = GetTimeNow() - 1000000;
	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(0, triggerName)) {
		}
		else if (inMsg = api->waitForSignal("Input", 100, lastTime)) {
			lastTime = inMsg->getCreatedTime();
			api->emitSignal("Output");
		}
	}
	delete [] myName;
	return 0;
}


// ***************************************
// ************ Retrieve Test ************
// ***************************************

int8 PsyTest_RetrieveTest(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);

	int64 steps = 10;
	int64 cycles;
	if (api->hasParameter("Cycles"))
		if (api->getParameter("Cycles", cycles))
			steps = cycles;

	api->logPrint(2,"[%s] started running, %lld steps...", myName, (long long)steps);

	int64 val;
	DataMessage* inMsg, *outMsg;
	const char* triggerName;
	uint64 start = GetTimeNow();

	std::list<DataMessage*> retrievedMsgs;
	std::list<DataMessage*>::iterator i, e;
	bool done = false;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
				api->logPrint(2,"[%s] got start message type '%s'...", myName, api->typeToText(inMsg->getType()).c_str());
				for (val = 0; val < 10; val++) {
					outMsg = new DataMessage();
					outMsg->setInt("count", val);
					api->postOutputMessage("OutputMsg", outMsg);
				}
				start = GetTimeNow();
			}
		}

		uint8 status;
		if (!done && GetTimeAgeMS(start) > 1000) {
			// Self-create the file-catalog fixture: a clean checkout has no
			// <execdir>/test/test.txt (bin outputs are gitignored), so prime
			// the MyFiles catalog with one write before the read/write loop.
			// This makes the test self-sufficient - no manual fixture setup
			// is needed on a fresh clone.
			{
				uint32 primesize = 0;
				char* primedata = utils::StringFormat(primesize, "Hello1 World");
				char* primeresult = NULL;
				api->queryCatalog(&primeresult, primesize, "MyFiles", "test", "write", primedata, primesize);
				delete [] primedata;
				delete [] primeresult;
			}
			uint64 opStart = GetTimeNow();
			int opsOk = 0, opsFail = 0;
			const char* firstFail = NULL;

			// Helper-free: count an op, remembering the first failure.
			#define PSYTEST_OP(st, name) do { \
				if ((st) == QUERY_SUCCESS) opsOk++; \
				else { opsFail++; if (!firstFail) firstFail = (name); } \
			} while (0)

			for (uint32 nn=0; nn<steps; nn++) {
				status = api->retrieve(retrievedMsgs, "r1");
				PSYTEST_OP(status, "r1");
				for (i = retrievedMsgs.begin(), e = retrievedMsgs.end(); i != e; i++) delete(*i);
				retrievedMsgs.clear();

				status = api->retrieveIntegerParam(retrievedMsgs, "r2", 2, 8, 4);
				PSYTEST_OP(status, "r2");
				for (i = retrievedMsgs.begin(), e = retrievedMsgs.end(); i != e; i++) delete(*i);
				retrievedMsgs.clear();
				utils::Sleep(100);

				status = api->retrieveIntegerParam(retrievedMsgs, "r3", 0, 100, 100);
				PSYTEST_OP(status, "r3");
				for (i = retrievedMsgs.begin(), e = retrievedMsgs.end(); i != e; i++) delete(*i);
				retrievedMsgs.clear();

				uint32 datasize = 0;
				char* data = NULL;
				char* result = NULL;
				status = api->queryCatalog(&result, datasize, "MyFiles", "test", "read");
				PSYTEST_OP(status, "MyFiles:read");
				delete [] result;

				data = utils::StringFormat(datasize, "Hello1 World");
				status = api->queryCatalog(&result, datasize, "MyFiles", "test", "write", data, datasize);
				PSYTEST_OP(status, "MyFiles:write");
				delete [] data;
				delete [] result;

				// First read of a data catalog may legitimately be empty on a fresh
				// run, so this one is NOT counted as a functional failure.
				result = NULL;
				status = api->queryCatalog(&result, datasize, "MyData", "test", "read");
				if (status == QUERY_SUCCESS) opsOk++;
				delete [] result;

				data = utils::StringFormat(datasize, "Hello2 World");
				status = api->queryCatalog(&result, datasize, "MyData", "test", "write", data, datasize);
				PSYTEST_OP(status, "MyData:write");
				delete [] data;
				delete [] result;

				result = NULL;
				status = api->queryCatalog(&result, datasize, "MyData", "test", "read");
				PSYTEST_OP(status, "MyData:read");
				delete [] result;
			}
			#undef PSYTEST_OP

			double ms = (double)(GetTimeNow() - opStart) / 1000.0;
			if (opsFail == 0)
				psytestResult(api, true, "%-13s steps=%lld ops_ok=%d (%.0f ms)",
					myName, (long long)steps, opsOk, ms);
			else
				psytestResult(api, false, "%-13s steps=%lld op '%s' failed (%d failures)",
					myName, (long long)steps, firstFail ? firstFail : "?", opsFail);

			// Three identical RetrieveTest modules run concurrently; only the first
			// contributes a structured summary report (so the summary shows one
			// "Catalogs" row), while all three keep their one-line PASS/FAIL log.
			if (strstr(myName, "1") != NULL) {
				if (opsFail == 0)
					psytestReport(7, true,
						"Whiteboard / file / data catalog retrieval",
						"Retrieves messages from whiteboards (by time and by key)\n"
						"and queries File + Data catalogs; verifies every op succeeds.",
						NULL);
				else {
					char det[256];
					snprintf(det, sizeof(det),
						"A retrieve/query op failed: first failing op was '%s'\n"
						"(%d failures total across the retrieve/query set). A whiteboard\n"
						"retrieve or a File/Data catalog query returned a non-success\n"
						"status — the named op above is the first that did not succeed.",
						firstFail ? firstFail : "?", opsFail);
					psytestReport(7, false,
						"Whiteboard / file / data catalog retrieval",
						"Retrieves messages from whiteboards (by time and by key)\n"
						"and queries File + Data catalogs; verifies every op succeeds.",
						det);
				}
			}

			api->postOutputMessage("Done");
			done = true;
			start = 0;
		}
	}

	delete [] myName;
	return 0;
}


// ***************************************
// ********** Context Switching **********
// ***************************************
//
// Demonstrates dynamic context switching (cf. Examples/contexts.xml) and
// interval (timer) triggers. The master lives in a stable context and, paced
// by a periodic "Tick" interval trigger, alternately activates two mutually
// exclusive responder contexts (Switch.A / Switch.B) and sends a "ping". Only
// the responder whose context is currently active should answer -- so each
// round verifies that activating a context correctly (re)routes messages. A
// wrong or missing responder means context switching is broken.

int8 PsyTest_ContextMaster(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);
	api->logPrint(2,"[%s] started running...", myName);

	DataMessage* inMsg;
	const char* triggerName;

	int64 rounds = 6;
	int64 val;
	if (api->hasParameter("Rounds"))
		if (api->getParameter("Rounds", val))
			rounds = val;

	bool running = false, done = false, awaitingPong = false;
	int  phase = 0;          // 0 = activate target context, 1 = send ping
	int64 round = 0;
	int  correct = 0, wrong = 0, missing = 0;
	uint64 heartbeats = 0;   // counts interval-trigger firings
	char expected = '?';
	uint64 pingSentAt = 0;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {

			if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
				running = true; done = false; awaitingPong = false;
				phase = 0; round = 0; correct = wrong = missing = 0; heartbeats = 0;
				api->logPrint(2,"[%s] driving %lld context switches...", myName, (long long)rounds);
			}
			else if (stricmp(triggerName, "Tick") == 0) {
				// Interval (timer) trigger -- also paces the test.
				heartbeats++;
				if (!running || done)
					continue;
				if (awaitingPong) {
					if (GetTimeAgeMS(pingSentAt) > 1000) { // responder never answered
						missing++; awaitingPong = false; round++; phase = 0;
					}
					continue;
				}
				if (round >= rounds) {
					bool pass = (wrong == 0 && missing == 0 && correct == (int)rounds);
					if (pass)
						psytestResultR(api, 6, true,
							"Dynamic context switching + timer triggers",
							"Master toggles two mutually-exclusive contexts (Switch.A/B),\n"
							"paced by an interval trigger; only the active one should answer.",
							NULL,
							"%-13s switches=%lld routed_ok=%d heartbeats=%llu",
							myName, (long long)rounds, correct, (unsigned long long)heartbeats);
					else
						psytestResultR(api, 6, false,
							"Dynamic context switching + timer triggers",
							"Master toggles two mutually-exclusive contexts (Switch.A/B),\n"
							"paced by an interval trigger; only the active one should answer.",
							"Context routing was wrong: a responder in the INACTIVE context\n"
							"answered (wrong>0) and/or the active responder failed to answer\n"
							"in time (missing>0). Either context activation/deactivation did\n"
							"not take effect, or ping/pong was mis-routed across contexts —\n"
							"see wrong/missing counts below.",
							"%-13s switches=%lld wrong=%d missing=%d (context mis-routing)",
							myName, (long long)rounds, wrong, missing);
					api->postOutputMessage("Done");
					done = true; running = false;
					continue;
				}
				if (phase == 0) {
					// Activate the next responder context (deactivates its sibling).
					expected = (round % 2 == 0) ? 'a' : 'b';
					api->postOutputMessage(expected == 'a' ? "GoA" : "GoB");
					phase = 1;
				}
				else {
					// Context has settled; ping it. Only the active responder answers.
					api->postOutputMessage("Ping");
					pingSentAt = GetTimeNow();
					awaitingPong = true;
					phase = 0;
				}
			}
			else if (!done && ((stricmp(triggerName, "PongA") == 0) || (stricmp(triggerName, "PongB") == 0))) {
				char got = (stricmp(triggerName, "PongA") == 0) ? 'a' : 'b';
				if (got == expected) correct++; else wrong++;
				awaitingPong = false;
				round++;
			}
		}
	}

	delete [] myName;
	return 0;
}


// ***************************************
// ************** Summary ****************
// ***************************************

// ***************************************
// ************** External ***************
// ***************************************
//
// Exercises the T1.4 <executable> auto-start feature end-to-end inside the
// standard psytest run. The spec declares an external space with an
// <executable> that auto-launches the ExternalModules helper process; that
// process connects the space and runs a crank that pongs each ping we send.
// This master pings ext.ping and counts ext.pong replies coming back through
// the external process, proving: auto-launch + space connect + real message
// round-trip across a separate OS process.
//
// Fail-soft: if the external process never connects/replies (e.g. the
// ExternalModules binary is missing or a partial build), a bounded watchdog
// reports FAIL and releases the chain rather than hanging the whole suite.

int8 PsyTest_External(PsyAPI* api) {

	char* myName = new char[256];
	api->getModuleName(myName, 256);
	api->logPrint(2, "[%s] started running...", myName);

	DataMessage* inMsg;
	const char* triggerName;
	int64 count = 20;			// number of ping/pong round trips to verify
	int64 val;
	if (api->hasParameter("Count") && api->getParameter("Count", val))
		count = val;

	// How long to wait for the auto-started external process to connect and
	// start ponging before we give up (covers launch + connect latency).
	int64 connectTimeoutMs = 15000;
	if (api->hasParameter("ConnectTimeout") && api->getParameter("ConnectTimeout", val))
		connectTimeoutMs = val;

	int64 counter = 0;
	bool running = false, done = false;
	uint64 stageStart = 0, lastMsg = 0;

	while (api->shouldContinue()) {
		inMsg = api->waitForNewMessage(100, triggerName);
		if (!inMsg) {
			if (running && !done) {
				// Before the first pong: waiting for the external process to come up.
				if (counter == 0 && GetTimeAgeMS(stageStart) > connectTimeoutMs) {
					psytestResultR(api, 5, false,
						"External process auto-start + round-trip",
						"Node pings ext.ping; a pong crank in an auto-launched\n"
						"external process (separate OS process) replies ext.pong.",
						"The external helper process never connected or replied within\n"
						"the connect timeout. Likely the ExternalModules binary is not\n"
						"present next to the Psyclone executable, the <executable>\n"
						"auto-start failed to launch it, or it could not connect the\n"
						"external space — so zero pongs were received.",
						"%-13s external process did not connect/reply within %lld ms "
						"(ExternalModules binary present next to Psyclone?)",
						myName, (long long)connectTimeoutMs);
					api->postOutputMessage("Done");
					done = true; running = false;
				}
				// After it started replying: a multi-second gap means it stalled/died.
				else if (counter > 0 && GetTimeAgeMS(lastMsg) > 5000) {
					psytestResultR(api, 5, false,
						"External process auto-start + round-trip",
						"Node pings ext.ping; a pong crank in an auto-launched\n"
						"external process (separate OS process) replies ext.pong.",
						"The external process connected and began replying but then\n"
						"stalled: no pong for over 5s before all round trips finished.\n"
						"The helper process likely crashed or hung mid-run — see the\n"
						"got-vs-expected pong counts below.",
						"%-13s stalled: got %lld of %lld pongs",
						myName, (long long)counter, (long long)count);
					api->postOutputMessage("Done");
					done = true; running = false;
				}
			}
			continue;
		}

		if ((stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0)) {
			api->logPrint(2, "[%s] pinging auto-started external module...", myName);
			running = true; done = false; counter = 0;
			stageStart = lastMsg = GetTimeNow();
			api->postOutputMessage("Ping");		// first ping to the external process
		}
		else if (!done && (stricmp(triggerName, "Pong") == 0)) {
			lastMsg = GetTimeNow();
			counter++;
			if (counter >= count) {
				double elapsedUs = (double)(GetTimeNow() - stageStart);
				double thr = (elapsedUs > 0) ? ((double)count * 1000000.0 / elapsedUs) : 0;
				psytestResultR(api, 5, true,
					"External process auto-start + round-trip",
					"Node pings ext.ping; a pong crank in an auto-launched\n"
					"external process (separate OS process) replies ext.pong.",
					NULL,
					"%-13s external auto-start OK: %lld round trips through a separate "
					"process, %.0f msg/s", myName, (long long)count, thr);
				api->postOutputMessage("Done");
				done = true; running = false;
			}
			else
				api->postOutputMessage("Ping");	// bounce the next ping
		}
	}

	delete [] myName;
	return 0;
}

// ***************************************
// ********** Trigger Group Test *********
// ***************************************
//
// Exercises the T1.6 declarative <triggergroup> join over the real node
// transport, single-node AND multi-node. TGMaster (on Main) kicks the round by
// posting a broadcast "tggroup.kick"; each TGMember slave (one per node, incl.
// Main) replies with its own member type "tggroup.mN". TGMaster hosts a
// <triggergroup> joining m1..mN and, once the full set joins, verifies the
// member count == node count and advances the test chain. In multi-node runs
// the member messages originate on different nodes, so this is the cross-node
// deliverTriggerGroup path that the by-name unit tests (single node / external
// space) do not cover.

int8 PsyTest_TGMaster(PsyAPI* api) {
	char* myName = new char[256];
	api->getModuleName(myName, 256);
	api->logPrint(2, "[%s] started running...", myName);

	const char* triggerName;
	std::string groupName;
	int64 nodes = 1;
	int64 val;
	if (api->hasParameter("Nodes") && api->getParameter("Nodes", val))
		nodes = val;

	uint64 kickTime = 0;
	bool   running = false, reported = false;

	while (api->shouldContinue()) {
		// Single wait point: waitForNewMessageGroup() returns the joined member map
		// for a CTRL_TRIGGER_GROUP activation, and hands back a single message (with
		// its trigger name) for an ordinary trigger like Start. Using one wait avoids
		// racing two dequeue calls on the same input queue.
		std::map<std::string, DataMessage*> members = api->waitForNewMessageGroup(100, groupName);

		if (!members.empty()) {
			// A completed <triggergroup> activation. Report + advance the chain here,
			// while still in the group-trigger context (postOutputMessage needs one).
			int got = (int)members.size();
			api->logPrint(2, "[%s] join '%s' completed with %d members", myName, groupName.c_str(), got);
			if (running && !reported) {
				if (got == (int)nodes)
					psytestResultR(api, 8, true,
						"Declarative <triggergroup> join (T1.6)",
						"A <triggergroup> joins N member triggers; the crank fires\n"
						"only once the full member set has joined (grouped delivery).",
						NULL,
						"%-13s join completed members=%d (nodes=%lld)",
						myName, got, (long long)nodes);
				else
					psytestResultR(api, 8, false,
						"Declarative <triggergroup> join (T1.6)",
						"A <triggergroup> joins N member triggers; the crank fires\n"
						"only once the full member set has joined (grouped delivery).",
						"The group activated with the wrong number of members: the\n"
						"<triggergroup> fired before all expected members had joined,\n"
						"or a member trigger was missed — got vs expected member counts\n"
						"are below (grouped-activation join did not gate correctly).",
						"%-13s join completed with %d members (expected %lld)",
						myName, got, (long long)nodes);
				reported = true;
				api->postOutputMessage("Done");
			}
		}
		else if (!groupName.empty()) {
			// An ordinary (non-group) trigger arrived: the Start kick-off.
			if ( (stricmp(groupName.c_str(), "Ready") == 0) || (stricmp(groupName.c_str(), "Start") == 0) ) {
				api->logPrint(2, "[%s] kicking triggergroup round (expecting %lld members)...", myName, (long long)nodes);
				running = true; reported = false;
				kickTime = GetTimeNow();
				api->postOutputMessage("Kick");   // broadcast tggroup.kick to all TGMembers
			}
		}
		// Note: (void)triggerName kept for symmetry with other masters.
		(void)triggerName;
	}
	delete [] myName;
	return 0;
}

int8 PsyTest_TGMember(PsyAPI* api) {
	// On each "kick" broadcast, post this member's own type (tggroup.mN) which is
	// wired via the module's single <post> entry. Runs on every node.
	DataMessage* inMsg;
	const char* triggerName;
	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			if (stricmp(triggerName, "Kick") == 0)
				api->postOutputMessage("Member");
		}
	}
	return 0;
}

int8 PsyTest_Summary(PsyAPI* api) {
	DataMessage* inMsg;
	const char* triggerName;

	while (api->shouldContinue()) {
		if ( inMsg = api->waitForNewMessage(100, triggerName)) {
			if ( (stricmp(triggerName, "Ready") == 0) || (stricmp(triggerName, "Start") == 0) ) {
				bool doIt = false;
				g_psytestMutex.enter(1000, __FUNCTION__);
				if (!g_psytestSummaryDone) { g_psytestSummaryDone = true; doIt = true; }
				g_psytestMutex.leave();

				if (doIt) {
					// The Catalog stages run concurrently; give any stragglers a
					// moment to finish recording before we snapshot the totals.
					utils::Sleep(1500);

					g_psytestMutex.enter(1000, __FUNCTION__);
					int pass = g_psytestPass, fail = g_psytestFail;
					std::vector<std::string> lines = g_psytestLines;
					std::vector<PsyTestReport> reports = g_psytestReports;
					g_psytestMutex.leave();

					// Present stages in chain order regardless of arrival order.
					std::sort(reports.begin(), reports.end(),
						[](const PsyTestReport& a, const PsyTestReport& b) { return a.order < b.order; });

					const int WIDTH = 60;   // right-align the status tag to this column
					const char* INDENT = "    ";

					api->logPrint(0, "============================================================");
					if (reports.empty()) {
						// Fallback: no structured reports arrived — use the old line table.
						if (fail == 0)
							api->logPrint(0, " PSYTEST RESULT: SUCCESS  (%d stages passed, 0 failed)", pass);
						else {
							api->logPrint(0, " PSYTEST RESULT: FAILED  (%d passed, %d failed)", pass, fail);
							for (size_t k = 0; k < lines.size(); k++)
								if (lines[k].find("[ FAIL ]") != std::string::npos)
									api->logPrint(0, "   %s", lines[k].c_str());
						}
						api->logPrint(0, "============================================================");
					}
					else {
						// Header: total tests + PASS/FAIL breakdown.
						int rpass = 0, rfail = 0;
						for (size_t k = 0; k < reports.size(); k++)
							if (reports[k].pass) rpass++; else rfail++;
						int total = (int)reports.size();
						if (rfail == 0)
							api->logPrint(0, " PSYTEST RESULT: SUCCESS — %d tests, %d PASS, 0 FAIL", total, rpass);
						else
							api->logPrint(0, " PSYTEST RESULT: FAILED — %d tests, %d PASS, %d FAIL", total, rpass, rfail);
						api->logPrint(0, "============================================================");

						// Per-test block: headline with right-aligned status, indented
						// description, and (on FAIL) an indented failure report, then a
						// blank-line spacer.
						for (size_t k = 0; k < reports.size(); k++) {
							const PsyTestReport& r = reports[k];
							const char* tag = r.pass ? "[ PASS ]" : "[ FAIL ]";

							// Right-align the status tag to WIDTH columns. If the headline
							// is too long to leave a gap, drop the tag onto its own line.
							int tagLen  = (int)strlen(tag);
							int headLen = (int)r.headline.size();
							if (headLen + 1 + tagLen <= WIDTH) {
								int pad = WIDTH - headLen - tagLen;
								std::string spaces(pad > 0 ? pad : 1, ' ');
								api->logPrint(0, "%s%s%s", r.headline.c_str(), spaces.c_str(), tag);
							}
							else {
								api->logPrint(0, "%s", r.headline.c_str());
								std::string spaces(WIDTH - tagLen > 0 ? WIDTH - tagLen : 0, ' ');
								api->logPrint(0, "%s%s", spaces.c_str(), tag);
							}

							// Indented description (may be multi-line; split on '\n').
							std::vector<std::string> dl = utils::TextListSplitLines(r.description.c_str(), false);
							for (size_t j = 0; j < dl.size(); j++)
								api->logPrint(0, "%s%s", INDENT, dl[j].c_str());

							// Indented failure report (FAIL only).
							if (!r.pass && !r.detail.empty()) {
								std::vector<std::string> fl = utils::TextListSplitLines(r.detail.c_str(), false);
								for (size_t j = 0; j < fl.size(); j++)
									api->logPrint(0, "%sFAILURE: %s", INDENT, fl[j].c_str());
							}

							api->logPrint(0, " ");   // blank-line spacer between tests
						}
						api->logPrint(0, "============================================================");
					}

					// logPrint is async; let the logging thread flush the banner before
					// we trigger shutdown (which tears that thread down).
					utils::Sleep(300);
					api->postOutputMessage("Done");	// -> Psyclone.Shutdown
				}
			}
		}
	}

	return 0;
}

} // namespace cmlabs
