Interprocess pipe support (for stdout/stderr capture) (#866)

- **RAII pipe handles for child process stdout/stderr capture**: `StdoutPipeHandles` is now a proper RAII type with automatic cleanup, move semantics, and partial close support. This makes it safe to use pipes for capturing child process output without risking handle/fd leaks.

- **Optional separate stderr pipe**: `CreateProcOptions` now accepts a `StderrPipe` field so callers can capture stdout and stderr independently. When null (default), stderr shares the stdout pipe as before.

- **LogStreamListener with pluggable handler**: The TCP log stream listener accepts connections from remote processes and delivers parsed log lines through a `LogStreamHandler` interface, set dynamically via `SetHandler()`. This allows any client to receive log messages without depending on a specific console implementation.

- **TcpLogStreamSink for zen::logging**: A logging sink that forwards log messages to a `LogStreamListener` over TCP, using the native `zen::logging::Sink` infrastructure with proper thread-safe synchronization.

- **Reliable child process exit codes on Linux**: `waitpid` result handling is fixed so `ProcessHandle::GetExitCode()` returns the real exit code. `ProcessHandle::Reset()` reaps zombies directly, replacing the global `IgnoreChildSignals()` which prevented exit code collection entirely. Also fixes a TOCTOU race in `ProcessHandle::Wait()` on Linux/Mac.

- **Pipe capture test suite**: Tests covering stdout/stderr capture via pipes (both shared and separate modes), RAII cleanup, move semantics, and exit code propagation using `zentest-appstub` as the child process.

- **Service command integration tests**: Shell-based integration tests for `zen service` covering the full lifecycle (install, status, start, stop, uninstall) on all three platforms — Linux (systemd), macOS (launchd), and Windows (SCM via PowerShell).

- **Test script reorganization**: Platform-specific test scripts moved from `scripts/test_scripts/` into `scripts/test_linux/`, `test_mac/`, and `test_windows/`.

15 files changed, 1154 insertions, 25 deletions

diff --git a/src/zencore/include/zencore/logging/sink.h b/src/zencore/include/zencore/logging/sink.h
index 172176a4e..3e6a1deed 100644
--- a/src/zencore/include/zencore/logging/sink.h
+++ b/src/zencore/include/zencore/logging/sink.h
@@ -11,6 +11,11 @@
 
 namespace zen::logging {
 
+/// Base class for log sinks.
+///
+/// Log() and Flush() may be called concurrently from multiple threads.
+/// Implementations must provide their own synchronization (e.g. a mutex
+/// or RwLock) to protect any mutable state including the formatter.
 class Sink : public RefCounted
 {
 public:
diff --git a/src/zencore/include/zencore/process.h b/src/zencore/include/zencore/process.h
index 96afd5950..75fd7b25a 100644
--- a/src/zencore/include/zencore/process.h
+++ b/src/zencore/include/zencore/process.h
@@ -54,6 +54,38 @@ private:
 /** Basic process creation
  */
 
+// Platform-agnostic RAII pipe handles for capturing child stdout/stderr.
+// The destructor closes any open handles/fds automatically.
+struct StdoutPipeHandles
+{
+	StdoutPipeHandles() = default;
+	~StdoutPipeHandles();
+
+	StdoutPipeHandles(const StdoutPipeHandles&) = delete;
+	StdoutPipeHandles& operator=(const StdoutPipeHandles&) = delete;
+
+	StdoutPipeHandles(StdoutPipeHandles&& Other) noexcept;
+	StdoutPipeHandles& operator=(StdoutPipeHandles&& Other) noexcept;
+
+	// Close only the write end (call after child is launched so parent doesn't hold it open).
+	void CloseWriteEnd();
+
+	// Close both ends of the pipe.
+	void Close();
+
+#if ZEN_PLATFORM_WINDOWS
+	void* ReadHandle  = nullptr;  // HANDLE for reading (parent side)
+	void* WriteHandle = nullptr;  // HANDLE for writing (child side)
+#else
+	int ReadFd	= -1;
+	int WriteFd = -1;
+#endif
+};
+
+// Create a pipe suitable for capturing child process stdout.
+// The write end is inheritable; the read end is not.
+bool CreateStdoutPipe(StdoutPipeHandles& OutPipe);
+
 struct CreateProcOptions
 {
 	enum
@@ -71,6 +103,8 @@ struct CreateProcOptions
 	const std::filesystem::path* WorkingDirectory = nullptr;
 	uint32_t					 Flags			  = 0;
 	std::filesystem::path		 StdoutFile;
+	StdoutPipeHandles*			 StdoutPipe = nullptr;	// Mutually exclusive with StdoutFile. Parent reads from ReadHandle after launch.
+	StdoutPipeHandles*			 StderrPipe = nullptr;	// Optional separate pipe for stderr. When null, stderr shares StdoutPipe.
 
 	/// Additional environment variables for the child process. These are merged
 	/// with the parent's environment — existing variables are inherited, and
diff --git a/src/zencore/process.cpp b/src/zencore/process.cpp
index 29de107bd..8a91ab287 100644
--- a/src/zencore/process.cpp
+++ b/src/zencore/process.cpp
@@ -137,6 +137,144 @@ IsZombieProcess(int pid, std::error_code& OutEc)
 }
 #endif	// ZEN_PLATFORM_MAC
 
+//////////////////////////////////////////////////////////////////////////
+// Pipe creation for child process stdout capture
+
+#if ZEN_PLATFORM_WINDOWS
+
+StdoutPipeHandles::~StdoutPipeHandles()
+{
+	Close();
+}
+
+StdoutPipeHandles::StdoutPipeHandles(StdoutPipeHandles&& Other) noexcept
+: ReadHandle(std::exchange(Other.ReadHandle, nullptr))
+, WriteHandle(std::exchange(Other.WriteHandle, nullptr))
+{
+}
+
+StdoutPipeHandles&
+StdoutPipeHandles::operator=(StdoutPipeHandles&& Other) noexcept
+{
+	if (this != &Other)
+	{
+		Close();
+		ReadHandle	= std::exchange(Other.ReadHandle, nullptr);
+		WriteHandle = std::exchange(Other.WriteHandle, nullptr);
+	}
+	return *this;
+}
+
+void
+StdoutPipeHandles::CloseWriteEnd()
+{
+	if (WriteHandle)
+	{
+		CloseHandle(WriteHandle);
+		WriteHandle = nullptr;
+	}
+}
+
+void
+StdoutPipeHandles::Close()
+{
+	if (ReadHandle)
+	{
+		CloseHandle(ReadHandle);
+		ReadHandle = nullptr;
+	}
+	CloseWriteEnd();
+}
+
+bool
+CreateStdoutPipe(StdoutPipeHandles& OutPipe)
+{
+	SECURITY_ATTRIBUTES Sa;
+	Sa.nLength				= sizeof(Sa);
+	Sa.lpSecurityDescriptor = nullptr;
+	Sa.bInheritHandle		= TRUE;
+
+	HANDLE ReadHandle  = nullptr;
+	HANDLE WriteHandle = nullptr;
+	if (!::CreatePipe(&ReadHandle, &WriteHandle, &Sa, 0))
+	{
+		return false;
+	}
+
+	// The read end should not be inherited by the child
+	SetHandleInformation(ReadHandle, HANDLE_FLAG_INHERIT, 0);
+
+	OutPipe.ReadHandle	= ReadHandle;
+	OutPipe.WriteHandle = WriteHandle;
+	return true;
+}
+
+#else
+
+StdoutPipeHandles::~StdoutPipeHandles()
+{
+	Close();
+}
+
+StdoutPipeHandles::StdoutPipeHandles(StdoutPipeHandles&& Other) noexcept
+: ReadFd(std::exchange(Other.ReadFd, -1))
+, WriteFd(std::exchange(Other.WriteFd, -1))
+{
+}
+
+StdoutPipeHandles&
+StdoutPipeHandles::operator=(StdoutPipeHandles&& Other) noexcept
+{
+	if (this != &Other)
+	{
+		Close();
+		ReadFd	= std::exchange(Other.ReadFd, -1);
+		WriteFd = std::exchange(Other.WriteFd, -1);
+	}
+	return *this;
+}
+
+void
+StdoutPipeHandles::CloseWriteEnd()
+{
+	if (WriteFd >= 0)
+	{
+		close(WriteFd);
+		WriteFd = -1;
+	}
+}
+
+void
+StdoutPipeHandles::Close()
+{
+	if (ReadFd >= 0)
+	{
+		close(ReadFd);
+		ReadFd = -1;
+	}
+	CloseWriteEnd();
+}
+
+bool
+CreateStdoutPipe(StdoutPipeHandles& OutPipe)
+{
+	int Fds[2];
+	if (pipe(Fds) != 0)
+	{
+		return false;
+	}
+	OutPipe.ReadFd	= Fds[0];
+	OutPipe.WriteFd = Fds[1];
+
+	// Set close-on-exec on the read end so the child doesn't inherit it
+	fcntl(OutPipe.ReadFd, F_SETFD, FD_CLOEXEC);
+	return true;
+}
+
+#endif
+
+//////////////////////////////////////////////////////////////////////////
+
 ProcessHandle::ProcessHandle() = default;
 
 #if ZEN_PLATFORM_WINDOWS
@@ -309,6 +447,10 @@ ProcessHandle::Reset()
 	{
 #if ZEN_PLATFORM_WINDOWS
 		CloseHandle(m_ProcessHandle);
+#elif ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
+		// Reap the child if it has already exited to prevent zombies.
+		// If still running, it will be reparented to init on our exit.
+		waitpid(m_Pid, nullptr, WNOHANG);
 #endif
 		m_ProcessHandle = nullptr;
 		m_Pid			= 0;
@@ -350,17 +492,26 @@ ProcessHandle::Wait(int TimeoutMs, std::error_code& OutEc)
 	timespec  SleepTime = {0, SleepMs * 1000 * 1000};
 	for (int SleepedTimeMS = 0;; SleepedTimeMS += SleepMs)
 	{
-		int WaitState = 0;
-		if (waitpid(m_Pid, &WaitState, WNOHANG | WCONTINUED | WUNTRACED) != -1)
+		int	  WaitState	 = 0;
+		pid_t WaitResult = waitpid(m_Pid, &WaitState, WNOHANG | WCONTINUED | WUNTRACED);
+		if (WaitResult > 0 && WIFEXITED(WaitState))
 		{
-			if (WIFEXITED(WaitState))
-			{
-				m_ExitCode = WEXITSTATUS(WaitState);
-			}
+			m_ExitCode = WEXITSTATUS(WaitState);
 		}
 
 		if (!IsProcessRunning(m_Pid, OutEc))
 		{
+			// Process is gone but waitpid(WNOHANG) may have missed the exit status
+			// due to a TOCTOU race (process became a zombie between waitpid and
+			// IsProcessRunning). Do a blocking reap now to capture the exit code.
+			if (WaitResult <= 0)
+			{
+				WaitState = 0;
+				if (waitpid(m_Pid, &WaitState, 0) > 0 && WIFEXITED(WaitState))
+				{
+					m_ExitCode = WEXITSTATUS(WaitState);
+				}
+			}
 			return true;
 		}
 		else if (OutEc)
@@ -381,6 +532,12 @@ ProcessHandle::Wait(int TimeoutMs, std::error_code& OutEc)
 		else if (IsZombieProcess(m_Pid, OutEc))
 		{
 			ZEN_INFO("Found process {} in zombie state, treating as not running", m_Pid);
+			// Reap the zombie to capture its exit code.
+			WaitState = 0;
+			if (waitpid(m_Pid, &WaitState, 0) > 0 && WIFEXITED(WaitState))
+			{
+				m_ExitCode = WEXITSTATUS(WaitState);
+			}
 			return true;
 		}
 
@@ -567,7 +724,40 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 	ExtendableWideStringBuilder<256> CommandLineZ;
 	CommandLineZ << CommandLine;
 
-	if (!Options.StdoutFile.empty())
+	bool DuplicatedStdErr = false;
+
+	if (Options.StdoutPipe != nullptr && Options.StdoutPipe->WriteHandle != nullptr)
+	{
+		StartupInfo.hStdInput  = nullptr;
+		StartupInfo.hStdOutput = (HANDLE)Options.StdoutPipe->WriteHandle;
+
+		if (Options.StderrPipe != nullptr && Options.StderrPipe->WriteHandle != nullptr)
+		{
+			// Use separate pipe for stderr
+			StartupInfo.hStdError = (HANDLE)Options.StderrPipe->WriteHandle;
+			StartupInfo.dwFlags |= STARTF_USESTDHANDLES;
+			InheritHandles = true;
+		}
+		else
+		{
+			// Duplicate stdout handle for stderr (both go to same pipe)
+			const BOOL DupSuccess = DuplicateHandle(GetCurrentProcess(),
+													StartupInfo.hStdOutput,
+													GetCurrentProcess(),
+													&StartupInfo.hStdError,
+													0,
+													TRUE,
+													DUPLICATE_SAME_ACCESS);
+
+			if (DupSuccess)
+			{
+				DuplicatedStdErr = true;
+				StartupInfo.dwFlags |= STARTF_USESTDHANDLES;
+				InheritHandles = true;
+			}
+		}
+	}
+	else if (!Options.StdoutFile.empty())
 	{
 		SECURITY_ATTRIBUTES sa;
 		sa.nLength				= sizeof sa;
@@ -593,6 +783,7 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 
 		if (Success)
 		{
+			DuplicatedStdErr = true;
 			StartupInfo.dwFlags |= STARTF_USESTDHANDLES;
 			InheritHandles = true;
 		}
@@ -616,8 +807,16 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 
 	if (StartupInfo.dwFlags & STARTF_USESTDHANDLES)
 	{
-		CloseHandle(StartupInfo.hStdError);
-		CloseHandle(StartupInfo.hStdOutput);
+		// Only close hStdError if we duplicated it (caller-owned pipe handles are not ours to close)
+		if (DuplicatedStdErr)
+		{
+			CloseHandle(StartupInfo.hStdError);
+		}
+		// Only close hStdOutput if it was a file handle we created (not a pipe handle owned by caller)
+		if (Options.StdoutPipe == nullptr || Options.StdoutPipe->WriteHandle == nullptr)
+		{
+			CloseHandle(StartupInfo.hStdOutput);
+		}
 	}
 
 	if (!Success)
@@ -826,7 +1025,25 @@ CreateProc(const std::filesystem::path& Executable, std::string_view CommandLine
 			ZEN_UNUSED(Result);
 		}
 
-		if (!Options.StdoutFile.empty())
+		if (Options.StdoutPipe != nullptr && Options.StdoutPipe->WriteFd >= 0)
+		{
+			dup2(Options.StdoutPipe->WriteFd, STDOUT_FILENO);
+
+			if (Options.StderrPipe != nullptr && Options.StderrPipe->WriteFd >= 0)
+			{
+				dup2(Options.StderrPipe->WriteFd, STDERR_FILENO);
+				close(Options.StderrPipe->WriteFd);
+				// StderrPipe ReadFd has FD_CLOEXEC so it's auto-closed on exec
+			}
+			else
+			{
+				dup2(Options.StdoutPipe->WriteFd, STDERR_FILENO);
+			}
+
+			close(Options.StdoutPipe->WriteFd);
+			// ReadFd has FD_CLOEXEC so it's auto-closed on exec
+		}
+		else if (!Options.StdoutFile.empty())
 		{
 			int Fd = open(Options.StdoutFile.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644);
 			if (Fd >= 0)
diff --git a/src/zencore/testing.cpp b/src/zencore/testing.cpp
index c6ee5ee6b..9f88a3365 100644
--- a/src/zencore/testing.cpp
+++ b/src/zencore/testing.cpp
@@ -309,10 +309,6 @@ RunTestMain(int Argc, char* Argv[], const char* ExecutableName, void (*ForceLink
 
 	ForceLink();
 
-#	if ZEN_PLATFORM_LINUX
-	zen::IgnoreChildSignals();
-#	endif
-
 #	if ZEN_WITH_TRACE
 	zen::TraceInit(ExecutableName);
 	zen::TraceOptions TraceCommandlineOptions;
diff --git a/src/zenserver-test/process-tests.cpp b/src/zenserver-test/process-tests.cpp
new file mode 100644
index 000000000..649f24f54
--- /dev/null
+++ b/src/zenserver-test/process-tests.cpp
@@ -0,0 +1,298 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zencore/zencore.h>
+
+#if ZEN_WITH_TESTS
+
+#	include "zenserver-test.h"
+
+#	include <zencore/filesystem.h>
+#	include <zencore/process.h>
+#	include <zencore/testing.h>
+
+#	if ZEN_PLATFORM_WINDOWS
+#		include <zencore/windows.h>
+#	else
+#		include <unistd.h>
+#	endif
+
+namespace zen::tests {
+
+using namespace std::literals;
+
+static std::filesystem::path
+GetAppStubPath()
+{
+	return TestEnv.ProgramBaseDir() / ("zentest-appstub" ZEN_EXE_SUFFIX_LITERAL);
+}
+
+// Read all available data from the read end of a StdoutPipeHandles.
+// Must be called after CloseWriteEnd() so that the read will see EOF.
+static std::string
+ReadAllFromPipe(StdoutPipeHandles& Pipe)
+{
+	std::string Result;
+	char		Buffer[4096];
+
+#	if ZEN_PLATFORM_WINDOWS
+	DWORD BytesRead = 0;
+	while (::ReadFile(Pipe.ReadHandle, Buffer, sizeof(Buffer), &BytesRead, nullptr) && BytesRead > 0)
+	{
+		Result.append(Buffer, BytesRead);
+	}
+#	else
+	ssize_t BytesRead = 0;
+	while ((BytesRead = read(Pipe.ReadFd, Buffer, sizeof(Buffer))) > 0)
+	{
+		Result.append(Buffer, static_cast<size_t>(BytesRead));
+	}
+#	endif
+
+	return Result;
+}
+
+TEST_SUITE_BEGIN("server.process");
+
+//////////////////////////////////////////////////////////////////////////
+
+TEST_CASE("pipe.capture_stdout")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	const std::string	  ExpectedOutput = "hello_from_pipe_test";
+	std::filesystem::path AppStub		 = GetAppStubPath();
+	std::string			  CommandLine	 = fmt::format("zentest-appstub -echo={}", ExpectedOutput);
+
+	CreateProcOptions Options;
+	Options.StdoutPipe = &Pipe;
+
+	CreateProcResult ProcResult = CreateProc(AppStub, CommandLine, Options);
+
+	ProcessHandle Process;
+	Process.Initialize(ProcResult);
+
+	// Close the write end, then drain before Wait() to avoid deadlock if output fills the pipe buffer.
+	Pipe.CloseWriteEnd();
+
+	std::string Output = ReadAllFromPipe(Pipe);
+
+	Process.Wait();
+
+	// The appstub also prints "[zentest] exiting with exit code: 0\n"
+	CHECK(Output.find(ExpectedOutput) != std::string::npos);
+	CHECK_EQ(Process.GetExitCode(), 0);
+}
+
+TEST_CASE("pipe.capture_multiline")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	std::filesystem::path AppStub	  = GetAppStubPath();
+	std::string			  CommandLine = "zentest-appstub -echo=line1 -echo=line2 -echo=line3";
+
+	CreateProcOptions Options;
+	Options.StdoutPipe = &Pipe;
+
+	CreateProcResult ProcResult = CreateProc(AppStub, CommandLine, Options);
+
+	ProcessHandle Process;
+	Process.Initialize(ProcResult);
+
+	Pipe.CloseWriteEnd();
+
+	std::string Output = ReadAllFromPipe(Pipe);
+
+	Process.Wait();
+
+	CHECK(Output.find("line1") != std::string::npos);
+	CHECK(Output.find("line2") != std::string::npos);
+	CHECK(Output.find("line3") != std::string::npos);
+	CHECK_EQ(Process.GetExitCode(), 0);
+}
+
+TEST_CASE("pipe.raii_cleanup")
+{
+	// Verify that StdoutPipeHandles cleans up handles when it goes out of scope
+	// (no leaked handles). We can't directly assert on handle counts, but we can
+	// verify that creating and destroying many pipes doesn't fail.
+	for (int i = 0; i < 100; ++i)
+	{
+		StdoutPipeHandles Pipe;
+		REQUIRE(CreateStdoutPipe(Pipe));
+		// Pipe goes out of scope here — destructor should close both ends
+	}
+}
+
+TEST_CASE("pipe.move_semantics")
+{
+	StdoutPipeHandles Original;
+	REQUIRE(CreateStdoutPipe(Original));
+
+	// Move-construct a new pipe from Original
+	StdoutPipeHandles Moved(std::move(Original));
+
+#	if ZEN_PLATFORM_WINDOWS
+	CHECK(Moved.ReadHandle != nullptr);
+	CHECK(Moved.WriteHandle != nullptr);
+	CHECK(Original.ReadHandle == nullptr);
+	CHECK(Original.WriteHandle == nullptr);
+#	else
+	CHECK(Moved.ReadFd >= 0);
+	CHECK(Moved.WriteFd >= 0);
+	CHECK(Original.ReadFd == -1);
+	CHECK(Original.WriteFd == -1);
+#	endif
+
+	// Move-assign
+	StdoutPipeHandles Assigned;
+	Assigned = std::move(Moved);
+
+#	if ZEN_PLATFORM_WINDOWS
+	CHECK(Assigned.ReadHandle != nullptr);
+	CHECK(Assigned.WriteHandle != nullptr);
+	CHECK(Moved.ReadHandle == nullptr);
+	CHECK(Moved.WriteHandle == nullptr);
+#	else
+	CHECK(Assigned.ReadFd >= 0);
+	CHECK(Assigned.WriteFd >= 0);
+	CHECK(Moved.ReadFd == -1);
+	CHECK(Moved.WriteFd == -1);
+#	endif
+
+	// Assigned goes out of scope — destructor closes handles
+}
+
+TEST_CASE("pipe.close_is_idempotent")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	Pipe.Close();
+	// Calling Close again should be safe (no double-close)
+	Pipe.Close();
+
+#	if ZEN_PLATFORM_WINDOWS
+	CHECK(Pipe.ReadHandle == nullptr);
+	CHECK(Pipe.WriteHandle == nullptr);
+#	else
+	CHECK(Pipe.ReadFd == -1);
+	CHECK(Pipe.WriteFd == -1);
+#	endif
+}
+
+TEST_CASE("pipe.close_write_end_only")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	Pipe.CloseWriteEnd();
+
+#	if ZEN_PLATFORM_WINDOWS
+	CHECK(Pipe.ReadHandle != nullptr);
+	CHECK(Pipe.WriteHandle == nullptr);
+#	else
+	CHECK(Pipe.ReadFd >= 0);
+	CHECK(Pipe.WriteFd == -1);
+#	endif
+
+	// Remaining read handle cleaned up by destructor
+}
+
+TEST_CASE("pipe.capture_with_nonzero_exit")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	std::filesystem::path AppStub	  = GetAppStubPath();
+	std::string			  CommandLine = "zentest-appstub -echo=before_exit -f=42";
+
+	CreateProcOptions Options;
+	Options.StdoutPipe = &Pipe;
+
+	CreateProcResult ProcResult = CreateProc(AppStub, CommandLine, Options);
+
+	ProcessHandle Process;
+	Process.Initialize(ProcResult);
+
+	Pipe.CloseWriteEnd();
+
+	std::string Output = ReadAllFromPipe(Pipe);
+
+	Process.Wait();
+
+	CHECK(Output.find("before_exit") != std::string::npos);
+	CHECK_EQ(Process.GetExitCode(), 42);
+}
+
+TEST_CASE("pipe.stderr_on_shared_pipe")
+{
+	StdoutPipeHandles Pipe;
+	REQUIRE(CreateStdoutPipe(Pipe));
+
+	std::filesystem::path AppStub	  = GetAppStubPath();
+	std::string			  CommandLine = "zentest-appstub -echo=from_stdout -echoerr=from_stderr";
+
+	CreateProcOptions Options;
+	Options.StdoutPipe = &Pipe;
+
+	CreateProcResult ProcResult = CreateProc(AppStub, CommandLine, Options);
+
+	ProcessHandle Process;
+	Process.Initialize(ProcResult);
+
+	Pipe.CloseWriteEnd();
+
+	std::string Output = ReadAllFromPipe(Pipe);
+
+	Process.Wait();
+
+	// Both stdout and stderr content should appear on the shared pipe
+	CHECK(Output.find("from_stdout") != std::string::npos);
+	CHECK(Output.find("from_stderr") != std::string::npos);
+	CHECK_EQ(Process.GetExitCode(), 0);
+}
+
+TEST_CASE("pipe.separate_stderr")
+{
+	StdoutPipeHandles StdoutPipe;
+	StdoutPipeHandles StderrPipe;
+	REQUIRE(CreateStdoutPipe(StdoutPipe));
+	REQUIRE(CreateStdoutPipe(StderrPipe));
+
+	std::filesystem::path AppStub	  = GetAppStubPath();
+	std::string			  CommandLine = "zentest-appstub -echo=on_stdout -echoerr=on_stderr";
+
+	CreateProcOptions Options;
+	Options.StdoutPipe = &StdoutPipe;
+	Options.StderrPipe = &StderrPipe;
+
+	CreateProcResult ProcResult = CreateProc(AppStub, CommandLine, Options);
+
+	ProcessHandle Process;
+	Process.Initialize(ProcResult);
+
+	StdoutPipe.CloseWriteEnd();
+	StderrPipe.CloseWriteEnd();
+
+	std::string StdoutOutput = ReadAllFromPipe(StdoutPipe);
+	std::string StderrOutput = ReadAllFromPipe(StderrPipe);
+
+	Process.Wait();
+
+	CHECK(StdoutOutput.find("on_stdout") != std::string::npos);
+	CHECK(StderrOutput.find("on_stderr") != std::string::npos);
+	// Verify separation: stderr content should NOT appear in stdout pipe
+	CHECK(StdoutOutput.find("on_stderr") == std::string::npos);
+	CHECK(StderrOutput.find("on_stdout") == std::string::npos);
+	CHECK_EQ(Process.GetExitCode(), 0);
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+TEST_SUITE_END();
+
+}  // namespace zen::tests
+
+#endif
diff --git a/src/zenserver-test/zenserver-test.cpp b/src/zenserver-test/zenserver-test.cpp
index 42632682b..fff77957d 100644
--- a/src/zenserver-test/zenserver-test.cpp
+++ b/src/zenserver-test/zenserver-test.cpp
@@ -76,10 +76,6 @@ main(int argc, char** argv)
 
 	zen::CommandLineConverter ArgConverter(argc, argv);
 
-#	if ZEN_PLATFORM_LINUX
-	IgnoreChildSignals();
-#	endif
-
 #	if ZEN_WITH_TRACE
 	zen::TraceInit("zenserver-test");
 	TraceOptions TraceCommandlineOptions;
diff --git a/src/zenserver/diag/logging.cpp b/src/zenserver/diag/logging.cpp
index 178c3d3b5..38b15480a 100644
--- a/src/zenserver/diag/logging.cpp
+++ b/src/zenserver/diag/logging.cpp
@@ -13,6 +13,7 @@
 #include <zencore/string.h>
 #include <zenutil/logging.h>
 #include <zenutil/logging/rotatingfilesink.h>
+#include <zenutil/splitconsole/tcplogstreamsink.h>
 
 #include "otlphttp.h"
 
@@ -28,6 +29,7 @@ InitializeServerLogging(const ZenServerConfig& InOptions, bool WithCacheService)
 									   .IsTest			= InOptions.IsTest,
 									   .NoConsoleOutput = InOptions.LoggingConfig.NoConsoleOutput,
 									   .QuietConsole	= InOptions.LoggingConfig.QuietConsole,
+									   .ForceColor		= InOptions.LoggingConfig.ForceColor,
 									   .AbsLogFile		= InOptions.LoggingConfig.AbsLogFile,
 									   .LogId			= InOptions.LoggingConfig.LogId};
 
@@ -81,6 +83,27 @@ InitializeServerLogging(const ZenServerConfig& InOptions, bool WithCacheService)
 	}
 #endif
 
+	if (!InOptions.LoggingConfig.LogStreamEndpoint.empty())
+	{
+		std::string Endpoint = InOptions.LoggingConfig.LogStreamEndpoint;
+		std::string Host	 = "localhost";
+		uint16_t	Port	 = 0;
+
+		auto ColonPos = Endpoint.rfind(':');
+		if (ColonPos != std::string::npos)
+		{
+			Host					  = Endpoint.substr(0, ColonPos);
+			std::optional<uint16_t> P = ParseInt<uint16_t>(std::string_view(Endpoint).substr(ColonPos + 1));
+			Port					  = P.value_or(0);
+		}
+
+		if (Port > 0)
+		{
+			logging::SinkPtr StreamSink(new TcpLogStreamSink(Host, Port, "zenserver"));
+			zen::logging::Default()->AddSink(std::move(StreamSink));
+		}
+	}
+
 	FinishInitializeLogging(LogOptions);
 
 	const zen::Oid ServerSessionId = zen::GetSessionId();
diff --git a/src/zenserver/main.cpp b/src/zenserver/main.cpp
index bf328c499..dff162b1c 100644
--- a/src/zenserver/main.cpp
+++ b/src/zenserver/main.cpp
@@ -123,10 +123,6 @@ AppMain(int argc, char* argv[])
 	signal(SIGINT, utils::SignalCallbackHandler);
 	signal(SIGTERM, utils::SignalCallbackHandler);
 
-#if ZEN_PLATFORM_LINUX
-	IgnoreChildSignals();
-#endif
-
 	try
 	{
 		typename Main::Config ServerOptions;
diff --git a/src/zentest-appstub/zentest-appstub.cpp b/src/zentest-appstub/zentest-appstub.cpp
index 509629739..13c96ebe2 100644
--- a/src/zentest-appstub/zentest-appstub.cpp
+++ b/src/zentest-appstub/zentest-appstub.cpp
@@ -291,6 +291,20 @@ main(int argc, char* argv[])
 				std::string_view ErrorArg = SplitArg(argv[i]);
 				ExitCode				  = ParseIntArg(ErrorArg);
 			}
+			else if (std::strncmp(argv[i], "-echo=", 6) == 0)
+			{
+				// Write a message to stdout. Useful for testing pipe capture.
+				std::string_view Message = SplitArg(argv[i]);
+				printf("%.*s", static_cast<int>(Message.size()), Message.data());
+				fflush(stdout);
+			}
+			else if (std::strncmp(argv[i], "-echoerr=", 9) == 0)
+			{
+				// Write a message to stderr. Useful for testing separate stderr pipe capture.
+				std::string_view Message = SplitArg(argv[i]);
+				fprintf(stderr, "%.*s", static_cast<int>(Message.size()), Message.data());
+				fflush(stderr);
+			}
 			else if ((_strnicmp(argv[i], "-input=", 7) == 0) || (_strnicmp(argv[i], "-i=", 3) == 0))
 			{
 				/* mimic DDC2
diff --git a/src/zenutil/config/loggingconfig.cpp b/src/zenutil/config/loggingconfig.cpp
index 5092c60aa..e2db31160 100644
--- a/src/zenutil/config/loggingconfig.cpp
+++ b/src/zenutil/config/loggingconfig.cpp
@@ -29,6 +29,8 @@ ZenLoggingCmdLineOptions::AddCliOptions(cxxopts::Options& options, ZenLoggingCon
 		("log-critical", 	"Change selected loggers to level CRITICAL", 				cxxopts::value<std::string>(LoggingConfig.Loggers[logging::Critical]))
 		("log-off", 		"Change selected loggers to level OFF", 					cxxopts::value<std::string>(LoggingConfig.Loggers[logging::Off]))
 		("otlp-endpoint", 	"OpenTelemetry endpoint URI (e.g http://localhost:4318)", 	cxxopts::value<std::string>(LoggingConfig.OtelEndpointUri))
+		("force-color",		"Force colored log output even when stdout is not a terminal",	cxxopts::value<bool>(LoggingConfig.ForceColor)->default_value("false"))
+		("log-stream",		"TCP log stream endpoint (host:port)",							cxxopts::value<std::string>(LoggingConfig.LogStreamEndpoint))
 	;
 	// clang-format on
 }
diff --git a/src/zenutil/include/zenutil/config/loggingconfig.h b/src/zenutil/include/zenutil/config/loggingconfig.h
index b55b2d9f7..33a5eb172 100644
--- a/src/zenutil/include/zenutil/config/loggingconfig.h
+++ b/src/zenutil/include/zenutil/config/loggingconfig.h
@@ -16,10 +16,12 @@ struct ZenLoggingConfig
 {
 	bool				  NoConsoleOutput = false;	// Control default use of stdout for diagnostics
 	bool				  QuietConsole	  = false;	// Configure console logger output to level WARN
+	bool				  ForceColor	  = false;	// Force colored output even when stdout is not a terminal
 	std::filesystem::path AbsLogFile;				// Absolute path to main log file
 	std::string			  Loggers[logging::LogLevelCount];
-	std::string			  LogId;			// Id for tagging log output
-	std::string			  OtelEndpointUri;	// OpenTelemetry endpoint URI
+	std::string			  LogId;			  // Id for tagging log output
+	std::string			  OtelEndpointUri;	  // OpenTelemetry endpoint URI
+	std::string			  LogStreamEndpoint;  // TCP log stream endpoint (host:port)
 };
 
 void ApplyLoggingOptions(cxxopts::Options& options, ZenLoggingConfig& LoggingConfig);
diff --git a/src/zenutil/include/zenutil/logging.h b/src/zenutil/include/zenutil/logging.h
index 95419c274..282ae1b9a 100644
--- a/src/zenutil/include/zenutil/logging.h
+++ b/src/zenutil/include/zenutil/logging.h
@@ -28,7 +28,8 @@ struct LoggingOptions
 	bool				  AllowAsync	  = true;
 	bool				  NoConsoleOutput = false;
 	bool				  QuietConsole	  = false;
-	std::filesystem::path AbsLogFile;  // Absolute path to main log file
+	bool				  ForceColor	  = false;	// Force colored output even when stdout is not a terminal
+	std::filesystem::path AbsLogFile;				// Absolute path to main log file
 	std::string			  LogId;
 };
 
diff --git a/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h b/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h
new file mode 100644
index 000000000..06544308c
--- /dev/null
+++ b/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h
@@ -0,0 +1,66 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/zencore.h>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio/io_context.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+/// Interface for receiving log lines from a LogStreamListener.
+/// Clients implement this to route received log messages to their desired output.
+class LogStreamHandler
+{
+public:
+	virtual ~LogStreamHandler() = default;
+
+	virtual void AppendLogLine(std::string Line) = 0;
+};
+
+/// TCP listener that accepts connections from remote processes streaming log messages.
+/// Each message is a CbObject with fields: "text" (string), "source" (string), "level" (string, optional).
+///
+/// A LogStreamHandler can be set to receive parsed log lines. If no handler is set,
+/// received messages are silently discarded.
+///
+/// Two modes of operation:
+///  - Owned thread: pass only Port; an internal IO thread is created.
+///  - External io_context: pass an existing asio::io_context; no thread is created,
+///    the caller is responsible for running the io_context.
+class LogStreamListener
+{
+public:
+	/// Start listening with an internal IO thread.
+	explicit LogStreamListener(uint16_t Port = 0);
+
+	/// Start listening on an externally-driven io_context (no thread created).
+	LogStreamListener(asio::io_context& IoContext, uint16_t Port = 0);
+
+	~LogStreamListener();
+
+	LogStreamListener(const LogStreamListener&) = delete;
+	LogStreamListener& operator=(const LogStreamListener&) = delete;
+
+	/// Set the handler that will receive parsed log lines. May be called at any time.
+	/// Pass nullptr to stop delivering messages.
+	void SetHandler(LogStreamHandler* Handler);
+
+	/// Returns the actual port the listener is bound to.
+	uint16_t GetPort() const;
+
+	/// Gracefully stop accepting new connections and shut down existing sessions.
+	void Shutdown();
+
+private:
+	struct Impl;
+	std::unique_ptr<Impl> m_Impl;
+};
+
+}  // namespace zen
diff --git a/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h b/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h
new file mode 100644
index 000000000..2ab7d469e
--- /dev/null
+++ b/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h
@@ -0,0 +1,209 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/logging.h>
+#include <zencore/logging/sink.h>
+#include <zencore/thread.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+#include <atomic>
+#include <condition_variable>
+#include <deque>
+#include <mutex>
+#include <string>
+#include <thread>
+
+namespace zen {
+
+/// Logging sink that connects to a LogStreamListener via TCP and sends CbObject-framed log messages.
+/// Connection is lazy (on first enqueued message) and silently reconnects on failure.
+/// Messages are serialized on the caller thread and written asynchronously from a dedicated IO thread.
+/// A bounded queue drops the oldest messages on overflow to prevent unbounded memory growth.
+class TcpLogStreamSink : public logging::Sink
+{
+public:
+	TcpLogStreamSink(const std::string& Host, uint16_t Port, std::string Source, uint32_t MaxQueueSize = 4096)
+	: m_Host(Host)
+	, m_Port(Port)
+	, m_Source(std::move(Source))
+	, m_MaxQueueSize(MaxQueueSize)
+	{
+		m_IoThread = std::thread([this]() { IoThreadMain(); });
+	}
+
+	~TcpLogStreamSink() override
+	{
+		{
+			std::lock_guard<std::mutex> Lock(m_QueueMutex);
+			m_Stopping		= true;
+			m_DrainDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
+		}
+		m_QueueCv.notify_one();
+		if (m_IoThread.joinable())
+		{
+			m_IoThread.join();
+		}
+	}
+
+	void Log(const logging::LogMessage& Msg) override
+	{
+		logging::MemoryBuffer Formatted;
+		{
+			RwLock::SharedLockScope Lock(m_FormatterLock);
+			if (m_Formatter)
+			{
+				m_Formatter->Format(Msg, Formatted);
+			}
+			else
+			{
+				// Fallback: use raw payload
+				auto Payload = Msg.GetPayload();
+				Formatted.append(Payload.data(), Payload.data() + Payload.size());
+			}
+		}
+
+		std::string_view Text(Formatted.data(), Formatted.size());
+
+		// Strip trailing newlines
+		while (!Text.empty() && (Text.back() == '\n' || Text.back() == '\r'))
+		{
+			Text.remove_suffix(1);
+		}
+
+		// Build CbObject with text, source, and level fields
+		CbObjectWriter Writer;
+		Writer.AddString("text", Text);
+		Writer.AddString("source", m_Source);
+		Writer.AddString("level", ToStringView(Msg.GetLevel()));
+		CbObject Obj = Writer.Save();
+
+		// Enqueue for async write
+		{
+			std::lock_guard<std::mutex> Lock(m_QueueMutex);
+			if (m_Queue.size() >= m_MaxQueueSize)
+			{
+				m_Queue.pop_front();
+				m_DroppedMessages.fetch_add(1, std::memory_order_relaxed);
+			}
+			m_Queue.push_back(std::move(Obj));
+		}
+		m_QueueCv.notify_one();
+	}
+
+	void Flush() override
+	{
+		// Nothing to flush — writes happen asynchronously
+	}
+
+	void SetFormatter(std::unique_ptr<logging::Formatter> InFormatter) override
+	{
+		RwLock::ExclusiveLockScope Lock(m_FormatterLock);
+		m_Formatter = std::move(InFormatter);
+	}
+
+private:
+	void IoThreadMain()
+	{
+		zen::SetCurrentThreadName("TcpLogSink");
+
+		for (;;)
+		{
+			std::deque<CbObject> Batch;
+			{
+				std::unique_lock<std::mutex> Lock(m_QueueMutex);
+				m_QueueCv.wait(Lock, [this]() { return m_Stopping || !m_Queue.empty(); });
+
+				if (m_Stopping && m_Queue.empty())
+				{
+					break;
+				}
+
+				if (m_Stopping && std::chrono::steady_clock::now() >= m_DrainDeadline)
+				{
+					break;
+				}
+
+				Batch.swap(m_Queue);
+			}
+
+			uint32_t Dropped = m_DroppedMessages.exchange(0, std::memory_order_relaxed);
+			if (Dropped > 0)
+			{
+				// We could/should log here, but that could cause a feedback loop which
+				// would trigger subsequent dropped message warnings
+			}
+
+			if (!m_Connected && !Connect())
+			{
+				if (m_Stopping)
+				{
+					break;	// don't retry during shutdown
+				}
+				continue;  // drop batch — will retry on next batch
+			}
+
+			for (auto& Obj : Batch)
+			{
+				MemoryView		 View = Obj.GetView();
+				asio::error_code Ec;
+				asio::write(m_Socket, asio::buffer(View.GetData(), View.GetSize()), Ec);
+				if (Ec)
+				{
+					m_Connected = false;
+					break;	// drop remaining messages in batch
+				}
+			}
+		}
+	}
+
+	bool Connect()
+	{
+		try
+		{
+			asio::ip::tcp::resolver Resolver(m_IoContext);
+			auto					Endpoints = Resolver.resolve(m_Host, std::to_string(m_Port));
+			asio::connect(m_Socket, Endpoints);
+			m_Connected = true;
+			return true;
+		}
+		catch (const std::exception&)
+		{
+			// Reset the socket for next attempt
+			m_Socket	= asio::ip::tcp::socket(m_IoContext);
+			m_Connected = false;
+			return false;
+		}
+	}
+
+	// IO thread state (only accessed from m_IoThread)
+	asio::io_context	  m_IoContext;
+	asio::ip::tcp::socket m_Socket{m_IoContext};
+	bool				  m_Connected = false;
+
+	// Configuration (immutable after construction)
+	std::string m_Host;
+	uint16_t	m_Port;
+	std::string m_Source;
+	uint32_t	m_MaxQueueSize;
+
+	// Formatter (protected by RwLock since Log is called from multiple threads)
+	RwLock								m_FormatterLock;
+	std::unique_ptr<logging::Formatter> m_Formatter;
+
+	// Queue shared between Log() callers and IO thread
+	std::mutex							  m_QueueMutex;
+	std::condition_variable				  m_QueueCv;
+	std::deque<CbObject>				  m_Queue;
+	std::atomic<uint32_t>				  m_DroppedMessages{0};
+	bool								  m_Stopping = false;
+	std::chrono::steady_clock::time_point m_DrainDeadline;
+
+	std::thread m_IoThread;
+};
+
+}  // namespace zen
diff --git a/src/zenutil/splitconsole/logstreamlistener.cpp b/src/zenutil/splitconsole/logstreamlistener.cpp
new file mode 100644
index 000000000..9f1d1a02c
--- /dev/null
+++ b/src/zenutil/splitconsole/logstreamlistener.cpp
@@ -0,0 +1,270 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/splitconsole/logstreamlistener.h>
+
+#include <zencore/compactbinary.h>
+#include <zencore/fmtutils.h>
+#include <zencore/logging.h>
+#include <zencore/thread.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+#include <atomic>
+#include <vector>
+
+namespace zen {
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamSession — reads CbObject-framed messages from a single TCP connection
+
+class LogStreamSession : public std::enable_shared_from_this<LogStreamSession>
+{
+public:
+	LogStreamSession(asio::ip::tcp::socket Socket, std::atomic<LogStreamHandler*>& Handler)
+	: m_Socket(std::move(Socket))
+	, m_Handler(Handler)
+	{
+	}
+
+	void Start() { DoRead(); }
+
+private:
+	void DoRead()
+	{
+		auto Self = shared_from_this();
+		m_Socket.async_read_some(asio::buffer(m_ReadBuf.data() + m_BufferUsed, m_ReadBuf.size() - m_BufferUsed),
+								 [Self](const asio::error_code& Ec, std::size_t BytesRead) {
+									 if (Ec)
+									 {
+										 return;  // connection closed or error — session ends
+									 }
+									 Self->m_BufferUsed += BytesRead;
+									 Self->ProcessBuffer();
+									 Self->DoRead();
+								 });
+	}
+
+	void ProcessBuffer()
+	{
+		// Try to consume as many complete CbObject messages as possible
+		while (m_BufferUsed > 0)
+		{
+			MemoryView	View = MakeMemoryView(m_ReadBuf.data(), m_BufferUsed);
+			CbFieldType Type;
+			uint64_t	Size = 0;
+
+			if (!TryMeasureCompactBinary(View, Type, Size))
+			{
+				break;	// need more data
+			}
+
+			if (Size > m_BufferUsed)
+			{
+				break;	// need more data
+			}
+
+			// Parse the CbObject
+			CbObject Obj = CbObject(SharedBuffer::Clone(MakeMemoryView(m_ReadBuf.data(), Size)));
+
+			std::string_view Text	= Obj["text"].AsString();
+			std::string_view Source = Obj["source"].AsString();
+
+			// Split multi-line messages into individual AppendLogLine calls so that
+			// each line gets its own row in the SplitConsole ring buffer.
+			while (!Text.empty())
+			{
+				std::string_view Line = Text;
+				auto			 Pos  = Text.find('\n');
+				if (Pos != std::string_view::npos)
+				{
+					Line = Text.substr(0, Pos);
+					Text.remove_prefix(Pos + 1);
+				}
+				else
+				{
+					Text = {};
+				}
+
+				// Strip trailing CR from CRLF
+				if (!Line.empty() && Line.back() == '\r')
+				{
+					Line.remove_suffix(1);
+				}
+
+				if (Line.empty())
+				{
+					continue;
+				}
+
+				LogStreamHandler* Handler = m_Handler.load(std::memory_order_acquire);
+				if (Handler)
+				{
+					if (!Source.empty())
+					{
+						Handler->AppendLogLine(fmt::format("[{}] {}", Source, Line));
+					}
+					else
+					{
+						Handler->AppendLogLine(std::string(Line));
+					}
+				}
+			}
+
+			// Remove consumed bytes from buffer
+			std::size_t Consumed = static_cast<std::size_t>(Size);
+			std::memmove(m_ReadBuf.data(), m_ReadBuf.data() + Consumed, m_BufferUsed - Consumed);
+			m_BufferUsed -= Consumed;
+		}
+
+		// If buffer is full and we can't parse a message, the message is too large — drop connection
+		if (m_BufferUsed == m_ReadBuf.size())
+		{
+			ZEN_WARN("LogStreamSession: buffer full with no complete message, dropping connection");
+			asio::error_code Ec;
+			m_Socket.close(Ec);
+			m_BufferUsed = 0;
+		}
+	}
+
+	asio::ip::tcp::socket			m_Socket;
+	std::atomic<LogStreamHandler*>& m_Handler;
+	std::array<uint8_t, 65536>		m_ReadBuf{};
+	std::size_t						m_BufferUsed = 0;
+};
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamListener::Impl
+
+struct LogStreamListener::Impl
+{
+	// Owned io_context mode — creates and runs its own thread
+	explicit Impl(uint16_t Port) : m_OwnedIoContext(std::make_unique<asio::io_context>()), m_Acceptor(*m_OwnedIoContext)
+	{
+		SetupAcceptor(Port);
+		m_IoThread = std::thread([this]() {
+			zen::SetCurrentThreadName("LogStreamIO");
+			m_OwnedIoContext->run();
+		});
+	}
+
+	// External io_context mode — caller drives the io_context
+	Impl(asio::io_context& IoContext, uint16_t Port) : m_Acceptor(IoContext) { SetupAcceptor(Port); }
+
+	~Impl() { Shutdown(); }
+
+	void Shutdown()
+	{
+		if (m_Stopped.exchange(true))
+		{
+			return;
+		}
+
+		asio::error_code Ec;
+		m_Acceptor.close(Ec);
+
+		if (m_OwnedIoContext)
+		{
+			m_OwnedIoContext->stop();
+		}
+
+		if (m_IoThread.joinable())
+		{
+			m_IoThread.join();
+		}
+	}
+
+	uint16_t GetPort() const { return m_Port; }
+
+	void SetHandler(LogStreamHandler* Handler) { m_Handler.store(Handler, std::memory_order_release); }
+
+private:
+	void SetupAcceptor(uint16_t Port)
+	{
+		// Try dual-stack IPv6 first (accepts both IPv4 and IPv6), fall back to IPv4-only
+		asio::error_code Ec;
+		m_Acceptor.open(asio::ip::tcp::v6(), Ec);
+		if (!Ec)
+		{
+			m_Acceptor.set_option(asio::ip::v6_only(false), Ec);
+			if (!Ec)
+			{
+				m_Acceptor.bind(asio::ip::tcp::endpoint(asio::ip::tcp::v6(), Port), Ec);
+			}
+		}
+
+		if (Ec)
+		{
+			// Fall back to IPv4-only
+			if (m_Acceptor.is_open())
+			{
+				m_Acceptor.close();
+			}
+			m_Acceptor.open(asio::ip::tcp::v4());
+			m_Acceptor.bind(asio::ip::tcp::endpoint(asio::ip::tcp::v4(), Port));
+		}
+
+		m_Acceptor.listen();
+		m_Port = m_Acceptor.local_endpoint().port();
+		StartAccept();
+	}
+
+	void StartAccept()
+	{
+		m_Acceptor.async_accept([this](const asio::error_code& Ec, asio::ip::tcp::socket Socket) {
+			if (Ec)
+			{
+				return;	 // acceptor closed
+			}
+
+			auto Session = std::make_shared<LogStreamSession>(std::move(Socket), m_Handler);
+			Session->Start();
+
+			if (!m_Stopped.load())
+			{
+				StartAccept();
+			}
+		});
+	}
+
+	std::atomic<LogStreamHandler*>	  m_Handler{nullptr};
+	std::unique_ptr<asio::io_context> m_OwnedIoContext;	 // null when using external io_context
+	asio::ip::tcp::acceptor			  m_Acceptor;
+	std::thread						  m_IoThread;
+	uint16_t						  m_Port = 0;
+	std::atomic<bool>				  m_Stopped{false};
+};
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamListener
+
+LogStreamListener::LogStreamListener(uint16_t Port) : m_Impl(std::make_unique<Impl>(Port))
+{
+}
+
+LogStreamListener::LogStreamListener(asio::io_context& IoContext, uint16_t Port) : m_Impl(std::make_unique<Impl>(IoContext, Port))
+{
+}
+
+LogStreamListener::~LogStreamListener() = default;
+
+void
+LogStreamListener::SetHandler(LogStreamHandler* Handler)
+{
+	m_Impl->SetHandler(Handler);
+}
+
+uint16_t
+LogStreamListener::GetPort() const
+{
+	return m_Impl->GetPort();
+}
+
+void
+LogStreamListener::Shutdown()
+{
+	m_Impl->Shutdown();
+}
+
+}  // namespace zen