13 files changed, 852 insertions, 37 deletions

diff --git a/src/zencore/compactbinaryvalidation.cpp b/src/zencore/compactbinaryvalidation.cpp
index 3e78f8ef1..5c3b9ff19 100644
--- a/src/zencore/compactbinaryvalidation.cpp
+++ b/src/zencore/compactbinaryvalidation.cpp
@@ -10,6 +10,9 @@
 
 #include <algorithm>
 
+#include <EASTL/fixed_vector.h>
+#include <EASTL/sort.h>
+
 namespace zen {
 
 namespace CbValidationPrivate {
@@ -240,7 +243,7 @@ ValidateCbObject(MemoryView& View, CbValidateMode Mode, CbValidateError& Error,
 
 	if (Size > 0)
 	{
-		std::vector<std::string_view> Names;
+		eastl::fixed_vector<std::string_view, 32> Names;
 
 		const bool				 bUniformObject = CbFieldTypeOps::GetType(ObjectType) == CbFieldType::UniformObject;
 		const CbFieldType		 ExternalType	= bUniformObject ? ValidateCbFieldType(ObjectView, Mode, Error) : CbFieldType::HasFieldType;
@@ -265,7 +268,7 @@ ValidateCbObject(MemoryView& View, CbValidateMode Mode, CbValidateError& Error,
 
 		if (EnumHasAnyFlags(Mode, CbValidateMode::Names) && Names.size() > 1)
 		{
-			std::sort(begin(Names), end(Names), [](std::string_view L, std::string_view R) { return L.compare(R) < 0; });
+			std::sort(Names.begin(), Names.end(), [](std::string_view L, std::string_view R) { return L.compare(R) < 0; });
 
 			for (const std::string_view *NamesIt = Names.data(), *NamesEnd = NamesIt + Names.size() - 1; NamesIt != NamesEnd; ++NamesIt)
 			{
@@ -639,8 +642,8 @@ ValidateObjectAttachment(MemoryView View, CbValidateMode Mode)
 CbValidateError
 ValidateCompactBinaryPackage(MemoryView View, CbValidateMode Mode)
 {
-	std::vector<IoHash> Attachments;
-	CbValidateError		Error = CbValidateError::None;
+	eastl::fixed_vector<IoHash, 32> Attachments;
+	CbValidateError					Error = CbValidateError::None;
 	if (EnumHasAnyFlags(Mode, CbValidateMode::All))
 	{
 		uint32_t ObjectCount = 0;
@@ -684,7 +687,7 @@ ValidateCompactBinaryPackage(MemoryView View, CbValidateMode Mode)
 
 		if (Attachments.size() && EnumHasAnyFlags(Mode, CbValidateMode::Package))
 		{
-			std::sort(begin(Attachments), end(Attachments));
+			eastl::sort(Attachments.begin(), Attachments.end());
 			for (const IoHash *It = Attachments.data(), *End = It + Attachments.size() - 1; It != End; ++It)
 			{
 				if (It[0] == It[1])
diff --git a/src/zencore/include/zencore/logging/backlogsink.h b/src/zencore/include/zencore/logging/backlogsink.h
new file mode 100644
index 000000000..be170db9d
--- /dev/null
+++ b/src/zencore/include/zencore/logging/backlogsink.h
@@ -0,0 +1,104 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/logging/sink.h>
+#include <zencore/memory/memoryarena.h>
+#include <zencore/thread.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <EASTL/vector.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+#include <atomic>
+#include <chrono>
+#include <limits>
+#include <memory>
+
+namespace zen::logging {
+
+/// Captures every message it sees so that sinks attached *after* logging
+/// has already been emitting can replay the early window into themselves.
+///
+/// Typical lifecycle:
+///   1. At process start, install one BacklogSink as a child of the
+///      default broadcast sink. Capture begins immediately.
+///   2. As real sinks come online (file, console, in-proc session, …),
+///      they're attached to the broadcast and can call Replay() to
+///      receive the lines emitted before they joined.
+///   3. At a well-defined "we're past the bootstrap window" point — the
+///      server's `OnReady()` for zenserver, command dispatch for the zen
+///      CLI — call Disable() to stop capturing and free everything.
+///
+/// Storage is bounded and arena-backed: the vector and every captured
+/// string are allocated from a single MemoryArena that gets dropped
+/// wholesale on Disable(). The vector is reserved up front so push_back
+/// never reallocates; once full we drop new entries (the early-startup
+/// lines are the ones we most want to preserve, so keep-oldest is the
+/// right policy here).
+class BacklogSink : public Sink
+{
+public:
+	/// @param MaxEntries  Hard cap on captured messages. The vector is
+	///                    reserved up front from the arena. 256 is a
+	///                    comfortable cap for the bootstrap window —
+	///                    typical zen process startup emits a few dozen
+	///                    lines, well under the limit.
+	explicit BacklogSink(size_t MaxEntries = 256);
+	~BacklogSink() override;
+
+	BacklogSink(const BacklogSink&) = delete;
+	BacklogSink& operator=(const BacklogSink&) = delete;
+
+	void Log(const LogMessage& Msg) override;
+	void Flush() override;
+	void SetFormatter(std::unique_ptr<Formatter> InFormatter) override;
+
+	/// Replay captured messages into Target in arrival order. By default
+	/// every entry currently in the buffer is replayed; pass @p MaxEntries
+	/// to stop after that many (useful when a caller has snapshotted a
+	/// cursor to avoid re-delivering messages the target already received
+	/// live — see AttachSinkWithBacklogReplay). Target's ShouldLog() filter
+	/// is honored. Replay does NOT remove entries from the buffer — it's
+	/// safe to call multiple times.
+	void Replay(Sink& Target, size_t MaxEntries = std::numeric_limits<size_t>::max());
+
+	/// Stop capturing and free the entire arena (vector buffer + every
+	/// captured string in one shot). Subsequent Log() calls return
+	/// immediately. Idempotent.
+	void Disable();
+
+	bool   IsEnabled() const { return m_Enabled.load(std::memory_order_acquire); }
+	size_t Size() const;
+	size_t Dropped() const { return m_DroppedCount.load(std::memory_order_relaxed); }
+
+private:
+	struct Entry
+	{
+		// LogPoint pointers come from static / near-static instances
+		// owned by the ZEN_LOG macros (see asyncsink.cpp's note); the
+		// pointer outlives any plausible backlog buffer.
+		const LogPoint* Point = nullptr;
+
+		// Both strings live in m_Arena — null-terminated, so we keep them
+		// as plain const char* and reconstruct string_views on replay.
+		const char* LoggerName = nullptr;
+		const char* Payload	   = nullptr;
+
+		std::chrono::system_clock::time_point Time;
+		int									  ThreadId = 0;
+	};
+
+	mutable RwLock m_Lock;
+	// Declaration order matters: m_Arena must be initialized before
+	// m_Entries because m_Entries' allocator references m_Arena. Both
+	// outlive every captured Entry — they're released as a unit when
+	// this BacklogSink is destroyed.
+	MemoryArena						 m_Arena{32768};
+	eastl::vector<Entry, ArenaAlloc> m_Entries{ArenaAlloc(m_Arena)};
+	std::atomic<bool>				 m_Enabled{true};
+	std::atomic<uint64_t>			 m_DroppedCount{0};
+	size_t							 m_MaxEntries;
+};
+
+}  // namespace zen::logging
diff --git a/src/zencore/include/zencore/logging/broadcastsink.h b/src/zencore/include/zencore/logging/broadcastsink.h
index 474662888..2b96e04f4 100644
--- a/src/zencore/include/zencore/logging/broadcastsink.h
+++ b/src/zencore/include/zencore/logging/broadcastsink.h
@@ -72,6 +72,21 @@ public:
 		m_Sinks.push_back(std::move(InSink));
 	}
 
+	/// Add a child sink and run @p OnAddedUnderLock with the broadcast's
+	/// exclusive lock still held. Useful for callers that need to capture
+	/// some external state atomically with the moment the sink becomes
+	/// visible to fanout — e.g. snapshotting a backlog cursor so a
+	/// subsequent replay can de-duplicate against messages that already
+	/// reached the new sink via Log(). The callback must not call back
+	/// into this BroadcastSink (would self-deadlock).
+	template<typename F>
+	void AddSinkAtomic(SinkPtr InSink, F&& OnAddedUnderLock)
+	{
+		RwLock::ExclusiveLockScope Lock(m_Lock);
+		m_Sinks.push_back(std::move(InSink));
+		OnAddedUnderLock();
+	}
+
 	void RemoveSink(const SinkPtr& InSink)
 	{
 		RwLock::ExclusiveLockScope Lock(m_Lock);
diff --git a/src/zencore/include/zencore/memory/memoryarena.h b/src/zencore/include/zencore/memory/memoryarena.h
index 551415aac..167449036 100644
--- a/src/zencore/include/zencore/memory/memoryarena.h
+++ b/src/zencore/include/zencore/memory/memoryarena.h
@@ -3,7 +3,8 @@
 #pragma once
 
 #include <zencore/thread.h>
-#include <vector>
+
+#include <EASTL/fixed_vector.h>
 
 namespace zen {
 
@@ -14,6 +15,10 @@ namespace zen {
  * All allocations are freed when the arena is destroyed, therefore there
  * is no support for individual deallocation.
  *
+ * Chunk size is configurable at construction (default 64 KiB). Allocations
+ * larger than the configured chunk size get their own dedicated chunk so
+ * they don't waste or overrun the normal chunk.
+ *
  * For convenience, we include operator new/delete overloads below which
  * take a MemoryArena reference as a placement argument.
  *
@@ -22,7 +27,12 @@ namespace zen {
 class MemoryArena
 {
 public:
-	MemoryArena() = default;
+	/// Default chunk size (64 KiB). Sized to fit comfortably in L2 and
+	/// to minimize the number of chunks needed for typical workloads.
+	static constexpr size_t kDefaultChunkSize = 64 * 1024;
+
+	/// @param ChunkSize  Bytes per chunk. Zero is treated as the default.
+	explicit MemoryArena(size_t ChunkSize = kDefaultChunkSize);
 	~MemoryArena();
 
 	void*		AllocateAligned(size_t ByteCount, size_t align);
@@ -30,17 +40,45 @@ public:
 	void*		Allocate(size_t Size);
 	const char* DuplicateString(std::string_view Str);
 
+	/// Configured chunk size in bytes.
+	size_t GetChunkSize() const { return m_ChunkSize; }
+
+	/// Total bytes the arena currently holds across all chunks
+	/// (regular + oversize). This is the reserved footprint, not the
+	/// fraction of it that's been bump-pointer-allocated to callers —
+	/// useful as a "how big did this arena ever grow" diagnostic.
+	size_t GetReservedBytes() const;
+
 	MemoryArena(const MemoryArena&) = delete;
 	MemoryArena& operator=(const MemoryArena&) = delete;
 
 private:
-	static constexpr size_t ChunkSize = 16 * 1024;
+	// True when the request's worst-case footprint exceeds m_ChunkSize.
+	// Such requests get a dedicated chunk so the regular bump pointer is
+	// preserved for small allocations.
+	bool IsOversizeRequest(size_t WorstCaseBytes) const;
+
+	// Allocates a chunk and pushes it into m_Chunks. Caller is responsible
+	// for any bookkeeping (current pointer/offset). Must be called with
+	// m_Lock held exclusive. Returns nullptr on out-of-memory.
+	uint8_t* AllocateDedicatedChunk(size_t Capacity);
+
 	// TODO: should just chain the memory blocks together and avoid this
 	// vector altogether, saving us a memory allocation
-	std::vector<uint8_t*> m_Chunks;
-	uint8_t*			  m_CurrentChunk  = nullptr;
-	size_t				  m_CurrentOffset = 0;
-	RwLock				  m_Lock;
+	struct Chunk
+	{
+		uint8_t* Base	  = nullptr;
+		size_t	 Capacity = 0;	// May exceed m_ChunkSize for oversize allocations.
+	};
+	// 32 inline chunks keeps the vector itself allocation-free for any
+	// arena that stays under ~2 MiB at the default chunk size; growth
+	// beyond that falls back to heap.
+	eastl::fixed_vector<Chunk, 32> m_Chunks;
+	uint8_t*					   m_CurrentChunk		  = nullptr;
+	size_t						   m_CurrentChunkCapacity = 0;
+	size_t						   m_CurrentOffset		  = 0;
+	size_t						   m_ChunkSize;
+	mutable RwLock				   m_Lock;
 };
 
 // Allocator suitable for use with EASTL
diff --git a/src/zencore/include/zencore/oidguid.h b/src/zencore/include/zencore/oidguid.h
new file mode 100644
index 000000000..16d8cd101
--- /dev/null
+++ b/src/zencore/include/zencore/oidguid.h
@@ -0,0 +1,81 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/guid.h>
+#include <zencore/uid.h>
+
+#include <cstdint>
+#include <cstring>
+
+namespace zen {
+
+/// Convert a 12-byte Oid into a UUID-shaped 16-byte Guid that
+/// round-trips back through GuidToOid. Used to bridge zen session ids
+/// (Oid, 12 bytes) and tools that consume UUID-shaped session
+/// identifiers — most notably UE Trace, whose session id is an FGuid
+/// memcpy'd into the trace stream's `Desc.SessionGuid`.
+///
+/// The transform packs the 12 bytes as three big-endian uint32s into
+/// (A, B, C), stashes the bits of B/C that we're about to overwrite
+/// into D, then sets the RFC 4122 version-4 (random) and variant
+/// (10xx) markers in B and C. The result is a syntactically valid v4
+/// UUID — analyzers won't reject it — and the original 12 bytes are
+/// recoverable byte-for-byte via GuidToOid.
+///
+/// Note: the inverse direction (Guid→Oid→Guid for an arbitrary input
+/// Guid) is NOT lossless. GuidToOid restores 6 bits (version + variant)
+/// from D regardless of where they came from in the input, so the
+/// version/variant nibbles of an arbitrary FGuid are clobbered.
+/// Lossless only in the Oid→Guid→Oid direction, which is exactly what
+/// we need for "given a session Oid, get a UUID I can put in the
+/// trace; given a UUID from a trace I produced, get the session Oid
+/// back."
+inline Guid
+OidToGuid(const Oid& Id)
+{
+	const uint8_t* Bytes = reinterpret_cast<const uint8_t*>(Id.OidBits);
+	uint32_t	   A	 = (uint32_t(Bytes[0]) << 24) | (uint32_t(Bytes[1]) << 16) | (uint32_t(Bytes[2]) << 8) | uint32_t(Bytes[3]);
+	uint32_t	   B	 = (uint32_t(Bytes[4]) << 24) | (uint32_t(Bytes[5]) << 16) | (uint32_t(Bytes[6]) << 8) | uint32_t(Bytes[7]);
+	uint32_t	   C	 = (uint32_t(Bytes[8]) << 24) | (uint32_t(Bytes[9]) << 16) | (uint32_t(Bytes[10]) << 8) | uint32_t(Bytes[11]);
+
+	// Stash the two bytes that the version/variant markers below are
+	// about to overwrite. D ends up with only those two bytes set; all
+	// other bits are zero.
+	uint32_t D = (B & 0x0000FF00u) | (C & 0xFF000000u);
+
+	// RFC 4122 markers: version 4 in B[15..12], variant 10xx in C[31..30].
+	B = (B & ~0x0000F000u) | 0x00004000u;
+	C = (C & ~0xC0000000u) | 0x80000000u;
+
+	return Guid{A, B, C, D};
+}
+
+/// Inverse of OidToGuid. Restores the two bytes of B and C that were
+/// clobbered with version/variant markers by pulling them back out of
+/// D, then unpacks the three uint32s into 12 big-endian bytes.
+inline Oid
+GuidToOid(const Guid& Source)
+{
+	const uint32_t Bits0 = Source.A;
+	const uint32_t Bits1 = (Source.B & ~0x0000FF00u) | (Source.D & 0x0000FF00u);
+	const uint32_t Bits2 = (Source.C & ~0xFF000000u) | (Source.D & 0xFF000000u);
+
+	Oid		 Out{};
+	uint8_t* Bytes = reinterpret_cast<uint8_t*>(Out.OidBits);
+	Bytes[0]	   = uint8_t(Bits0 >> 24);
+	Bytes[1]	   = uint8_t(Bits0 >> 16);
+	Bytes[2]	   = uint8_t(Bits0 >> 8);
+	Bytes[3]	   = uint8_t(Bits0);
+	Bytes[4]	   = uint8_t(Bits1 >> 24);
+	Bytes[5]	   = uint8_t(Bits1 >> 16);
+	Bytes[6]	   = uint8_t(Bits1 >> 8);
+	Bytes[7]	   = uint8_t(Bits1);
+	Bytes[8]	   = uint8_t(Bits2 >> 24);
+	Bytes[9]	   = uint8_t(Bits2 >> 16);
+	Bytes[10]	   = uint8_t(Bits2 >> 8);
+	Bytes[11]	   = uint8_t(Bits2);
+	return Out;
+}
+
+}  // namespace zen
diff --git a/src/zencore/include/zencore/profiling/counterstrace.h b/src/zencore/include/zencore/profiling/counterstrace.h
new file mode 100644
index 000000000..1fa40a5f3
--- /dev/null
+++ b/src/zencore/include/zencore/profiling/counterstrace.h
@@ -0,0 +1,221 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+// Counter trace points: emit named time-series of int or float values into the
+// active .utrace stream. Wire-compatible with UE's Counters.Spec /
+// SetValueInt / SetValueFloat events (CountersChannel), so traces produced by
+// zen are readable by Unreal Insights and zen's own `zen trace` tooling.
+//
+// Typical use:
+//
+//   ZEN_TRACE_INT_VALUE("MySubsystem/QueueDepth", QueueSize);
+//   ZEN_TRACE_FLOAT_VALUE("MySubsystem/HitRate", Hits / Total);
+//   ZEN_TRACE_MEMORY_VALUE("MySubsystem/CacheBytes", BytesInUse);
+//
+// Each macro instantiates a function-static counter object the first time it
+// is hit, lazily registering the counter (Spec event) and then emitting a
+// SetValue event. Repeating the same value is suppressed (matching UE's
+// TCounter::Set semantics) -- use the _ALWAYS variants if every sample must
+// be emitted.
+
+#include <zencore/zencore.h>
+
+#include <atomic>
+#include <cstdint>
+#include <string_view>
+#include <type_traits>
+
+namespace zen {
+
+enum class TraceCounterType : uint8_t
+{
+	Int	  = 0,
+	Float = 1,
+};
+
+enum class TraceCounterDisplayHint : uint8_t
+{
+	None   = 0,
+	Memory = 1,
+};
+
+#if ZEN_WITH_TRACE
+
+namespace counters_detail {
+
+	// Returns the assigned counter id (1..0xFFFF) on success, or 0 if tracing is
+	// disabled / the channel is off. Emits a Counters.Spec event on first
+	// successful registration.
+	uint16_t OutputInitCounter(const char* Name, TraceCounterType Type, TraceCounterDisplayHint Hint);
+
+	// Emit a SetValueInt / SetValueFloat event. No-op when Id == 0.
+	void OutputSetValueInt(uint16_t Id, int64_t Value);
+	void OutputSetValueFloat(uint16_t Id, double Value);
+
+	// True when the Counters channel is currently active. The macros use this to
+	// skip work when nobody is listening.
+	bool IsCountersChannelEnabled();
+
+}  // namespace counters_detail
+
+template<typename ValueType, TraceCounterType Type, typename StoredType = ValueType, bool bUnchecked = false>
+class TraceCounter
+{
+public:
+	TraceCounter(const char* InName, TraceCounterDisplayHint InHint) : m_Name(InName), m_Hint(InHint), m_Id(0)
+	{
+		uint32_t Id = counters_detail::OutputInitCounter(InName, Type, InHint);
+		m_Id.store(Id, std::memory_order_relaxed);
+	}
+
+	TraceCounter(const TraceCounter&) = delete;
+	TraceCounter& operator=(const TraceCounter&) = delete;
+
+	ValueType Get() const { return ValueType(m_Value); }
+
+	void Set(ValueType InValue)
+	{
+		if (bUnchecked || ValueType(m_Value) != InValue)
+		{
+			m_Value = InValue;
+			Emit();
+		}
+	}
+
+	void SetAlways(ValueType InValue)
+	{
+		m_Value = InValue;
+		Emit();
+	}
+
+	void Add(ValueType InValue)
+	{
+		if (bUnchecked || InValue != ValueType{})
+		{
+			m_Value += InValue;
+			Emit();
+		}
+	}
+
+	void Subtract(ValueType InValue)
+	{
+		if (bUnchecked || InValue != ValueType{})
+		{
+			m_Value -= InValue;
+			Emit();
+		}
+	}
+
+	void Increment()
+	{
+		++m_Value;
+		Emit();
+	}
+
+	void Decrement()
+	{
+		--m_Value;
+		Emit();
+	}
+
+private:
+	void Emit()
+	{
+		if (!counters_detail::IsCountersChannelEnabled())
+		{
+			return;
+		}
+		LateInit();
+		uint16_t Id = uint16_t(m_Id.load(std::memory_order_relaxed));
+		if constexpr (Type == TraceCounterType::Int)
+		{
+			counters_detail::OutputSetValueInt(Id, int64_t(ValueType(m_Value)));
+		}
+		else
+		{
+			counters_detail::OutputSetValueFloat(Id, double(ValueType(m_Value)));
+		}
+	}
+
+	void LateInit()
+	{
+		uint32_t Old = m_Id.load(std::memory_order_relaxed);
+		if (Old != 0)
+		{
+			return;
+		}
+		uint32_t New = counters_detail::OutputInitCounter(m_Name, Type, m_Hint);
+		if (New != 0)
+		{
+			m_Id.compare_exchange_strong(Old, New, std::memory_order_relaxed);
+		}
+	}
+
+	StoredType				m_Value{};
+	const char*				m_Name = nullptr;
+	TraceCounterDisplayHint m_Hint = TraceCounterDisplayHint::None;
+	std::atomic<uint32_t>	m_Id;
+};
+
+using TraceCounterInt		= TraceCounter<int64_t, TraceCounterType::Int>;
+using TraceCounterFloat		= TraceCounter<double, TraceCounterType::Float>;
+using TraceCounterAtomicInt = TraceCounter<int64_t, TraceCounterType::Int, std::atomic<int64_t>>;
+
+#else  // ZEN_WITH_TRACE
+
+// Stub implementation -- inlines compile to nothing.
+namespace counters_detail {
+	inline uint16_t OutputInitCounter(const char*, TraceCounterType, TraceCounterDisplayHint) { return 0; }
+	inline void		OutputSetValueInt(uint16_t, int64_t) {}
+	inline void		OutputSetValueFloat(uint16_t, double) {}
+	inline bool		IsCountersChannelEnabled() { return false; }
+}  // namespace counters_detail
+
+template<typename ValueType, TraceCounterType, typename = ValueType, bool = false>
+class TraceCounter
+{
+public:
+	TraceCounter(const char*, TraceCounterDisplayHint) {}
+	ValueType Get() const { return ValueType{}; }
+	void	  Set(ValueType) {}
+	void	  SetAlways(ValueType) {}
+	void	  Add(ValueType) {}
+	void	  Subtract(ValueType) {}
+	void	  Increment() {}
+	void	  Decrement() {}
+};
+
+using TraceCounterInt		= TraceCounter<int64_t, TraceCounterType::Int>;
+using TraceCounterFloat		= TraceCounter<double, TraceCounterType::Float>;
+using TraceCounterAtomicInt = TraceCounter<int64_t, TraceCounterType::Int>;
+
+#endif	// ZEN_WITH_TRACE
+
+}  // namespace zen
+
+// ---------------------------------------------------------------------------
+// Public emit macros (mirror UE TRACE_INT_VALUE / TRACE_FLOAT_VALUE etc.)
+// ---------------------------------------------------------------------------
+
+#define ZEN_COUNTERS_TRACE_CONCAT_IMPL(a, b) a##b
+#define ZEN_COUNTERS_TRACE_CONCAT(a, b)		 ZEN_COUNTERS_TRACE_CONCAT_IMPL(a, b)
+
+#define ZEN_COUNTERS_TRACE_DECLARE_AND_SET(CounterCls, Name, Hint, Value)                                 \
+	do                                                                                                    \
+	{                                                                                                     \
+		static ::zen::CounterCls ZEN_COUNTERS_TRACE_CONCAT(__ZenTraceCounter_, __LINE__){(Name), (Hint)}; \
+		ZEN_COUNTERS_TRACE_CONCAT(__ZenTraceCounter_, __LINE__).Set(Value);                               \
+	} while (0)
+
+// Set an integer counter to Value. Suppresses repeated identical values.
+#define ZEN_TRACE_INT_VALUE(Name, Value) \
+	ZEN_COUNTERS_TRACE_DECLARE_AND_SET(TraceCounterInt, Name, ::zen::TraceCounterDisplayHint::None, int64_t(Value))
+
+// Set a floating-point counter to Value.
+#define ZEN_TRACE_FLOAT_VALUE(Name, Value) \
+	ZEN_COUNTERS_TRACE_DECLARE_AND_SET(TraceCounterFloat, Name, ::zen::TraceCounterDisplayHint::None, double(Value))
+
+// Set a memory-style counter (formatted as bytes by the viewer).
+#define ZEN_TRACE_MEMORY_VALUE(Name, Bytes) \
+	ZEN_COUNTERS_TRACE_DECLARE_AND_SET(TraceCounterInt, Name, ::zen::TraceCounterDisplayHint::Memory, int64_t(Bytes))
diff --git a/src/zencore/include/zencore/session.h b/src/zencore/include/zencore/session.h
index 10c33da24..807af878e 100644
--- a/src/zencore/include/zencore/session.h
+++ b/src/zencore/include/zencore/session.h
@@ -12,4 +12,8 @@ struct Oid;
 [[nodiscard]] Oid			   GetSessionId();
 [[nodiscard]] std::string_view GetSessionIdString();
 
+void						   SetParentSessionId(const Oid& ParentSessionId);
+[[nodiscard]] Oid			   GetParentSessionId();
+[[nodiscard]] std::string_view GetParentSessionIdString();
+
 }  // namespace zen
diff --git a/src/zencore/logging/backlogsink.cpp b/src/zencore/logging/backlogsink.cpp
new file mode 100644
index 000000000..f7c02c2e4
--- /dev/null
+++ b/src/zencore/logging/backlogsink.cpp
@@ -0,0 +1,118 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zencore/logging/backlogsink.h>
+
+#include <zencore/logging/logmsg.h>
+
+namespace zen::logging {
+
+BacklogSink::BacklogSink(size_t MaxEntries) : m_MaxEntries(MaxEntries)
+{
+	// Reserve up front so push_back never reallocates. The reserve itself
+	// takes one chunk from the arena; subsequent string duplications carve
+	// from the same arena, so the entire backlog footprint is one
+	// contiguous allocation domain released wholesale when this object is
+	// destroyed (see DisableLogBacklog: it drops the last Ref).
+	m_Entries.reserve(MaxEntries);
+}
+
+BacklogSink::~BacklogSink() = default;
+
+void
+BacklogSink::Log(const LogMessage& Msg)
+{
+	// Cheap acquire-load gate: once Disable() has been called we want
+	// further log calls to do as little as possible. This is the common
+	// case after the bootstrap window closes — every log line in the
+	// process flows through here for the rest of the run.
+	if (!m_Enabled.load(std::memory_order_acquire))
+	{
+		return;
+	}
+
+	RwLock::ExclusiveLockScope Lock(m_Lock);
+
+	// Re-check under the lock: a Disable() racing with Log() may have
+	// flipped the flag between our acquire and the lock acquisition.
+	if (!m_Enabled.load(std::memory_order_relaxed))
+	{
+		return;
+	}
+
+	if (m_Entries.size() >= m_MaxEntries)
+	{
+		m_DroppedCount.fetch_add(1, std::memory_order_relaxed);
+		return;
+	}
+
+	Entry& Captured		= m_Entries.emplace_back();
+	Captured.Point		= &Msg.GetLogPoint();
+	Captured.LoggerName = m_Arena.DuplicateString(Msg.GetLoggerName());
+	Captured.Payload	= m_Arena.DuplicateString(Msg.GetPayload());
+	Captured.Time		= Msg.GetTime();
+	Captured.ThreadId	= Msg.GetThreadId();
+}
+
+void
+BacklogSink::Flush()
+{
+	// Nothing to do — we're an in-memory buffer, not an output device.
+}
+
+void
+BacklogSink::SetFormatter(std::unique_ptr<Formatter> InFormatter)
+{
+	// We don't format anything — entries are replayed verbatim into the
+	// target sink, which applies its own formatter. Accept and discard
+	// for interface compatibility.
+	(void)InFormatter;
+}
+
+void
+BacklogSink::Replay(Sink& Target, size_t MaxEntries)
+{
+	RwLock::SharedLockScope Lock(m_Lock);
+	const size_t			Limit = std::min(MaxEntries, m_Entries.size());
+	for (size_t Index = 0; Index < Limit; ++Index)
+	{
+		const Entry& Captured = m_Entries[Index];
+		if (!Captured.Point)
+		{
+			continue;
+		}
+		const std::string_view LoggerName = Captured.LoggerName ? std::string_view(Captured.LoggerName) : std::string_view{};
+		const std::string_view Payload	  = Captured.Payload ? std::string_view(Captured.Payload) : std::string_view{};
+		LogMessage			   Msg(*Captured.Point, LoggerName, Payload);
+		Msg.SetTime(Captured.Time);
+		Msg.SetThreadId(Captured.ThreadId);
+		if (Target.ShouldLog(Msg.GetLevel()))
+		{
+			try
+			{
+				Target.Log(Msg);
+			}
+			catch (const std::exception&)
+			{
+				// Match the swallow-on-replay semantics of AsyncSink's
+				// dispatcher so a misbehaving target doesn't poison the
+				// rest of the replay.
+			}
+		}
+	}
+}
+
+void
+BacklogSink::Disable()
+{
+	RwLock::ExclusiveLockScope Lock(m_Lock);
+	m_Enabled.store(false, std::memory_order_release);
+}
+
+size_t
+BacklogSink::Size() const
+{
+	RwLock::SharedLockScope Lock(m_Lock);
+	return m_Entries.size();
+}
+
+}  // namespace zen::logging
diff --git a/src/zencore/memory/memoryarena.cpp b/src/zencore/memory/memoryarena.cpp
index 8807f3264..9eabaf71e 100644
--- a/src/zencore/memory/memoryarena.cpp
+++ b/src/zencore/memory/memoryarena.cpp
@@ -2,18 +2,52 @@
 
 #include <zencore/memory/memoryarena.h>
 
+#include <algorithm>
 #include <cstring>
 
 namespace zen {
 
+MemoryArena::MemoryArena(size_t ChunkSize) : m_ChunkSize(ChunkSize == 0 ? kDefaultChunkSize : ChunkSize)
+{
+}
+
 MemoryArena::~MemoryArena()
 {
-	for (auto Chunk : m_Chunks)
-		delete[] Chunk;
+	for (const Chunk& C : m_Chunks)
+	{
+		delete[] C.Base;
+	}
+}
+
+// An allocation whose worst-case footprint won't fit inside a freshly
+// allocated standard-sized chunk is "oversize" — we route it to its own
+// dedicated chunk so it doesn't waste the rest of the regular chunk's
+// space, and so a subsequent small allocation still has a near-empty
+// regular chunk to bump-pointer into.
+//
+// Note: WorstCaseBytes is the upper bound on bytes the allocation could
+// consume from a freshly aligned chunk start. The actual returned
+// pointer + size fits within this footprint by construction.
+bool
+MemoryArena::IsOversizeRequest(size_t WorstCaseBytes) const
+{
+	return WorstCaseBytes > m_ChunkSize;
+}
+
+uint8_t*
+MemoryArena::AllocateDedicatedChunk(size_t Capacity)
+{
+	uint8_t* NewChunk = new (std::nothrow) uint8_t[Capacity];
+	if (!NewChunk)
+	{
+		return nullptr;
+	}
+	m_Chunks.push_back({NewChunk, Capacity});
+	return NewChunk;
 }
 
 void*
-MemoryArena::AllocateAligned(size_t ByteCount, size_t align)
+MemoryArena::AllocateAligned(size_t ByteCount, size_t Align)
 {
 	if (ByteCount == 0)
 	{
@@ -23,19 +57,31 @@ MemoryArena::AllocateAligned(size_t ByteCount, size_t align)
 	void* Ptr = nullptr;
 
 	m_Lock.WithExclusiveLock([&] {
-		size_t AlignedOffset = (m_CurrentOffset + (align - 1)) & ~(align - 1);
+		// Oversize fast path — give the request its own chunk, leave the
+		// regular bump pointer untouched so subsequent small allocs still
+		// reuse it.
+		const size_t WorstCase = ByteCount + Align - 1;
+		if (IsOversizeRequest(WorstCase))
+		{
+			if (uint8_t* Dedicated = AllocateDedicatedChunk(WorstCase))
+			{
+				const size_t Aligned = (reinterpret_cast<uintptr_t>(Dedicated) + (Align - 1)) & ~(uintptr_t(Align - 1));
+				Ptr					 = reinterpret_cast<void*>(Aligned);
+			}
+			return;
+		}
 
-		if (m_CurrentChunk == nullptr || AlignedOffset + ByteCount > ChunkSize)
+		size_t AlignedOffset = (m_CurrentOffset + (Align - 1)) & ~(Align - 1);
+		if (m_CurrentChunk == nullptr || AlignedOffset + ByteCount > m_CurrentChunkCapacity)
 		{
-			uint8_t* NewChunk = new uint8_t[ChunkSize];
+			uint8_t* NewChunk = AllocateDedicatedChunk(m_ChunkSize);
 			if (!NewChunk)
 			{
 				return;
 			}
-
-			m_Chunks.push_back(NewChunk);
-			m_CurrentChunk = NewChunk;
-			AlignedOffset  = 0;
+			m_CurrentChunk		   = NewChunk;
+			m_CurrentChunkCapacity = m_ChunkSize;
+			AlignedOffset		   = 0;
 		}
 
 		Ptr				= m_CurrentChunk + AlignedOffset;
@@ -46,7 +92,7 @@ MemoryArena::AllocateAligned(size_t ByteCount, size_t align)
 }
 
 void*
-MemoryArena::AllocateAlignedWithOffset(size_t ByteCount, size_t align, size_t offset)
+MemoryArena::AllocateAlignedWithOffset(size_t ByteCount, size_t Align, size_t Offset)
 {
 	if (ByteCount == 0)
 	{
@@ -56,19 +102,29 @@ MemoryArena::AllocateAlignedWithOffset(size_t ByteCount, size_t align, size_t of
 	void* Ptr = nullptr;
 
 	m_Lock.WithExclusiveLock([&] {
-		size_t AlignedOffset = (m_CurrentOffset + (align - 1) + offset) & ~(align - 1);
+		const size_t WorstCase = ByteCount + Align - 1 + Offset;
+		if (IsOversizeRequest(WorstCase))
+		{
+			if (uint8_t* Dedicated = AllocateDedicatedChunk(WorstCase))
+			{
+				const uintptr_t Base	= reinterpret_cast<uintptr_t>(Dedicated);
+				const uintptr_t Aligned = (Base + (Align - 1) + Offset) & ~(uintptr_t(Align - 1));
+				Ptr						= reinterpret_cast<void*>(Aligned);
+			}
+			return;
+		}
 
-		if (m_CurrentChunk == nullptr || AlignedOffset + ByteCount > ChunkSize)
+		size_t AlignedOffset = (m_CurrentOffset + (Align - 1) + Offset) & ~(Align - 1);
+		if (m_CurrentChunk == nullptr || AlignedOffset + ByteCount > m_CurrentChunkCapacity)
 		{
-			uint8_t* NewChunk = new uint8_t[ChunkSize];
+			uint8_t* NewChunk = AllocateDedicatedChunk(m_ChunkSize);
 			if (!NewChunk)
 			{
 				return;
 			}
-
-			m_Chunks.push_back(NewChunk);
-			m_CurrentChunk = NewChunk;
-			AlignedOffset  = offset;
+			m_CurrentChunk		   = NewChunk;
+			m_CurrentChunkCapacity = m_ChunkSize;
+			AlignedOffset		   = Offset;
 		}
 
 		Ptr				= m_CurrentChunk + AlignedOffset;
@@ -90,19 +146,28 @@ MemoryArena::Allocate(size_t Size)
 	constexpr size_t Alignment = alignof(std::max_align_t);
 
 	m_Lock.WithExclusiveLock([&] {
-		size_t AlignedOffset = (m_CurrentOffset + Alignment - 1) & ~(Alignment - 1);
+		const size_t WorstCase = Size + Alignment - 1;
+		if (IsOversizeRequest(WorstCase))
+		{
+			if (uint8_t* Dedicated = AllocateDedicatedChunk(WorstCase))
+			{
+				const uintptr_t Aligned = (reinterpret_cast<uintptr_t>(Dedicated) + Alignment - 1) & ~(uintptr_t(Alignment - 1));
+				Ptr						= reinterpret_cast<void*>(Aligned);
+			}
+			return;
+		}
 
-		if (m_CurrentChunk == nullptr || AlignedOffset + Size > ChunkSize)
+		size_t AlignedOffset = (m_CurrentOffset + Alignment - 1) & ~(Alignment - 1);
+		if (m_CurrentChunk == nullptr || AlignedOffset + Size > m_CurrentChunkCapacity)
 		{
-			uint8_t* NewChunk = new uint8_t[ChunkSize];
+			uint8_t* NewChunk = AllocateDedicatedChunk(m_ChunkSize);
 			if (!NewChunk)
 			{
 				return;
 			}
-
-			m_Chunks.push_back(NewChunk);
-			m_CurrentChunk = NewChunk;
-			AlignedOffset  = 0;
+			m_CurrentChunk		   = NewChunk;
+			m_CurrentChunkCapacity = m_ChunkSize;
+			AlignedOffset		   = 0;
 		}
 
 		Ptr				= m_CurrentChunk + AlignedOffset;
@@ -112,6 +177,19 @@ MemoryArena::Allocate(size_t Size)
 	return Ptr;
 }
 
+size_t
+MemoryArena::GetReservedBytes() const
+{
+	size_t Total = 0;
+	m_Lock.WithSharedLock([&] {
+		for (const Chunk& Item : m_Chunks)
+		{
+			Total += Item.Capacity;
+		}
+	});
+	return Total;
+}
+
 const char*
 MemoryArena::DuplicateString(std::string_view Str)
 {
diff --git a/src/zencore/profiling/counterstrace.cpp b/src/zencore/profiling/counterstrace.cpp
new file mode 100644
index 000000000..63e8a2dd3
--- /dev/null
+++ b/src/zencore/profiling/counterstrace.cpp
@@ -0,0 +1,99 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zencore/profiling/counterstrace.h>
+
+#if ZEN_WITH_TRACE
+
+#	include <zencore/timer.h>
+#	include <zencore/trace.h>
+
+#	include <atomic>
+#	include <cstring>
+
+namespace {
+
+// Wire-compatible with UE's CountersTrace events (Counters.Spec /
+// SetValueInt / SetValueFloat). Channel name matches UE so users can mix &
+// match analyzers.
+UE_TRACE_CHANNEL_DEFINE(CountersChannel)
+
+UE_TRACE_EVENT_BEGIN(Counters, Spec, NoSync | Important)
+	UE_TRACE_EVENT_FIELD(uint16_t, Id)
+	UE_TRACE_EVENT_FIELD(uint8_t, Type)
+	UE_TRACE_EVENT_FIELD(uint8_t, DisplayHint)
+	UE_TRACE_EVENT_FIELD(UE::Trace::AnsiString, Name)
+UE_TRACE_EVENT_END()
+
+UE_TRACE_EVENT_BEGIN(Counters, SetValueInt)
+	UE_TRACE_EVENT_FIELD(uint64_t, Cycle)
+	UE_TRACE_EVENT_FIELD(int64_t, Value)
+	UE_TRACE_EVENT_FIELD(uint16_t, CounterId)
+UE_TRACE_EVENT_END()
+
+UE_TRACE_EVENT_BEGIN(Counters, SetValueFloat)
+	UE_TRACE_EVENT_FIELD(uint64_t, Cycle)
+	UE_TRACE_EVENT_FIELD(double, Value)
+	UE_TRACE_EVENT_FIELD(uint16_t, CounterId)
+UE_TRACE_EVENT_END()
+
+}  // namespace
+
+namespace zen::counters_detail {
+
+uint16_t
+OutputInitCounter(const char* Name, TraceCounterType Type, TraceCounterDisplayHint Hint)
+{
+	if (!UE_TRACE_CHANNELEXPR_IS_ENABLED(CountersChannel) || Name == nullptr)
+	{
+		return 0;
+	}
+
+	// Counter ids are uint16; tourist's analyzer truncates anyway. Wrapping
+	// past 0xFFFF would re-use ids, so we cap allocation -- in practice no
+	// real-world trace uses anywhere near 65k distinct counters.
+	static std::atomic<uint32_t> g_NextId{0};
+	uint32_t					 Allocated = g_NextId.fetch_add(1, std::memory_order_relaxed) + 1;
+	if (Allocated > 0xFFFFu)
+	{
+		return 0;
+	}
+	uint16_t Id = uint16_t(Allocated);
+
+	uint16_t NameLen = uint16_t(std::strlen(Name));
+	UE_TRACE_LOG(Counters, Spec, CountersChannel, NameLen * sizeof(char))
+		<< Spec.Id(Id) << Spec.Type(uint8_t(Type)) << Spec.DisplayHint(uint8_t(Hint)) << Spec.Name(Name, NameLen);
+
+	return Id;
+}
+
+void
+OutputSetValueInt(uint16_t Id, int64_t Value)
+{
+	if (Id == 0 || !UE_TRACE_CHANNELEXPR_IS_ENABLED(CountersChannel))
+	{
+		return;
+	}
+	UE_TRACE_LOG(Counters, SetValueInt, CountersChannel)
+		<< SetValueInt.Cycle(zen::GetHifreqTimerValue()) << SetValueInt.Value(Value) << SetValueInt.CounterId(Id);
+}
+
+void
+OutputSetValueFloat(uint16_t Id, double Value)
+{
+	if (Id == 0 || !UE_TRACE_CHANNELEXPR_IS_ENABLED(CountersChannel))
+	{
+		return;
+	}
+	UE_TRACE_LOG(Counters, SetValueFloat, CountersChannel)
+		<< SetValueFloat.Cycle(zen::GetHifreqTimerValue()) << SetValueFloat.Value(Value) << SetValueFloat.CounterId(Id);
+}
+
+bool
+IsCountersChannelEnabled()
+{
+	return UE_TRACE_CHANNELEXPR_IS_ENABLED(CountersChannel);
+}
+
+}  // namespace zen::counters_detail
+
+#endif	// ZEN_WITH_TRACE
diff --git a/src/zencore/session.cpp b/src/zencore/session.cpp
index 50c47c628..d6386e43b 100644
--- a/src/zencore/session.cpp
+++ b/src/zencore/session.cpp
@@ -12,6 +12,10 @@ static Oid			  GlobalSessionId;
 static Oid::String_t  GlobalSessionString;
 static std::once_flag SessionInitFlag;
 
+static Oid			 GlobalParentSessionId;
+static Oid::String_t GlobalParentSessionString;
+static std::mutex	 GlobalParentSessionMutex;
+
 Oid
 GetSessionId()
 {
@@ -32,4 +36,30 @@ GetSessionIdString()
 	return std::string_view(GlobalSessionString, Oid::StringLength);
 }
 
+void
+SetParentSessionId(const Oid& ParentSessionId)
+{
+	std::lock_guard<std::mutex> Lock(GlobalParentSessionMutex);
+	GlobalParentSessionId = ParentSessionId;
+	GlobalParentSessionId.ToString(GlobalParentSessionString);
+}
+
+Oid
+GetParentSessionId()
+{
+	std::lock_guard<std::mutex> Lock(GlobalParentSessionMutex);
+	return GlobalParentSessionId;
+}
+
+std::string_view
+GetParentSessionIdString()
+{
+	std::lock_guard<std::mutex> Lock(GlobalParentSessionMutex);
+	if (GlobalParentSessionId == Oid::Zero)
+	{
+		return {};
+	}
+	return std::string_view(GlobalParentSessionString, Oid::StringLength);
+}
+
 }  // namespace zen
diff --git a/src/zencore/trace.cpp b/src/zencore/trace.cpp
index d6a0b2e92..61db5a714 100644
--- a/src/zencore/trace.cpp
+++ b/src/zencore/trace.cpp
@@ -76,6 +76,11 @@ TraceConfigure(const TraceOptions& Options)
 			// does not emit. Redirect to the zenlog channel so --trace=log does what users expect.
 			ProcessChannelList("zenlog"sv);
 		}
+		else if (Arg == "metrics"sv)
+		{
+			// Convenience alias for counter-based metric tracing.
+			ProcessChannelList("counters"sv);
+		}
 		else
 		{
 			// Presume that the argument is a trace channel name
diff --git a/src/zencore/uid.cpp b/src/zencore/uid.cpp
index 971683721..a7e68acb2 100644
--- a/src/zencore/uid.cpp
+++ b/src/zencore/uid.cpp
@@ -3,6 +3,7 @@
 #include <zencore/uid.h>
 
 #include <zencore/endian.h>
+#include <zencore/oidguid.h>
 #include <zencore/string.h>
 #include <zencore/testing.h>
 
@@ -187,6 +188,24 @@ TEST_CASE("Oid")
 	}
 }
 
+TEST_CASE("OidGuid roundtrip")
+{
+	// Lossless in the Oid -> Guid -> Oid direction by construction —
+	// see oidguid.h. The version (4) and variant (10xx) markers in the
+	// produced Guid mean it's also a syntactically valid v4 UUID.
+	for (int Iteration = 0; Iteration < 1000; ++Iteration)
+	{
+		Oid	 Original = Oid::NewOid();
+		Guid AsGuid	  = OidToGuid(Original);
+		Oid	 Restored = GuidToOid(AsGuid);
+		CHECK(Restored == Original);
+
+		// Version and variant bits should always be set.
+		CHECK((AsGuid.B & 0x0000F000u) == 0x00004000u);
+		CHECK((AsGuid.C & 0xC0000000u) == 0x80000000u);
+	}
+}
+
 TEST_SUITE_END();
 
 void