static thread_local std::mt19937_64 rng{std::random_device{}()};

std::uniform_int_distribution<int> dist(0, 255);

std::string out;

out.reserve(bytes * 2);

const char* hex = "0123456789abcdef";

for (std::size_t i = 0; i < bytes; ++i) {

const int v = dist(rng);

out.push_back(hex[(v >> 4) & 0xF]);

out.push_back(hex[v & 0xF]);

}

return out;

std::vector<std::string> result;

if (s.empty()) {

return result;

}

std::size_t i = 0;

const std::size_t n = s.size();

while (i < n) {

std::size_t end = std::min(i + max_bytes, n);

// If end points into a UTF-8 continuation byte, walk back to a lead byte.

// NOTE: end is an index into the string; ensure end < n before indexing.

while (end > i && end < n && (static_cast<unsigned char>(s[end]) & 0xC0) == 0x80) {

--end;

}

// If end==n, we may still have landed on a continuation byte due to min().

while (end > i && end == n && (static_cast<unsigned char>(s[end - 1]) & 0xC0) == 0x80) {

// Walk back until we hit a lead byte boundary.

// This isn't perfect but is consistent with your earlier intent.

--end;

// If we backed up too far, fall through to fallback.

}

if (end == i) {

// Fallback: avoid infinite loop if bytes are pathological.

end = std::min(i + max_bytes, n);

}

result.emplace_back(s.substr(i, end - i));

i = end;

}

return result;

const std::string& topic, std::int64_t timestamp_ms, const std::optional<std::size_t>& total_size,

const std::map<std::string, std::string>& attrs) {

dst.stream_id = stream_id;

dst.mime_type = mime_type;

dst.topic = topic;

dst.timestamp = timestamp_ms;

dst.size = total_size;

dst.attributes = attrs;

{

const std::scoped_lock<std::mutex> lock(mutex_);

if (closed_) {

return;

}

queue_.push_back(text);

}

cv_.notify_one();

{

const std::scoped_lock<std::mutex> lock(mutex_);

for (const auto& kv : trailer_attrs) {

info_.attributes[kv.first] = kv.second;

}

closed_ = true;

}

cv_.notify_all();

std::unique_lock<std::mutex> lock(mutex_);

cv_.wait(lock, [this] { return !queue_.empty() || closed_; });

if (!queue_.empty()) {

out = std::move(queue_.front());

queue_.pop_front();

return true;

}

return false; // closed_ and empty

std::string result;

std::string chunk;

while (readNext(chunk)) {

result += chunk;

}

return result;

{

const std::scoped_lock<std::mutex> lock(mutex_);

if (closed_) {

return;

}

queue_.push_back(bytes);

}

cv_.notify_one();

{

const std::scoped_lock<std::mutex> lock(mutex_);

for (const auto& kv : trailer_attrs) {

info_.attributes[kv.first] = kv.second;

}

closed_ = true;

}

cv_.notify_all();

std::unique_lock<std::mutex> lock(mutex_);

cv_.wait(lock, [this] { return !queue_.empty() || closed_; });

if (!queue_.empty()) {

out = std::move(queue_.front());

queue_.pop_front();

return true;

}

return false;

std::map<std::string, std::string> attributes, std::string stream_id,

std::optional<std::size_t> total_size, std::string mime_type,

std::vector<std::string> destination_identities, std::string sender_identity)

: local_participant_(local_participant),

stream_id_(stream_id.empty() ? generateRandomId() : std::move(stream_id)),

mime_type_(std::move(mime_type)),

topic_(std::move(topic)),

timestamp_ms_(

std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch())

.count()),

total_size_(total_size),

attributes_(std::move(attributes)),

destination_identities_(std::move(destination_identities)),

