diff --git a/README.md b/README.md
index 408820b..8de428b 100644
--- a/README.md
+++ b/README.md
@@ -573,6 +573,30 @@ Practical notes:
 - if HEVC samples include in-band parameter sets, the publisher preserves `hev1` because rewriting those samples would be incorrect
 - if you start from a progressive MP4, this project can remux it internally, but pre-fragmented input is still the simpler and more efficient path
 
+### Live fragmented MP4 stdin publishing
+
+For live encoder pipelines, the publisher can consume fragmented MP4 directly from
+standard input.
+
+This live path expects ffmpeg to emit track-separated fragments, where each
+`moof` + `mdat` pair belongs to a single media track. Use `+separate_moof` when
+generating the stream. Without `+separate_moof`, audio and video may be carried
+inside the same `moof`, which is not the intended input layout for the current
+live parser.
+
+```bash
+ffmpeg -stream_loop -1 -re -i bbb_sunflower_1080p_30fps_normal.mp4 \
+  -map 0:v:0 -map 0:a:0 \
+  -c:v libx264 -preset medium -r 30 -g 60 -keyint_min 60 -sc_threshold 0 -bf 0 \
+  -c:a aac -b:a 160k -ar 48000 -ac 2 \
+  -movflags +frag_keyframe+empty_moov+default_base_moof+separate_moof \
+  -f mp4 - | ./build/openmoq-publisher \
+    --input - \
+    --endpoint moqt://relay.example.com:443/moq \
+    --namespace live/demo \
+    --timeout 120
+```
+
 ## CI
 
 GitHub Actions is configured to build and test the project on:
diff --git a/include/openmoq/publisher/cmaf_segmenter.h b/include/openmoq/publisher/cmaf_segmenter.h
index c0ba43f..18b3b49 100644
--- a/include/openmoq/publisher/cmaf_segmenter.h
+++ b/include/openmoq/publisher/cmaf_segmenter.h
@@ -25,6 +25,7 @@ struct MediaFragment {
     std::uint64_t duration_us = 0;
     std::uint64_t earliest_presentation_time_us = 0;
     std::uint8_t sap_type = 0;
+    bool is_video_keyframe = false;  // True if this is a video track IDR/keyframe fragment
     PayloadBuffer payload;
 };
 
@@ -38,4 +39,12 @@ SegmentedMp4 segment_for_cmaf(const ParsedMp4& parsed_mp4, CmafObjectMode object
 std::string summarize_tracks(const std::vector<TrackDescription>& tracks);
 std::size_t payload_size(const PayloadBuffer& payload);
 
+// Build a MediaFragment from a single moof+mdat pair for live streaming.
+// group_id is assigned by the caller (incremented per track).
+// The fragment owns the combined moof+mdat bytes.
+MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
+                                  std::span<const std::uint8_t> mdat_bytes,
+                                  const std::vector<TrackDescription>& tracks,
+                                  std::size_t group_id);
+
 }  // namespace openmoq::publisher
diff --git a/include/openmoq/publisher/cmsf_packager.h b/include/openmoq/publisher/cmsf_packager.h
index a998133..f670e43 100644
--- a/include/openmoq/publisher/cmsf_packager.h
+++ b/include/openmoq/publisher/cmsf_packager.h
@@ -48,4 +48,13 @@ void emit_plan_objects(const PublishPlan& plan,
                        std::span<const std::uint8_t> bytes,
                        const std::filesystem::path& output_dir);
 
+// Build catalog JSON and track-specific init segments for live streaming
+struct LiveCatalog {
+    std::vector<std::uint8_t> catalog_payload;
+    std::vector<TrackInitialization> track_initializations;
+};
+LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
+                               std::span<const std::uint8_t> init_segment,
+                               bool is_live = true);
+
 }  // namespace openmoq::publisher
diff --git a/include/openmoq/publisher/mp4_box.h b/include/openmoq/publisher/mp4_box.h
index d520695..2a1650c 100644
--- a/include/openmoq/publisher/mp4_box.h
+++ b/include/openmoq/publisher/mp4_box.h
@@ -3,6 +3,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <iosfwd>
+#include <optional>
 #include <span>
 #include <string>
 #include <string_view>
@@ -57,4 +58,31 @@ std::vector<const Mp4Box*> find_boxes(const std::vector<Mp4Box>& boxes, std::str
 const Mp4Box* find_child_box(const Mp4Box& box, std::string_view type);
 std::span<const std::uint8_t> slice_bytes(std::span<const std::uint8_t> bytes, const ByteSpan& span);
 
+// Incremental MP4 box reader for streaming input (e.g. piped ffmpeg).
+// Buffers raw bytes and yields complete top-level boxes one at a time.
+struct StreamingBoxResult {
+    std::string type;
+    std::vector<std::uint8_t> bytes;
+};
+
+class StreamingMp4Reader {
+public:
+    // Append raw data to internal buffer.
+    void append(const std::uint8_t* data, std::size_t len);
+
+    // Read up to chunk_size bytes from input and append.
+    // Returns number of bytes read; 0 means EOF.
+    std::size_t read_from(std::istream& input, std::size_t chunk_size = 16384);
+
+    // Try to extract the next complete top-level box from the buffer.
+    // Returns std::nullopt if not enough data is available yet.
+    std::optional<StreamingBoxResult> next_box();
+
+private:
+    std::vector<std::uint8_t> buffer_;
+    std::size_t consumed_ = 0;
+
+    void compact();
+};
+
 }  // namespace openmoq::publisher
diff --git a/include/openmoq/publisher/transport/moqt_session.h b/include/openmoq/publisher/transport/moqt_session.h
index bb3fbc4..cd646e7 100644
--- a/include/openmoq/publisher/transport/moqt_session.h
+++ b/include/openmoq/publisher/transport/moqt_session.h
@@ -3,6 +3,7 @@
 #include "openmoq/publisher/cmsf_packager.h"
 #include "openmoq/publisher/transport/publisher_transport.h"
 
+#include <iosfwd>
 #include <optional>
 #include <chrono>
 #include <span>
@@ -31,6 +32,9 @@ class MoqtSession {
 
     TransportStatus connect(const EndpointConfig& endpoint, const TlsConfig& tls);
     TransportStatus publish(const openmoq::publisher::PublishPlan& plan);
+    TransportStatus publish_live(std::istream& input,
+                                 openmoq::publisher::DraftVersion draft_version,
+                                 bool split_cmaf_chunks);
     TransportStatus close(std::uint64_t application_error_code = 0);
 
 private:
diff --git a/src/cmaf_segmenter.cpp b/src/cmaf_segmenter.cpp
index 34ba790..99c47f5 100644
--- a/src/cmaf_segmenter.cpp
+++ b/src/cmaf_segmenter.cpp
@@ -1005,4 +1005,149 @@ std::size_t payload_size(const PayloadBuffer& payload) {
     return payload.owned_bytes.empty() ? payload.span.size : payload.owned_bytes.size();
 }
 
+MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
+                                  std::span<const std::uint8_t> mdat_bytes,
+                                  const std::vector<TrackDescription>& tracks,
+                                  std::size_t group_id) {
+    // Extract timing data directly from the standalone moof bytes.
+    // For fragmented MP4 from ffmpeg, tfhd carries default_sample_flags so we
+    // do not need to look up trex defaults from the init segment.
+    const std::vector<Mp4Box> moof_boxes = parse_mp4_boxes(moof_bytes);
+    if (moof_boxes.empty() || moof_boxes.front().type != "moof") {
+        throw std::runtime_error("build_live_fragment: expected moof box");
+    }
+    const Mp4Box& moof = moof_boxes.front();
+
+    // Extract track name from moof -> traf -> tfhd -> track_id.
+    const std::string track_name = fragment_track_name(moof, tracks, moof_bytes);
+
+    // Extract timing information directly from moof bytes.
+    const TrackDescription* track_desc = fragment_track_description(moof, tracks, moof_bytes);
+    if (track_desc == nullptr || track_desc->timescale == 0) {
+        throw std::runtime_error("build_live_fragment: cannot find track for fragment");
+    }
+
+    const Mp4Box* traf = find_child_box(moof, "traf");
+    if (traf == nullptr) {
+        throw std::runtime_error("build_live_fragment: moof has no traf");
+    }
+
+    std::uint64_t base_decode_time = 0;
+    if (const Mp4Box* tfdt = find_child_box(*traf, "tfdt")) {
+        const std::uint8_t version = moof_bytes[tfdt->payload.offset];
+        const std::size_t time_offset = tfdt->payload.offset + 4;
+        if (version == 1 && time_offset + 8 <= moof_bytes.size()) {
+            std::uint64_t val = 0;
+            for (int i = 0; i < 8; ++i) {
+                val = (val << 8U) | moof_bytes[time_offset + i];
+            }
+            base_decode_time = val;
+        } else if (time_offset + 4 <= moof_bytes.size()) {
+            base_decode_time = read_be32(moof_bytes, time_offset);
+        }
+    }
+
+    std::uint32_t default_sample_duration = 0;
+    std::uint32_t default_sample_flags = 0x02000000U;
+    if (const Mp4Box* tfhd = find_child_box(*traf, "tfhd")) {
+        const std::uint32_t flags = read_full_box_flags(*tfhd, moof_bytes);
+        std::size_t cursor = tfhd->payload.offset + 8;
+        if ((flags & 0x000001U) != 0) cursor += 8;
+        if ((flags & 0x000002U) != 0) cursor += 4;
+        if ((flags & 0x000008U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            default_sample_duration = read_be32(moof_bytes, cursor);
+            cursor += 4;
+        }
+        if ((flags & 0x000010U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            cursor += 4;
+        }
+        if ((flags & 0x000020U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            default_sample_flags = read_be32(moof_bytes, cursor);
+        }
+    }
+
+    std::uint64_t duration = 0;
+    std::uint64_t earliest_presentation_time = base_decode_time;
+    bool earliest_presentation_time_set = false;
+    std::uint32_t first_sample_flags = default_sample_flags;
+    if (const Mp4Box* trun = find_child_box(*traf, "trun")) {
+        const std::uint32_t flags = read_full_box_flags(*trun, moof_bytes);
+        std::size_t cursor = trun->payload.offset + 4;
+        if (cursor + 4 <= moof_bytes.size()) {
+            const std::uint32_t sample_count = read_be32(moof_bytes, cursor);
+            cursor += 4;
+            if ((flags & 0x000001U) != 0) cursor += 4;
+            bool first_sample_flags_present = false;
+            if ((flags & 0x000004U) != 0) {
+                first_sample_flags = read_be32_or_zero(moof_bytes, cursor);
+                first_sample_flags_present = true;
+                cursor += 4;
+            }
+            std::uint64_t sample_decode_time = base_decode_time;
+            for (std::uint32_t i = 0; i < sample_count && cursor <= moof_bytes.size(); ++i) {
+                std::uint32_t sample_duration = default_sample_duration;
+                if ((flags & 0x000100U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    sample_duration = read_be32(moof_bytes, cursor);
+                    cursor += 4;
+                }
+
+                if ((flags & 0x000200U) != 0) cursor += 4;  // skip sample_size
+                std::uint32_t sample_flags = first_sample_flags_present && i == 0
+                                                 ? first_sample_flags : default_sample_flags;
+                if ((flags & 0x000400U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    sample_flags = read_be32(moof_bytes, cursor);
+                    cursor += 4;
+                }
+                std::int32_t composition_offset = 0;
+                if ((flags & 0x000800U) != 0 && cursor + 4 <= moof_bytes.size()) {
+                    composition_offset = static_cast<std::int32_t>(read_be32(moof_bytes, cursor));
+                    cursor += 4;
+                }
+                const std::int64_t pt_signed =
+                    static_cast<std::int64_t>(sample_decode_time) + static_cast<std::int64_t>(composition_offset);
+                const std::uint64_t pt = pt_signed < 0 ? 0 : static_cast<std::uint64_t>(pt_signed);
+                if (!earliest_presentation_time_set || pt < earliest_presentation_time) {
+                    earliest_presentation_time = pt;
+                    earliest_presentation_time_set = true;
+                }
+                if (i == 0 && (flags & 0x000400U) != 0) {
+                    first_sample_flags = sample_flags;
+                }
+                duration += sample_duration;
+                sample_decode_time += sample_duration;
+            }
+        }
+    }
+
+    std::uint8_t sap_type = 0;
+    const bool first_sample_is_sync = (first_sample_flags & 0x00010000U) == 0;
+    const bool is_video = (track_desc->handler_type == "vide");
+    if (!is_video) {
+        sap_type = 1;
+    } else if (first_sample_is_sync) {
+        sap_type = 2;
+    }
+
+    // A video keyframe: video track with sync first sample
+    const bool is_video_keyframe = is_video && first_sample_is_sync;
+
+    // Combine moof+mdat into a single owned payload (CMSF compliance).
+    std::vector<std::uint8_t> payload;
+    payload.reserve(moof_bytes.size() + mdat_bytes.size());
+    payload.insert(payload.end(), moof_bytes.begin(), moof_bytes.end());
+    payload.insert(payload.end(), mdat_bytes.begin(), mdat_bytes.end());
+
+    return MediaFragment{
+        .group_id = group_id,
+        .object_id = 0,
+        .track_name = track_name,
+        .start_time_us = scale_to_us(base_decode_time, track_desc->timescale),
+        .duration_us = scale_to_us(duration, track_desc->timescale),
+        .earliest_presentation_time_us = scale_to_us(earliest_presentation_time, track_desc->timescale),
+        .sap_type = sap_type,
+        .is_video_keyframe = is_video_keyframe,
+        .payload = {.span = {}, .owned_bytes = std::move(payload)},
+    };
+}
+
 }  // namespace openmoq::publisher
diff --git a/src/cmsf_packager.cpp b/src/cmsf_packager.cpp
index f1d7b03..3c028de 100644
--- a/src/cmsf_packager.cpp
+++ b/src/cmsf_packager.cpp
@@ -609,4 +609,104 @@ void emit_plan_objects(const PublishPlan& plan,
     manifest << render_publish_plan(plan);
 }
 
+LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
+                               std::span<const std::uint8_t> init_segment,
+                               bool is_live) {
+    // Local base64 encoder
+    auto local_base64_encode = [](std::span<const std::uint8_t> bytes) -> std::string {
+        static const char kAlphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+        std::string result;
+        result.reserve(((bytes.size() + 2) / 3) * 4);
+        for (std::size_t i = 0; i < bytes.size(); i += 3) {
+            const std::uint32_t b0 = bytes[i];
+            const std::uint32_t b1 = i + 1 < bytes.size() ? bytes[i + 1] : 0;
+            const std::uint32_t b2 = i + 2 < bytes.size() ? bytes[i + 2] : 0;
+            result.push_back(kAlphabet[b0 >> 2]);
+            result.push_back(kAlphabet[((b0 & 0x3) << 4) | (b1 >> 4)]);
+            result.push_back(i + 1 < bytes.size() ? kAlphabet[((b1 & 0xf) << 2) | (b2 >> 6)] : '=');
+            result.push_back(i + 2 < bytes.size() ? kAlphabet[b2 & 0x3f] : '=');
+        }
+        return result;
+    };
+
+    auto local_json_escape = [](std::string_view s) -> std::string {
+        std::string result;
+        result.reserve(s.size());
+        for (char c : s) {
+            switch (c) {
+                case '"': result += "\\\""; break;
+                case '\\': result += "\\\\"; break;
+                case '\n': result += "\\n"; break;
+                case '\r': result += "\\r"; break;
+                case '\t': result += "\\t"; break;
+                default: result += c; break;
+            }
+        }
+        return result;
+    };
+
+    auto local_track_role = [](std::string_view handler_type) -> std::string_view {
+        if (handler_type == "vide") return "video";
+        if (handler_type == "soun") return "audio";
+        if (handler_type == "meta") return "data";
+        return "data";
+    };
+
+    LiveCatalog result;
+
+    // For live streaming, use full init segment for all tracks (simpler than building track-specific)
+    const std::string full_init_base64 = local_base64_encode(init_segment);
+    for (const auto& track : tracks) {
+        result.track_initializations.push_back({
+            .track_name = track.track_name,
+            .codec_payload = {},
+            .init_segment = std::vector<std::uint8_t>(init_segment.begin(), init_segment.end()),
+        });
+    }
+
+    // Build catalog JSON
+    std::ostringstream catalog;
+    catalog << "{";
+    catalog << "\"version\":1,";
+    catalog << "\"format\":\"cmsf\",";
+    catalog << "\"tracks\":[";
+    bool first_track = true;
+    for (const auto& track : tracks) {
+        if (!first_track) {
+            catalog << ',';
+        }
+        first_track = false;
+
+        catalog << '{'
+                << "\"name\":\"" << local_json_escape(track.track_name) << "\","
+                << "\"id\":" << track.track_id << ','
+                << "\"role\":\"" << local_track_role(track.handler_type) << "\","
+                << "\"packaging\":\"" << local_json_escape(track.packaging) << "\","
+                << "\"renderGroup\":1,"
+                << "\"isLive\":" << (is_live ? "true" : "false");
+        if (!track.codec.empty()) {
+            catalog << ",\"codec\":\"" << local_json_escape(track.codec) << '"';
+        }
+        if (track.handler_type == "vide") {
+            catalog << ",\"width\":" << track.width
+                    << ",\"height\":" << track.height;
+            if (track.frame_rate > 0.0) {
+                catalog << std::fixed << std::setprecision(2) << ",\"frameRate\":" << track.frame_rate;
+            }
+        } else if (track.handler_type == "soun") {
+            catalog << ",\"sampleRate\":" << track.sample_rate
+                    << ",\"channelCount\":" << track.channel_count;
+        }
+        // Use full init segment for all tracks
+        catalog << ",\"initData\":\"" << full_init_base64 << '"';
+        catalog << '}';
+    }
+    catalog << "]}";
+
+    const std::string catalog_text = catalog.str();
+    result.catalog_payload = std::vector<std::uint8_t>(catalog_text.begin(), catalog_text.end());
+
+    return result;
+}
+
 }  // namespace openmoq::publisher
diff --git a/src/main.cpp b/src/main.cpp
index 8520d0d..4417ca7 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -51,26 +51,15 @@ int main(int argc, char** argv) {
 
     try {
         const CliOptions options = parse_cli_options(argc, argv);
-        const ParsedMp4 parsed_mp4 = options.input_source.kind == InputSourceKind::kStdin
-                                         ? parse_mp4_stream(std::cin, "stdin")
-                                         : parse_mp4_file(options.input_source.path.string());
-        const SegmentedMp4 segmented_mp4 = segment_for_cmaf(parsed_mp4,
-                                                            options.split_cmaf_chunks ? CmafObjectMode::kSplit
-                                                                                      : CmafObjectMode::kCoalesced);
-        const PublishPlan plan = build_publish_plan(segmented_mp4, options.draft_version, options.include_sap);
-
-        if (options.dump_plan || !options.emit_dir.has_value()) {
-            std::cout << render_publish_plan(plan);
-        }
 
-        if (options.emit_dir.has_value()) {
-            emit_plan_objects(plan, parsed_mp4.bytes, *options.emit_dir);
-        }
+        // Live stdin mode: when reading from stdin with an endpoint, use
+        // incremental streaming instead of buffering everything to EOF.
+        const bool live_stdin = options.input_source.kind == InputSourceKind::kStdin
+                                && options.endpoint.has_value();
 
-        if (options.endpoint.has_value()) {
+        if (live_stdin) {
             using namespace openmoq::publisher::transport;
 
-            const PublishPlan materialized_plan = materialize_publish_plan(plan, parsed_mp4.bytes);
             EndpointConfig endpoint = *options.endpoint;
             endpoint.application_protocol = endpoint.transport == transport::TransportKind::kWebTransport
                                                ? webtransport_protocol_offer(options.draft_version)
@@ -100,9 +89,65 @@ int main(int argc, char** argv) {
                 throw std::runtime_error("transport connect failed: " + status.message);
             }
 
-            status = session.publish(materialized_plan);
+            status = session.publish_live(std::cin, options.draft_version, options.split_cmaf_chunks);
             if (!status.ok) {
-                throw std::runtime_error("transport publish failed: " + status.message);
+                throw std::runtime_error("transport live publish failed: " + status.message);
+            }
+        } else {
+            // Original batch mode: read entire file/stdin, segment, plan, publish.
+            const ParsedMp4 parsed_mp4 = options.input_source.kind == InputSourceKind::kStdin
+                                             ? parse_mp4_stream(std::cin, "stdin")
+                                             : parse_mp4_file(options.input_source.path.string());
+            const SegmentedMp4 segmented_mp4 = segment_for_cmaf(parsed_mp4,
+                                                                options.split_cmaf_chunks ? CmafObjectMode::kSplit
+                                                                                          : CmafObjectMode::kCoalesced);
+            const PublishPlan plan = build_publish_plan(segmented_mp4, options.draft_version, options.include_sap);
+
+            if (options.dump_plan || !options.emit_dir.has_value()) {
+                std::cout << render_publish_plan(plan);
+            }
+
+            if (options.emit_dir.has_value()) {
+                emit_plan_objects(plan, parsed_mp4.bytes, *options.emit_dir);
+            }
+
+            if (options.endpoint.has_value()) {
+                using namespace openmoq::publisher::transport;
+
+                const PublishPlan materialized_plan = materialize_publish_plan(plan, parsed_mp4.bytes);
+                EndpointConfig endpoint = *options.endpoint;
+                endpoint.application_protocol = endpoint.transport == transport::TransportKind::kWebTransport
+                                                   ? webtransport_protocol_offer(options.draft_version)
+                                                   : default_alpn(options.draft_version);
+                if (!options.endpoint_alpn_overridden && endpoint.transport == transport::TransportKind::kRawQuic &&
+                    options.draft_version != DraftVersion::kDraft14) {
+                    endpoint.alpn = default_alpn(options.draft_version);
+                } else if (!options.endpoint_alpn_overridden &&
+                           endpoint.transport == transport::TransportKind::kWebTransport) {
+                    endpoint.alpn = "h3";
+                }
+                auto transport = create_transport(endpoint.transport);
+                if (!transport) {
+                    throw std::runtime_error("failed to create requested transport");
+                }
+                MoqtSession session(
+                    *transport,
+                    options.track_namespace,
+                    options.forward,
+                    options.publish_catalog,
+                    options.paced,
+                    options.loop,
+                    options.subscriber_timeout);
+
+                TransportStatus status = session.connect(endpoint, options.tls);
+                if (!status.ok) {
+                    throw std::runtime_error("transport connect failed: " + status.message);
+                }
+
+                status = session.publish(materialized_plan);
+                if (!status.ok) {
+                    throw std::runtime_error("transport publish failed: " + status.message);
+                }
             }
         }
     } catch (const std::exception& exception) {
diff --git a/src/mp4_box.cpp b/src/mp4_box.cpp
index 6c1f601..100470d 100644
--- a/src/mp4_box.cpp
+++ b/src/mp4_box.cpp
@@ -756,4 +756,80 @@ std::span<const std::uint8_t> slice_bytes(std::span<const std::uint8_t> bytes, c
     return bytes.subspan(span.offset, span.size);
 }
 
+// --- StreamingMp4Reader ---
+
+void StreamingMp4Reader::append(const std::uint8_t* data, std::size_t len) {
+    buffer_.insert(buffer_.end(), data, data + len);
+}
+
+std::size_t StreamingMp4Reader::read_from(std::istream& input, std::size_t chunk_size) {
+    const std::size_t old_size = buffer_.size();
+    buffer_.resize(old_size + chunk_size);
+    input.read(reinterpret_cast<char*>(buffer_.data() + old_size),
+               static_cast<std::streamsize>(chunk_size));
+    const auto bytes_read = static_cast<std::size_t>(input.gcount());
+    buffer_.resize(old_size + bytes_read);
+    return bytes_read;
+}
+
+std::optional<StreamingBoxResult> StreamingMp4Reader::next_box() {
+    const std::size_t avail = buffer_.size() - consumed_;
+    if (avail < 8) {
+        return std::nullopt;
+    }
+
+    const std::uint8_t* p = buffer_.data() + consumed_;
+
+    const std::uint32_t small_size =
+        (static_cast<std::uint32_t>(p[0]) << 24U) |
+        (static_cast<std::uint32_t>(p[1]) << 16U) |
+        (static_cast<std::uint32_t>(p[2]) << 8U) |
+        static_cast<std::uint32_t>(p[3]);
+
+    std::string type(reinterpret_cast<const char*>(p + 4), 4);
+
+    std::uint64_t box_size = small_size;
+    if (small_size == 1) {
+        if (avail < 16) {
+            return std::nullopt;
+        }
+        box_size = 0;
+        for (int i = 0; i < 8; ++i) {
+            box_size = (box_size << 8U) | p[8 + i];
+        }
+        if (box_size < 16) {
+            throw std::runtime_error("impossible extended box size");
+        }
+    } else if (small_size == 0) {
+        // size==0 means "runs to EOF"; can't handle incrementally unless
+        // we have all remaining data. Return nullopt to wait for more data.
+        return std::nullopt;
+    } else if (small_size < 8) {
+        throw std::runtime_error("impossible MP4 box size");
+    }
+
+    if (avail < box_size) {
+        return std::nullopt;
+    }
+
+    StreamingBoxResult result;
+    result.type = std::move(type);
+    result.bytes.assign(p, p + static_cast<std::size_t>(box_size));
+    consumed_ += static_cast<std::size_t>(box_size);
+
+    if (consumed_ > 64 * 1024) {
+        compact();
+    }
+
+    return result;
+}
+
+void StreamingMp4Reader::compact() {
+    if (consumed_ == 0) {
+        return;
+    }
+    buffer_.erase(buffer_.begin(), buffer_.begin() + static_cast<std::ptrdiff_t>(consumed_));
+    consumed_ = 0;
+}
+
 }  // namespace openmoq::publisher
diff --git a/src/transport/moqt_session.cpp b/src/transport/moqt_session.cpp
index 124a6c6..b78a9e7 100644
--- a/src/transport/moqt_session.cpp
+++ b/src/transport/moqt_session.cpp
@@ -1,5 +1,7 @@
 #include "openmoq/publisher/transport/moqt_session.h"
 #include "openmoq/publisher/transport/moqt_control_messages.h"
+#include "openmoq/publisher/cmaf_segmenter.h"
+#include "openmoq/publisher/mp4_box.h"
 
 #include <algorithm>
 #include <chrono>
@@ -1089,6 +1091,18 @@ class SubgroupSenderState {
 
     std::uint64_t stream_count() const { return stream_count_; }
 
+    // FIN all currently open streams (used when transitioning to a new group).
+    TransportStatus finish_group(PublisherTransport& transport) {
+        for (auto& [key, stream] : streams_) {
+            TransportStatus status = transport.write_stream(stream.stream_id, {}, true);
+            if (!status.ok) {
+                return status;
+            }
+        }
+        streams_.clear();
+        return TransportStatus::success();
+    }
+
 private:
     struct Key {
         std::uint64_t group_id;
@@ -2173,6 +2187,482 @@ TransportStatus MoqtSession::publish(const openmoq::publisher::PublishPlan& plan
                                pending_control_bytes_);
 }
 
+TransportStatus MoqtSession::publish_live(std::istream& input,
+                                           openmoq::publisher::DraftVersion draft_version,
+                                           bool /*split_cmaf_chunks*/) {
+    if (transport_.state() != ConnectionState::kConnected) {
+        return TransportStatus::failure("transport is not connected");
+    }
+
+    TransportStatus status = ensure_setup(draft_version);
+    if (!status.ok) {
+        return status;
+    }
+    std::cout << "connection_id=" << transport_.connection_id() << '\n' << std::flush;
+
+    // Phase 1: Read stdin until we have ftyp + moov (track discovery).
+    openmoq::publisher::StreamingMp4Reader reader;
+    std::vector<std::uint8_t> ftyp_bytes;
+    std::vector<std::uint8_t> moov_bytes;
+    std::vector<openmoq::publisher::TrackDescription> tracks;
+    std::vector<openmoq::publisher::Mp4Box> init_top_level_boxes;
+
+    std::cerr << "[moqt-session] live: waiting for ftyp+moov from stdin...\n";
+
+    while (ftyp_bytes.empty() || moov_bytes.empty()) {
+        const std::size_t bytes_read = reader.read_from(input);
+        if (bytes_read == 0 && ftyp_bytes.empty()) {
+            return TransportStatus::failure("stdin EOF before ftyp box");
+        }
+        if (bytes_read == 0 && moov_bytes.empty()) {
+            return TransportStatus::failure("stdin EOF before moov box");
+        }
+
+        while (auto box = reader.next_box()) {
+            if (box->type == "ftyp") {
+                ftyp_bytes = std::move(box->bytes);
+            } else if (box->type == "moov") {
+                moov_bytes = std::move(box->bytes);
+                break;  // Stop consuming boxes; remaining data is for Phase 4.
+            }
+            // Skip other pre-moov boxes (e.g. free, styp)
+        }
+    }
+
+    // Build initialization segment (ftyp + moov)
+    std::vector<std::uint8_t> init_segment;
+    init_segment.reserve(ftyp_bytes.size() + moov_bytes.size());
+    init_segment.insert(init_segment.end(), ftyp_bytes.begin(), ftyp_bytes.end());
+    init_segment.insert(init_segment.end(), moov_bytes.begin(), moov_bytes.end());
+
+    init_top_level_boxes = openmoq::publisher::parse_mp4_boxes(init_segment);
+    tracks = openmoq::publisher::extract_tracks(init_top_level_boxes, init_segment);
+    if (tracks.empty()) {
+        return TransportStatus::failure("no tracks found in moov box");
+    }
+
+    std::cerr << "[moqt-session] live: discovered " << tracks.size() << " track(s): "
+              << openmoq::publisher::summarize_tracks(tracks) << '\n';
+
+    // Build catalog and init data
+    openmoq::publisher::LiveCatalog live_catalog =
+        openmoq::publisher::build_live_catalog(tracks, init_segment, true);
+
+    // Phase 2: Publish namespace + PUBLISH all tracks
+    NamespaceMessage namespace_message{
+        .draft = draft_version,
+        .track_namespace = track_namespace_,
+        .request_id = 0,
+    };
+    status = write_frame(control_stream_id_, encode_namespace_message(namespace_message), false);
+    if (!status.ok) {
+        return status;
+    }
+    status = collect_control_acknowledgements(
+        transport_, control_stream_id_, draft_version, 1, 0, pending_control_bytes_);
+    if (!status.ok) {
+        return status;
+    }
+
+    // Build track alias map (self-assigned, matching legacy serve_subscriptions behavior)
+    std::map<std::string, std::uint64_t> alias_by_track;
+    std::uint64_t next_alias = 0;
+    alias_by_track.emplace("catalog", next_alias++);
+    for (const auto& track : tracks) {
+        alias_by_track.emplace(track.track_name, next_alias++);
+    }
+
+    std::cerr << "[moqt-session] live: awaiting subscriptions, mode="
+              << (auto_forward_ ? "forward" : "await-subscribe") << '\n';
+
+    // Phase 3: Catalog will be sent when a SUBSCRIBE arrives for it.
+    // The relay forwards SUBSCRIBE to us; we respond with SUBSCRIBE_OK,
+    // send data, then PUBLISH_DONE -- matching the legacy serve_subscriptions flow.
+    bool catalog_sent = false;
+
+    auto send_catalog = [&](std::uint64_t track_alias) -> TransportStatus {
+        if (catalog_sent) {
+            return TransportStatus::success();
+        }
+        const openmoq::publisher::CmsfObject catalog_object{
+            .kind = openmoq::publisher::CmsfObjectKind::kInitialization,
+            .track_name = "catalog",
+            .group_id = 0,
+            .subgroup_id = 0,
+            .object_id = 0,
+            .media_time_us = 0,
+            .media_duration_us = 0,
+            .payload = {},
+            .owned_payload = live_catalog.catalog_payload,
+        };
+        SubgroupSenderState catalog_sender;
+        TransportStatus cat_status = catalog_sender.serve(
+            transport_, draft_version, track_alias, 0,
+            catalog_object, true, true,
+            std::span<const std::uint8_t>(live_catalog.catalog_payload));
+        if (!cat_status.ok) {
+            return cat_status;
+        }
+        catalog_sent = true;
+        std::cerr << "[moqt-session] live: catalog published (" << live_catalog.catalog_payload.size() << " bytes)\n";
+        return TransportStatus::success();
+    };
+
+    // Phase 4: Stream media from stdin.
+    // Use a reader thread so we can also handle control messages.
+    struct LiveMediaQueue {
+        std::mutex mutex;
+        std::deque<openmoq::publisher::MediaFragment> fragments;
+        bool eof = false;
+    };
+    auto queue = std::make_shared<LiveMediaQueue>();
+
+    std::thread stdin_thread([&reader, &input, &tracks, queue]() {
+        std::vector<std::uint8_t> pending_moof;
+        std::size_t shared_group_id = 0;
+        std::map<std::string, std::size_t> object_id_in_group;  // per track, resets on new group
+        bool first_keyframe_seen = false;
+
+        while (true) {
+            const std::size_t bytes_read = reader.read_from(input);
+
+            while (auto box = reader.next_box()) {
+                if (box->type == "moof") {
+                    pending_moof = std::move(box->bytes);
+                } else if (box->type == "mdat") {
+                    if (pending_moof.empty()) {
+                        std::cerr << "[moqt-session] live: mdat without preceding moof, skipping\n";
+                        continue;
+                    }
+                    try {
+                        // Build fragment (group_id=0 placeholder, we'll assign below)
+                        auto fragment = openmoq::publisher::build_live_fragment(
+                            pending_moof, box->bytes, tracks, 0);
+
+                        // Keyframe-based grouping:
+                        // When a video keyframe arrives, start a new group for ALL tracks.
+                        if (fragment.is_video_keyframe) {
+                            if (first_keyframe_seen) {
+                                ++shared_group_id;
+                            }
+                            first_keyframe_seen = true;
+                            // Reset object counters for all tracks on new group
+                            object_id_in_group.clear();
+                        }
+
+                        if (!first_keyframe_seen) {
+                            // Drop fragments before first keyframe (can't decode without IDR)
+                            pending_moof.clear();
+                            continue;
+                        }
+
+                        // Assign shared group_id and per-track object_id
+                        fragment.group_id = shared_group_id;
+                        fragment.object_id = object_id_in_group[fragment.track_name]++;
+
+                        {
+                            std::lock_guard<std::mutex> lock(queue->mutex);
+                            queue->fragments.push_back(std::move(fragment));
+                            // Trim queue: keep only fragments from the latest 2 groups.
+                            // This prevents unbounded backlog when ffmpeg encodes
+                            // faster than realtime, while keeping enough data for
+                            // A/V sync (audio from the previous group).
+                            if (!queue->fragments.empty()) {
+                                const std::size_t latest = queue->fragments.back().group_id;
+                                const std::size_t min_keep = latest > 1 ? latest - 1 : 0;
+                                while (!queue->fragments.empty() &&
+                                       queue->fragments.front().group_id < min_keep) {
+                                    queue->fragments.pop_front();
+                                }
+                            }
+                        }
+                    } catch (const std::exception& e) {
+                        std::cerr << "[moqt-session] live: fragment parse error: " << e.what() << '\n';
+                    }
+                    pending_moof.clear();
+                }
+                // Skip other box types (styp, free, etc.)
+            }
+
+            if (bytes_read == 0) {
+                std::lock_guard<std::mutex> lock(queue->mutex);
+                queue->eof = true;
+                break;
+            }
+        }
+    });
+
+    // Main loop: drain queue and publish fragments
+    std::map<std::string, SubgroupSenderState> sender_by_track;
+    std::map<std::uint64_t, SubscribeMessage> active_subscriptions;
+    std::set<std::string> subscribed_tracks;
+
+    auto drain_queue = [&]() -> TransportStatus {
+        // Send all available fragments. The queue is bounded by stdin_thread's
+        // trim logic (keeps at most the latest 2 groups), so we never burst
+        // more than ~4 fragments (2 groups × 2 tracks) at a time.
+        // No pacing needed: trimming prevents backlog accumulation and
+        // ffmpeg's realtime encoding rate naturally limits throughput.
+        while (true) {
+            openmoq::publisher::MediaFragment fragment;
+            {
+                std::lock_guard<std::mutex> lock(queue->mutex);
+                if (queue->fragments.empty()) break;
+                fragment = std::move(queue->fragments.front());
+                queue->fragments.pop_front();
+            }
+
+            // Only send data for tracks that have an active subscription.
+            if (!subscribed_tracks.count(fragment.track_name)) {
+                continue;
+            }
+
+            const auto alias_it = alias_by_track.find(fragment.track_name);
+            if (alias_it == alias_by_track.end()) {
+                continue;
+            }
+
+            const openmoq::publisher::CmsfObject object{
+                .kind = openmoq::publisher::CmsfObjectKind::kMedia,
+                .track_name = fragment.track_name,
+                .group_id = fragment.group_id,
+                .subgroup_id = 0,
+                .object_id = fragment.object_id,
+                .media_time_us = fragment.start_time_us,
+                .media_duration_us = fragment.duration_us,
+                .payload = {},
+                .owned_payload = fragment.payload.owned_bytes,
+            };
+
+            const std::uint64_t send_seq = next_send_seq();
+            const std::span<const std::uint8_t> payload(fragment.payload.owned_bytes);
+
+            // Each fragment is a single object; FIN the stream immediately.
+            // With keyframe-based grouping each group typically has only 1 object
+            // per track (since GOP interval = fragment duration).
+            TransportStatus write_status = sender_by_track[fragment.track_name].serve(
+                transport_, draft_version, alias_it->second, send_seq,
+                object, true, true, payload);
+            if (!write_status.ok) {
+                return write_status;
+            }
+            std::cerr << "[moqt-session] live: sent track=" << fragment.track_name
+                      << " group=" << fragment.group_id
+                      << " obj=" << fragment.object_id
+                      << " time_us=" << fragment.start_time_us
+                      << " bytes=" << fragment.payload.owned_bytes.size()
+                      << " sap=" << static_cast<unsigned int>(fragment.sap_type) << '\n';
+        }
+        return TransportStatus::success();
+    };
+
+    // Helper: process pending SUBSCRIBE/SUBSCRIBE_NAMESPACE messages
+    // from the control stream. Returns the number of new subscriptions accepted.
+    auto process_control_messages = [&]() -> std::pair<TransportStatus, std::size_t> {
+        std::size_t new_subs = 0;
+        std::size_t message_size = 0;
+        while (next_control_message(pending_control_bytes_, draft_version, message_size)) {
+            const std::vector<std::uint8_t> message_bytes(
+                pending_control_bytes_.begin(),
+                pending_control_bytes_.begin() + static_cast<std::ptrdiff_t>(message_size));
+            std::size_t offset = 0;
+            std::uint64_t message_type = 0;
+            if (!decode_varint(message_bytes, offset, message_type)) {
+                return {TransportStatus::failure("failed to parse control request type"), 0};
+            }
+            trace_control_message(message_bytes, draft_version);
+
+            if (message_type == 0x03) {  // SUBSCRIBE
+                SubscribeMessage subscribe;
+                if (!decode_subscribe_message(message_bytes, draft_version, subscribe)) {
+                    return {TransportStatus::failure("received invalid SUBSCRIBE"), 0};
+                }
+
+                const auto track_it = alias_by_track.find(subscribe.track_name);
+                if (track_it == alias_by_track.end()) {
+                    auto ws = transport_.write_stream(control_stream_id_,
+                        encode_subscribe_error_message(subscribe.request_id, 0x2, "track does not exist"), false);
+                    if (!ws.ok) {
+                        return {ws, 0};
+                    }
+                } else {
+                    auto ws = transport_.write_stream(control_stream_id_,
+                        encode_subscribe_ok_message(draft_version, subscribe.request_id,
+                                                    track_it->second, 0, 0, false), false);
+                    if (!ws.ok) {
+                        return {ws, 0};
+                    }
+                    active_subscriptions.emplace(subscribe.request_id, subscribe);
+                    ++new_subs;
+                    std::cerr << "[moqt-session] live: accepted subscribe track=" << subscribe.track_name
+                              << " request_id=" << subscribe.request_id << '\n';
+
+                    // Catalog is a one-shot track: send data + PUBLISH_DONE immediately
+                    if (subscribe.track_name == "catalog") {
+                        ws = send_catalog(track_it->second);
+                        if (!ws.ok) {
+                            return {ws, 0};
+                        }
+                        ws = transport_.write_stream(control_stream_id_,
+                            encode_publish_done_message(subscribe.request_id, 1), false);
+                        if (!ws.ok) {
+                            return {ws, 0};
+                        }
+                    } else {
+                        // Only add media tracks to subscribed_tracks (not catalog).
+                        // This gates drain_queue until media is actually subscribed.
+                        subscribed_tracks.insert(subscribe.track_name);
+                    }
+                }
+            } else if (message_type == 0x11) {  // SUBSCRIBE_NAMESPACE
+                SubscribeNamespaceMessage subscribe_namespace;
+                if (decode_subscribe_namespace_message(message_bytes, draft_version, subscribe_namespace)) {
+                    auto ws = transport_.write_stream(control_stream_id_,
+                        encode_subscribe_namespace_ok_message(draft_version, subscribe_namespace.request_id), false);
+                    if (!ws.ok) {
+                        return {ws, 0};
+                    }
+                }
+            }
+            // Skip other message types
+
+            pending_control_bytes_.erase(
+                pending_control_bytes_.begin(),
+                pending_control_bytes_.begin() + static_cast<std::ptrdiff_t>(message_size));
+        }
+        return {TransportStatus::success(), new_subs};
+    };
+
+    if (auto_forward_) {
+        // Forward mode: wait for subscriptions, then stream objects as they arrive
+        while (true) {
+            // Only drain media after at least one non-catalog track is subscribed
+            if (!subscribed_tracks.empty()) {
+                status = drain_queue();
+                if (!status.ok) {
+                    stdin_thread.join();
+                    return status;
+                }
+            }
+
+            bool is_eof;
+            {
+                std::lock_guard<std::mutex> lock(queue->mutex);
+                is_eof = queue->eof && queue->fragments.empty();
+            }
+            if (is_eof && !subscribed_tracks.empty()) {
+                break;
+            }
+
+            // Brief sleep to avoid busy-waiting
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+
+            // Read and process control messages (SUBSCRIBE, SUBSCRIBE_NAMESPACE)
+            std::vector<std::uint8_t> chunk;
+            bool fin = false;
+            const TransportStatus read_status =
+                transport_.read_stream(control_stream_id_, chunk, fin, std::chrono::milliseconds(0));
+            if (read_status.ok && !chunk.empty()) {
+                pending_control_bytes_.insert(pending_control_bytes_.end(), chunk.begin(), chunk.end());
+            }
+            auto [ctrl_status, new_subs] = process_control_messages();
+            if (!ctrl_status.ok) {
+                stdin_thread.join();
+                return ctrl_status;
+            }
+        }
+    } else {
+        // Await-subscribe mode: wait for subscriptions, then stream
+        bool fin = false;
+
+        while (true) {
+            bool is_eof;
+            {
+                std::lock_guard<std::mutex> lock(queue->mutex);
+                is_eof = queue->eof && queue->fragments.empty();
+            }
+            if (is_eof && !active_subscriptions.empty()) {
+                // Drain remaining
+                status = drain_queue();
+                if (!status.ok) {
+                    stdin_thread.join();
+                    return status;
+                }
+                break;
+            }
+
+            // Read control messages
+            std::vector<std::uint8_t> chunk;
+            bool immediate_fin = false;
+            const auto read_timeout = active_subscriptions.empty()
+                                          ? subscriber_timeout_
+                                          : std::chrono::milliseconds(0);
+            const TransportStatus read_status =
+                transport_.read_stream(control_stream_id_, chunk, immediate_fin, read_timeout);
+
+            if (read_status.ok) {
+                pending_control_bytes_.insert(pending_control_bytes_.end(), chunk.begin(), chunk.end());
+                fin = immediate_fin;
+            } else if (read_status.message == "timed out waiting for stream data" ||
+                       read_status.message == "no queued read for stream") {
+                if (active_subscriptions.empty()) {
+                    std::cerr << "[moqt-session] live: no subscribers before timeout\n";
+                    break;
+                }
+            } else {
+                stdin_thread.join();
+                return read_status;
+            }
+
+            // Process control messages
+            auto [ctrl_status, new_subs] = process_control_messages();
+            if (!ctrl_status.ok) {
+                stdin_thread.join();
+                return ctrl_status;
+            }
+
+            if (fin) {
+                break;
+            }
+
+            // Drain any available media from the queue
+            if (!subscribed_tracks.empty()) {
+                status = drain_queue();
+                if (!status.ok) {
+                    stdin_thread.join();
+                    return status;
+                }
+            }
+
+            if (active_subscriptions.empty()) {
+                continue;
+            }
+
+            // Small sleep to prevent busy-waiting
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+    }
+
+    stdin_thread.join();
+
+    // Send PUBLISH_DONE for each subscribed media track (catalog already handled)
+    for (const auto& [request_id, subscribe] : active_subscriptions) {
+        if (subscribe.track_name == "catalog") {
+            continue;  // Already sent PUBLISH_DONE for catalog
+        }
+        status = transport_.write_stream(control_stream_id_,
+            encode_publish_done_message(request_id, sender_by_track[subscribe.track_name].stream_count()), false);
+        if (!status.ok) {
+            return status;
+        }
+    }
+
+    std::cerr << "[moqt-session] live: stdin EOF, publishing complete\n";
+
+    return transport_.write_stream(control_stream_id_,
+        encode_publish_namespace_done_message(namespace_message), false);
+}
+
 TransportStatus MoqtSession::close(std::uint64_t application_error_code) {
     control_stream_open_ = false;
     control_stream_id_ = 0;
@@ -2312,7 +2802,10 @@ TransportStatus MoqtSession::ensure_setup(openmoq::publisher::DraftVersion draft
                 return TransportStatus::failure("received invalid SERVER_SETUP message");
             }
             saw_server_setup = true;
-            if (draft == openmoq::publisher::DraftVersion::kDraft14) {
+            // Accept max_request_id from SERVER_SETUP parameters for both
+            // draft-14 and draft-16. Some relays include it here rather than
+            // (or in addition to) sending a separate MAX_REQUEST_ID message.
+            if (server_setup.max_request_id != 0) {
                 peer_max_request_id_ = server_setup.max_request_id;
             }
         } else if (draft == openmoq::publisher::DraftVersion::kDraft16 && message_type == 0x15) {