sender_identity_(std::move(sender_identity)) {

if (sender_identity_.empty()) {

sender_identity_ = local_participant_.identity();

}

if (header_sent_) {

return;

}

proto::DataStream::Header header;

header.set_stream_id(stream_id_);

header.set_timestamp(timestamp_ms_);

header.set_mime_type(mime_type_);

header.set_topic(topic_);

if (total_size_.has_value()) {

header.set_total_length(static_cast<std::uint64_t>(*total_size_));

}

for (const auto& kv : attributes_) {

(*header.mutable_attributes())[kv.first] = kv.second;

}

if (kind_ == StreamKind::kText) {

auto* th = header.mutable_text_header();

th->set_operation_type(proto::DataStream::OperationType::DataStream_OperationType_CREATE);

if (!reply_to_id_.empty()) {

th->set_reply_to_stream_id(reply_to_id_);

}

} else if (kind_ == StreamKind::kByte) {

header.mutable_byte_header()->set_name(byte_name_);

}

FfiClient::instance()

.sendStreamHeaderAsync(local_participant_.ffiHandleId(), header, destination_identities_, sender_identity_)

.get();

header_sent_ = true;

if (closed_) {

throw std::runtime_error("Cannot send chunk after stream is closed");

}

ensureHeaderSent();

proto::DataStream::Chunk chunk;

chunk.set_stream_id(stream_id_);

chunk.set_chunk_index(next_chunk_index_++); // ✅ uses the new member

chunk.set_content(content.data(), content.size());

FfiClient::instance()

.sendStreamChunkAsync(local_participant_.ffiHandleId(), chunk, destination_identities_, sender_identity_)

.get();

ensureHeaderSent();

proto::DataStream::Trailer trailer;

trailer.set_stream_id(stream_id_);

trailer.set_reason(reason);

for (const auto& kv : attributes) {

(*trailer.mutable_attributes())[kv.first] = kv.second;

}

FfiClient::instance().sendStreamTrailerAsync(local_participant_.ffiHandleId(), trailer, sender_identity_).get();

if (closed_) {

throw std::runtime_error("Stream already closed");

}

closed_ = true;

sendTrailer(reason, attributes);

const std::map<std::string, std::string>& attributes, const std::string& stream_id,

std::optional<std::size_t> total_size, const std::string& reply_to_id,

const std::vector<std::string>& destination_identities,

const std::string& sender_identity)

: BaseStreamWriter(local_participant, topic, attributes, stream_id, total_size,

/*mime_type=*/"text/plain", destination_identities, sender_identity) {

kind_ = StreamKind::kText;

reply_to_id_ = reply_to_id;

// ✅ Canonical user-facing metadata comes from BaseStreamWriter fields.

fillBaseInfo(info_, stream_id_, mime_type_, topic_, timestamp_ms_, total_size_, attributes_);

if (!reply_to_id.empty()) {

info_.reply_to_stream_id = reply_to_id;

}

const std::scoped_lock<std::mutex> lock(write_mutex_);

if (closed_) {

throw std::runtime_error("Cannot write to closed TextStreamWriter");

}

for (const auto& chunk_str : splitUtf8(text, kStreamChunkSize)) {

const auto* p = reinterpret_cast<const std::uint8_t*>(chunk_str.data());

const std::vector<std::uint8_t> bytes(p, p + chunk_str.size());

LK_LOG_DEBUG("sending chunk");

sendChunk(bytes);

}

const std::string& topic, const std::map<std::string, std::string>& attributes,

const std::string& stream_id, std::optional<std::size_t> total_size,

const std::string& mime_type, const std::vector<std::string>& destination_identities,

const std::string& sender_identity)

: BaseStreamWriter(local_participant, topic, attributes, stream_id, total_size, mime_type, destination_identities,

sender_identity) {

kind_ = StreamKind::kByte;

byte_name_ = name;

fillBaseInfo(info_, stream_id_, mime_type_, topic_, timestamp_ms_, total_size_, attributes_);

info_.name = name;

const std::scoped_lock<std::mutex> lock(write_mutex_);

if (closed_) {

throw std::runtime_error("Cannot write to closed ByteStreamWriter");

}

std::size_t offset = 0;

while (offset < data.size()) {

const std::size_t n = std::min<std::size_t>(kStreamChunkSize, data.size() - offset);

auto it = data.begin() + static_cast<std::ptrdiff_t>(offset);

const std::vector<std::uint8_t> chunk(it, it + static_cast<std::ptrdiff_t>(n));

sendChunk(chunk);

offset += n;

}