feat: add streamer-owned recording control service

Introduce a dedicated streamer-side recording control plane instead of sharing the producer recorder API.

- register streamer-owned recorder endpoints as a NATS micro service
- add explicit MP4 and MCAP recorder control protobufs and subject helpers
- wire recorder lifecycle handling into the pipeline runtime
- add MP4 writer and depth-alignment support files used by the new recording flow

src/pipeline/pipeline_runtime.cpp

@@ -3,13 +3,16 @@
 #include "cvmmap_streamer/encode/encoder_backend.hpp"
 #include "cvmmap_streamer/ipc/contracts.hpp"
 #include "cvmmap_streamer/metrics/latency_tracker.hpp"
+#include "cvmmap_streamer/protocol/nats_request_reply_server.hpp"
 #include "cvmmap_streamer/protocol/rtmp_output.hpp"
 #include "cvmmap_streamer/protocol/rtp_publisher.hpp"
+#include "cvmmap_streamer/protocol/streamer_subjects.hpp"
 #include "cvmmap_streamer/record/mcap_record_sink.hpp"
+#include "cvmmap_streamer/record/mp4_record_writer.hpp"
+#include "proto/cvmmap_streamer/recorder_control.pb.h"
 
 #include <cvmmap/client.hpp>
 #include <cvmmap/nats_client.hpp>
-#include <cvmmap/nats_service.hpp>
 #include <cvmmap/parser.hpp>
 
 #include <chrono>
@@ -19,6 +22,7 @@
 #include <deque>
 #include <exception>
 #include <expected>
+#include <filesystem>
 #include <mutex>
 #include <optional>
 #include <span>
@@ -46,6 +50,7 @@ namespace cvmmap_streamer::core {
 namespace {
 
 namespace ipc = cvmmap_streamer::ipc;
+namespace recorder_pb = cvmmap_streamer::proto;
 
 enum class PipelineExitCode : int {
 	Success = 0,
@@ -418,23 +423,58 @@ std::uint64_t body_tracking_timestamp_ns(const cvmmap::body_tracking_frame_t &fr
 	return frame.header.sdk_timestamp_ns;
 }
 
+[[nodiscard]]
+float stream_fps(const encode::EncodedStreamInfo &stream_info) {
+	if (stream_info.frame_rate_num == 0 || stream_info.frame_rate_den == 0) {
+		return 30.0f;
+	}
+	return static_cast<float>(stream_info.frame_rate_num) /
+		   static_cast<float>(stream_info.frame_rate_den);
+}
+
 struct McapRecorderState {
 	mutable std::mutex mutex{};
 	RuntimeConfig base_config{};
 	std::optional<RuntimeConfig> active_record_config{};
 	std::optional<encode::EncodedStreamInfo> current_stream_info{};
 	std::optional<record::McapRecordSink> sink{};
-	cvmmap::RecordingStatus status{
-		.format = cvmmap::RecordingFormat::Mcap,
-		.can_record = true,
-	};
+	struct Status {
+		bool can_record{true};
+		bool is_recording{false};
+		bool last_frame_ok{false};
+		std::uint32_t frames_ingested{0};
+		std::uint32_t frames_encoded{0};
+		std::string active_path{};
+		std::string error_message{};
+	} status{};
+};
+
+struct Mp4RecorderStatus {
+	bool can_record{false};
+	bool is_recording{false};
+	bool last_frame_ok{false};
+	std::uint32_t frames_ingested{0};
+	std::uint32_t frames_encoded{0};
+	std::string active_path{};
+	std::string error_message{};
+};
+
+struct Mp4RecorderState {
+	mutable std::mutex mutex{};
+	RuntimeConfig base_config{};
+	std::optional<ipc::FrameInfo> current_frame_info{};
+	std::optional<record::Mp4InputPixelFormat> current_input_pixel_format{};
+	float current_fps{30.0f};
+	std::optional<record::Mp4RecordWriter> writer{};
+	std::optional<std::uint64_t> first_frame_timestamp_ns{};
+	Mp4RecorderStatus status{};
 };
 
 [[nodiscard]]
-cvmmap::ControlError make_recording_control_error(
-	const int32_t code,
+protocol::RpcError make_recorder_rpc_error(
+	const protocol::RpcErrorCode code,
 	std::string message) {
-	return cvmmap::ControlError{
+	return protocol::RpcError{
 		.code = code,
 		.message = std::move(message),
 	};
@@ -443,60 +483,128 @@ cvmmap::ControlError make_recording_control_error(
 [[nodiscard]]
 RuntimeConfig make_mcap_record_config(
 	const RuntimeConfig &base_config,
-	const cvmmap::RecordingRequest &request) {
+	const recorder_pb::McapStartRequest &request) {
 	auto record_config = base_config;
 	record_config.record.mcap.enabled = true;
-	record_config.record.mcap.path = request.output_path;
-	if (request.mcap_options) {
-		if (request.mcap_options->topic) {
-			record_config.record.mcap.topic = *request.mcap_options->topic;
-		}
-		if (request.mcap_options->depth_topic) {
-			record_config.record.mcap.depth_topic = *request.mcap_options->depth_topic;
-		}
-		if (request.mcap_options->body_topic) {
-			record_config.record.mcap.body_topic = *request.mcap_options->body_topic;
-		}
-		if (request.mcap_options->frame_id) {
-			record_config.record.mcap.frame_id = *request.mcap_options->frame_id;
-		}
+	record_config.record.mcap.path = request.output_path();
+	if (request.has_topic()) {
+		record_config.record.mcap.topic = request.topic();
+	}
+	if (request.has_depth_topic()) {
+		record_config.record.mcap.depth_topic = request.depth_topic();
+	}
+	if (request.has_body_topic()) {
+		record_config.record.mcap.body_topic = request.body_topic();
+	}
+	if (request.has_frame_id()) {
+		record_config.record.mcap.frame_id = request.frame_id();
 	}
 	return record_config;
 }
 
-void reset_mcap_status_after_stop(cvmmap::RecordingStatus &status) {
+[[nodiscard]]
+record::Mp4EncodeTuning make_mp4_encode_tuning(const RuntimeConfig &base_config) {
+	return record::Mp4EncodeTuning{
+		.quality = record::kDefaultMp4Quality,
+		.gop = base_config.encoder.gop,
+		.b_frames = base_config.encoder.b_frames,
+	};
+}
+
+[[nodiscard]]
+std::expected<record::Mp4InputPixelFormat, std::string> mp4_input_pixel_format(
+	const ipc::FrameInfo &frame_info) {
+	if (frame_info.depth != ipc::Depth::U8) {
+		return std::unexpected("MP4 recorder requires 8-bit color frames");
+	}
+	switch (frame_info.pixel_format) {
+	case ipc::PixelFormat::BGR:
+		return record::Mp4InputPixelFormat::Bgr24;
+	case ipc::PixelFormat::RGB:
+		return record::Mp4InputPixelFormat::Rgb24;
+	case ipc::PixelFormat::BGRA:
+		return record::Mp4InputPixelFormat::Bgra32;
+	case ipc::PixelFormat::RGBA:
+		return record::Mp4InputPixelFormat::Rgba32;
+	case ipc::PixelFormat::GRAY:
+		return record::Mp4InputPixelFormat::Gray8;
+	case ipc::PixelFormat::YUV:
+	case ipc::PixelFormat::YUYV:
+		return std::unexpected("MP4 recorder does not support packed YUV snapshot frames");
+	}
+	return std::unexpected("MP4 recorder does not support the snapshot pixel format");
+}
+
+void reset_mcap_status_after_stop(McapRecorderState::Status &status) {
+	status.is_recording = false;
+	status.active_path.clear();
+}
+
+void reset_mp4_status_after_stop(Mp4RecorderStatus &status) {
 	status.is_recording = false;
-	status.is_paused = false;
 	status.active_path.clear();
 }
 
 [[nodiscard]]
-std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> start_mcap_recording(
+recorder_pb::McapRecorderState to_proto_mcap_state(
+	const McapRecorderState::Status &status) {
+	recorder_pb::McapRecorderState wire_status;
+	wire_status.set_can_record(status.can_record);
+	wire_status.set_is_recording(status.is_recording);
+	wire_status.set_last_frame_ok(status.last_frame_ok);
+	wire_status.set_frames_ingested(status.frames_ingested);
+	wire_status.set_frames_encoded(status.frames_encoded);
+	wire_status.set_active_path(status.active_path);
+	wire_status.set_error_message(status.error_message);
+	return wire_status;
+}
+
+template <class Response>
+[[nodiscard]]
+Response make_ok_mcap_response(const McapRecorderState::Status &status) {
+	Response response;
+	response.set_code(recorder_pb::RPC_CODE_OK);
+	*response.mutable_state() = to_proto_mcap_state(status);
+	return response;
+}
+
+[[nodiscard]]
+std::expected<McapRecorderState::Status, protocol::RpcError> start_mcap_recording(
 	McapRecorderState &recorder_state,
-	const cvmmap::RecordingRequest &request) {
+	const recorder_pb::McapStartRequest &request) {
 	std::lock_guard lock(recorder_state.mutex);
-	if (request.format != cvmmap::RecordingFormat::Mcap) {
-		return std::unexpected(make_recording_control_error(
-			cvmmap::CONTROL_RESPONSE_UNSUPPORTED,
-			"recording format is not supported by the streamer"));
-	}
 	if (!recorder_state.current_stream_info) {
-		return std::unexpected(make_recording_control_error(
-			cvmmap::CONTROL_RESPONSE_ERROR,
+		return std::unexpected(make_recorder_rpc_error(
+			protocol::RpcErrorCode::Internal,
 			"MCAP recorder is not ready; stream info unavailable"));
 	}
 	if (recorder_state.sink) {
-		return std::unexpected(make_recording_control_error(
-			cvmmap::CONTROL_RESPONSE_ERROR,
+		return std::unexpected(make_recorder_rpc_error(
+			protocol::RpcErrorCode::Busy,
 			"MCAP recording is already active"));
 	}
+	if (request.output_path().empty()) {
+		return std::unexpected(make_recorder_rpc_error(
+			protocol::RpcErrorCode::InvalidRequest,
+			"output_path must not be empty"));
+	}
+	const auto output_path = std::filesystem::path(request.output_path());
+	std::error_code mkdir_error{};
+	if (output_path.has_parent_path()) {
+		std::filesystem::create_directories(output_path.parent_path(), mkdir_error);
+		if (mkdir_error) {
+			return std::unexpected(make_recorder_rpc_error(
+				protocol::RpcErrorCode::Internal,
+				"failed to create MCAP output directory: " + mkdir_error.message()));
+		}
+	}
 
 	auto record_config = make_mcap_record_config(recorder_state.base_config, request);
-	if (request.mcap_options && request.mcap_options->compression) {
-		auto parsed = parse_mcap_compression(*request.mcap_options->compression);
+	if (request.has_compression()) {
+		auto parsed = parse_mcap_compression(request.compression());
 		if (!parsed) {
-			return std::unexpected(make_recording_control_error(
-				cvmmap::CONTROL_RESPONSE_INVALID_PAYLOAD,
+			return std::unexpected(make_recorder_rpc_error(
+				protocol::RpcErrorCode::InvalidRequest,
 				parsed.error()));
 		}
 		record_config.record.mcap.compression = *parsed;
@@ -504,8 +612,8 @@ std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> start_mcap_recordin
 
 	auto created = record::McapRecordSink::create(record_config, *recorder_state.current_stream_info);
 	if (!created) {
-		return std::unexpected(make_recording_control_error(
-			cvmmap::CONTROL_RESPONSE_ERROR,
+		return std::unexpected(make_recorder_rpc_error(
+			protocol::RpcErrorCode::Internal,
 			"pipeline MCAP sink init failed: " + created.error()));
 	}
 
@@ -513,35 +621,278 @@ std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> start_mcap_recordin
 	recorder_state.sink.emplace(std::move(*created));
 	recorder_state.status.can_record = true;
 	recorder_state.status.is_recording = true;
-	recorder_state.status.is_paused = false;
 	recorder_state.status.last_frame_ok = true;
 	recorder_state.status.frames_ingested = 0;
 	recorder_state.status.frames_encoded = 0;
-	recorder_state.status.active_path = request.output_path;
+	recorder_state.status.active_path = request.output_path();
+	recorder_state.status.error_message.clear();
 	return recorder_state.status;
 }
 
 [[nodiscard]]
-std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> stop_mcap_recording(
-	McapRecorderState &recorder_state) {
+std::expected<McapRecorderState::Status, protocol::RpcError> stop_mcap_recording(
+	McapRecorderState &recorder_state,
+	const recorder_pb::McapStopRequest &) {
 	std::lock_guard lock(recorder_state.mutex);
 	if (recorder_state.sink) {
 		recorder_state.sink->close();
 		recorder_state.sink.reset();
 	}
 	recorder_state.active_record_config.reset();
+	recorder_state.status.last_frame_ok = true;
+	recorder_state.status.error_message.clear();
 	reset_mcap_status_after_stop(recorder_state.status);
 	return recorder_state.status;
 }
 
 [[nodiscard]]
-std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> get_mcap_recording_status(
-	McapRecorderState &recorder_state) {
+std::expected<McapRecorderState::Status, protocol::RpcError> get_mcap_recording_status(
+	McapRecorderState &recorder_state,
+	const recorder_pb::McapStatusRequest &) {
 	std::lock_guard lock(recorder_state.mutex);
 	recorder_state.status.can_record = true;
 	return recorder_state.status;
 }
 
+void close_mp4_writer_with_error(
+	Mp4RecorderState &recorder_state,
+	const std::string &message) {
+	spdlog::error("pipeline MP4 recorder stopping after error: {}", message);
+	if (recorder_state.writer) {
+		auto flush = recorder_state.writer->flush();
+		if (!flush) {
+			spdlog::warn("pipeline MP4 flush failed while closing recorder: {}", flush.error());
+		}
+		recorder_state.writer.reset();
+	}
+	recorder_state.first_frame_timestamp_ns.reset();
+	recorder_state.status.last_frame_ok = false;
+	recorder_state.status.error_message = message;
+	reset_mp4_status_after_stop(recorder_state.status);
+}
+
+[[nodiscard]]
+protocol::RpcError make_mp4_rpc_error(
+	const protocol::RpcErrorCode code,
+	std::string message) {
+	return protocol::RpcError{
+		.code = code,
+		.message = std::move(message),
+	};
+}
+
+[[nodiscard]]
+recorder_pb::Mp4RecorderState to_proto_mp4_state(const Mp4RecorderStatus &status) {
+	recorder_pb::Mp4RecorderState wire_status;
+	wire_status.set_can_record(status.can_record);
+	wire_status.set_is_recording(status.is_recording);
+	wire_status.set_last_frame_ok(status.last_frame_ok);
+	wire_status.set_frames_ingested(status.frames_ingested);
+	wire_status.set_frames_encoded(status.frames_encoded);
+	wire_status.set_active_path(status.active_path);
+	wire_status.set_error_message(status.error_message);
+	return wire_status;
+}
+
+template <class Response>
+[[nodiscard]]
+Response make_ok_mp4_response(const Mp4RecorderStatus &status) {
+	Response response;
+	response.set_code(recorder_pb::RPC_CODE_OK);
+	*response.mutable_state() = to_proto_mp4_state(status);
+	return response;
+}
+
+[[nodiscard]]
+std::expected<Mp4RecorderStatus, protocol::RpcError> start_mp4_recording(
+	Mp4RecorderState &recorder_state,
+	const recorder_pb::Mp4StartRequest &request) {
+	std::lock_guard lock(recorder_state.mutex);
+	if (!recorder_state.current_frame_info) {
+		return std::unexpected(make_mp4_rpc_error(
+			protocol::RpcErrorCode::Internal,
+			"MP4 recorder is not ready; snapshot frame info unavailable"));
+	}
+	if (!recorder_state.current_input_pixel_format) {
+		return std::unexpected(make_mp4_rpc_error(
+			protocol::RpcErrorCode::Unsupported,
+			recorder_state.status.error_message.empty()
+				? "MP4 recorder does not support the current snapshot format"
+				: recorder_state.status.error_message));
+	}
+	if (recorder_state.writer) {
+		return std::unexpected(make_mp4_rpc_error(
+			protocol::RpcErrorCode::Busy,
+			"MP4 recording is already active"));
+	}
+
+	const auto output_path = std::filesystem::path(request.output_path());
+	if (output_path.empty()) {
+		return std::unexpected(make_mp4_rpc_error(
+			protocol::RpcErrorCode::InvalidRequest,
+			"output_path must not be empty"));
+	}
+	std::error_code mkdir_error{};
+	if (output_path.has_parent_path()) {
+		std::filesystem::create_directories(output_path.parent_path(), mkdir_error);
+		if (mkdir_error) {
+			return std::unexpected(make_mp4_rpc_error(
+				protocol::RpcErrorCode::Internal,
+				"failed to create MP4 output directory: " + mkdir_error.message()));
+		}
+	}
+
+	record::Mp4RecordWriter writer{};
+	auto open = writer.open(
+		output_path,
+		recorder_state.base_config.encoder.codec,
+		recorder_state.base_config.encoder.device,
+		recorder_state.current_frame_info->width,
+		recorder_state.current_frame_info->height,
+		recorder_state.current_fps,
+		make_mp4_encode_tuning(recorder_state.base_config),
+		*recorder_state.current_input_pixel_format);
+	if (!open) {
+		return std::unexpected(make_mp4_rpc_error(
+			protocol::RpcErrorCode::Internal,
+			"pipeline MP4 writer init failed: " + open.error()));
+	}
+
+	recorder_state.writer.emplace(std::move(writer));
+	recorder_state.first_frame_timestamp_ns.reset();
+	recorder_state.status.can_record = true;
+	recorder_state.status.is_recording = true;
+	recorder_state.status.last_frame_ok = true;
+	recorder_state.status.frames_ingested = 0;
+	recorder_state.status.frames_encoded = 0;
+	recorder_state.status.active_path = request.output_path();
+	recorder_state.status.error_message.clear();
+	return recorder_state.status;
+}
+
+[[nodiscard]]
+std::expected<Mp4RecorderStatus, protocol::RpcError> stop_mp4_recording(
+	Mp4RecorderState &recorder_state,
+	const recorder_pb::Mp4StopRequest &) {
+	std::lock_guard lock(recorder_state.mutex);
+	if (recorder_state.writer) {
+		auto flush = recorder_state.writer->flush();
+		recorder_state.writer.reset();
+		recorder_state.first_frame_timestamp_ns.reset();
+		if (!flush) {
+			recorder_state.status.last_frame_ok = false;
+			recorder_state.status.error_message = "MP4 recording flush failed: " + flush.error();
+			reset_mp4_status_after_stop(recorder_state.status);
+			return std::unexpected(make_mp4_rpc_error(
+				protocol::RpcErrorCode::Internal,
+				recorder_state.status.error_message));
+		}
+	}
+	recorder_state.status.last_frame_ok = true;
+	recorder_state.status.error_message.clear();
+	reset_mp4_status_after_stop(recorder_state.status);
+	return recorder_state.status;
+}
+
+[[nodiscard]]
+std::expected<Mp4RecorderStatus, protocol::RpcError> get_mp4_recording_status(
+	Mp4RecorderState &recorder_state,
+	const recorder_pb::Mp4StatusRequest &) {
+	std::lock_guard lock(recorder_state.mutex);
+	recorder_state.status.can_record =
+		recorder_state.current_frame_info.has_value() &&
+		recorder_state.current_input_pixel_format.has_value();
+	return recorder_state.status;
+}
+
+void update_mp4_source_info(
+	Mp4RecorderState &recorder_state,
+	const ipc::FrameInfo &frame_info,
+	const float fps) {
+	std::lock_guard lock(recorder_state.mutex);
+	const auto previous_frame_info = recorder_state.current_frame_info;
+	recorder_state.current_frame_info = frame_info;
+	if (fps > 0.0f) {
+		recorder_state.current_fps = fps;
+	}
+
+	auto input_pixel_format = mp4_input_pixel_format(frame_info);
+	if (!input_pixel_format) {
+		recorder_state.current_input_pixel_format.reset();
+		recorder_state.status.can_record = false;
+		recorder_state.status.error_message = input_pixel_format.error();
+		if (recorder_state.writer) {
+			close_mp4_writer_with_error(recorder_state, input_pixel_format.error());
+		}
+		return;
+	}
+
+	const bool source_changed =
+		recorder_state.writer.has_value() &&
+		previous_frame_info.has_value() &&
+		!frame_info_equal(*previous_frame_info, frame_info);
+	recorder_state.current_input_pixel_format = *input_pixel_format;
+	recorder_state.status.can_record = true;
+	recorder_state.status.error_message.clear();
+	if (source_changed) {
+		close_mp4_writer_with_error(
+			recorder_state,
+			"MP4 recording stopped after snapshot frame format changed");
+		return;
+	}
+}
+
+void write_mp4_frame(
+	Mp4RecorderState *recorder_state,
+	const ipc::CoherentSnapshot &snapshot) {
+	if (recorder_state == nullptr) {
+		return;
+	}
+
+	std::lock_guard lock(recorder_state->mutex);
+	if (!recorder_state->writer) {
+		return;
+	}
+	if (snapshot.left.empty()) {
+		close_mp4_writer_with_error(*recorder_state, "MP4 recorder received an empty left frame");
+		return;
+	}
+	if (snapshot.metadata.info.height == 0) {
+		close_mp4_writer_with_error(*recorder_state, "MP4 recorder received an invalid frame height");
+		return;
+	}
+
+	const auto row_stride_bytes =
+		static_cast<std::size_t>(snapshot.metadata.info.buffer_size) /
+		static_cast<std::size_t>(snapshot.metadata.info.height);
+	if (row_stride_bytes == 0) {
+		close_mp4_writer_with_error(*recorder_state, "MP4 recorder computed a zero row stride");
+		return;
+	}
+	if (!recorder_state->first_frame_timestamp_ns) {
+		recorder_state->first_frame_timestamp_ns = snapshot.metadata.timestamp_ns;
+	}
+	const auto relative_timestamp_ns =
+		snapshot.metadata.timestamp_ns >= *recorder_state->first_frame_timestamp_ns
+			? snapshot.metadata.timestamp_ns - *recorder_state->first_frame_timestamp_ns
+			: 0ull;
+
+	auto write = recorder_state->writer->write_frame(
+		snapshot.left.data(),
+		row_stride_bytes,
+		relative_timestamp_ns);
+	if (!write) {
+		close_mp4_writer_with_error(*recorder_state, "MP4 frame write failed: " + write.error());
+		spdlog::error("pipeline MP4 frame write failed: {}", write.error());
+		return;
+	}
+
+	recorder_state->status.last_frame_ok = true;
+	recorder_state->status.is_recording = true;
+	recorder_state->status.frames_ingested += 1;
+	recorder_state->status.frames_encoded += 1;
+}
+
 void update_mcap_stream_info(
 	McapRecorderState &recorder_state,
 	const encode::EncodedStreamInfo &stream_info) {
@@ -776,30 +1127,29 @@ int run_pipeline(const RuntimeConfig &config) {
 	std::optional<protocol::UdpRtpPublisher> rtp_publisher{};
 	std::optional<protocol::RtmpOutput> rtmp_output{};
 	McapRecorderState mcap_recorder{};
+	Mp4RecorderState mp4_recorder{};
 	mcap_recorder.base_config = config;
 	mcap_recorder.status.can_record = true;
+	mp4_recorder.base_config = config;
 	cvmmap::NatsControlClient nats_client(
 		input_endpoints->nats_target_key,
 		config.input.nats_url);
-	cvmmap::NatsControlService recorder_service(
-		cvmmap::NatsControlServiceOptions{
-			.instance_name = input_endpoints->instance_name,
-			.namespace_name = input_endpoints->namespace_name,
-			.ipc_prefix = input_endpoints->ipc_prefix,
-			.base_name = input_endpoints->base_name,
-			.target_key = input_endpoints->nats_target_key,
-			.shm_name = input_endpoints->shm_name,
-			.zmq_addr = input_endpoints->zmq_endpoint,
-			.backend = std::string((*source)->backend_name()),
-			.nats_url = config.input.nats_url,
-		});
+	protocol::NatsRequestReplyServer recorder_rpc_server({
+		.nats_url = config.input.nats_url,
+		.instance_name = input_endpoints->instance_name,
+		.namespace_name = input_endpoints->namespace_name,
+		.ipc_prefix = input_endpoints->ipc_prefix,
+		.base_name = input_endpoints->base_name,
+		.nats_target_key = input_endpoints->nats_target_key,
+		.recording_formats = "mp4,mcap",
+	});
 	std::mutex nats_event_mutex{};
 	std::deque<std::vector<std::uint8_t>> pending_body_packets{};
 	std::deque<int32_t> pending_status_codes{};
 
-	nats_client.SetModuleStatusCallback([&nats_event_mutex, &pending_status_codes](int32_t status_code) {
+	nats_client.SetModuleStatusCallback([&nats_event_mutex, &pending_status_codes](cvmmap::ModuleStatus status_code) {
 		std::lock_guard lock(nats_event_mutex);
-		pending_status_codes.push_back(status_code);
+		pending_status_codes.push_back(static_cast<int32_t>(status_code));
 	});
 	nats_client.SetBodyTrackingRawCallback(
 		[&nats_event_mutex, &pending_body_packets](std::span<const std::uint8_t> bytes) {
@@ -811,38 +1161,86 @@ int run_pipeline(const RuntimeConfig &config) {
 		return exit_code(PipelineExitCode::SubscriberError);
 	}
 
-	cvmmap::NatsControlHandlers recorder_handlers{};
-	recorder_handlers.on_recording_available =
-		[](const cvmmap::RecordingFormat format) {
-		return format == cvmmap::RecordingFormat::Mcap;
-	};
-	recorder_handlers.on_start_recording =
-		[&mcap_recorder](const cvmmap::RecordingRequest &request) {
-		return start_mcap_recording(mcap_recorder, request);
-	};
-	recorder_handlers.on_stop_recording =
-		[&mcap_recorder](const cvmmap::RecordingFormat format)
-			-> std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> {
-		if (format != cvmmap::RecordingFormat::Mcap) {
-			return std::unexpected(make_recording_control_error(
-				cvmmap::CONTROL_RESPONSE_UNSUPPORTED,
-				"recording format is not supported by the streamer"));
-		}
-		return stop_mcap_recording(mcap_recorder);
-	};
-	recorder_handlers.on_get_recording_status =
-		[&mcap_recorder](const cvmmap::RecordingFormat format)
-			-> std::expected<cvmmap::RecordingStatus, cvmmap::ControlError> {
-		if (format != cvmmap::RecordingFormat::Mcap) {
-			return std::unexpected(make_recording_control_error(
-				cvmmap::CONTROL_RESPONSE_UNSUPPORTED,
-				"recording format is not supported by the streamer"));
-		}
-		return get_mcap_recording_status(mcap_recorder);
-	};
-	recorder_service.SetHandlers(std::move(recorder_handlers));
-	if (!recorder_service.Start()) {
-		spdlog::error("pipeline recorder control service failed on '{}'", config.input.nats_url);
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::Mp4StartRequest,
+		recorder_pb::Mp4StartResponse>(
+		"recorder_mp4_start",
+		protocol::subject_recorder_mp4_start(input_endpoints->nats_target_key),
+		[&mp4_recorder](const recorder_pb::Mp4StartRequest &request)
+			-> std::expected<recorder_pb::Mp4StartResponse, protocol::RpcError> {
+			auto status = start_mp4_recording(mp4_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mp4_response<recorder_pb::Mp4StartResponse>(*status);
+		});
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::Mp4StopRequest,
+		recorder_pb::Mp4StopResponse>(
+		"recorder_mp4_stop",
+		protocol::subject_recorder_mp4_stop(input_endpoints->nats_target_key),
+		[&mp4_recorder](const recorder_pb::Mp4StopRequest &request)
+			-> std::expected<recorder_pb::Mp4StopResponse, protocol::RpcError> {
+			auto status = stop_mp4_recording(mp4_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mp4_response<recorder_pb::Mp4StopResponse>(*status);
+		});
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::Mp4StatusRequest,
+		recorder_pb::Mp4StatusResponse>(
+		"recorder_mp4_status",
+		protocol::subject_recorder_mp4_status(input_endpoints->nats_target_key),
+		[&mp4_recorder](const recorder_pb::Mp4StatusRequest &request)
+			-> std::expected<recorder_pb::Mp4StatusResponse, protocol::RpcError> {
+			auto status = get_mp4_recording_status(mp4_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mp4_response<recorder_pb::Mp4StatusResponse>(*status);
+		});
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::McapStartRequest,
+		recorder_pb::McapStartResponse>(
+		"recorder_mcap_start",
+		protocol::subject_recorder_mcap_start(input_endpoints->nats_target_key),
+		[&mcap_recorder](const recorder_pb::McapStartRequest &request)
+			-> std::expected<recorder_pb::McapStartResponse, protocol::RpcError> {
+			auto status = start_mcap_recording(mcap_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mcap_response<recorder_pb::McapStartResponse>(*status);
+		});
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::McapStopRequest,
+		recorder_pb::McapStopResponse>(
+		"recorder_mcap_stop",
+		protocol::subject_recorder_mcap_stop(input_endpoints->nats_target_key),
+		[&mcap_recorder](const recorder_pb::McapStopRequest &request)
+			-> std::expected<recorder_pb::McapStopResponse, protocol::RpcError> {
+			auto status = stop_mcap_recording(mcap_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mcap_response<recorder_pb::McapStopResponse>(*status);
+		});
+	recorder_rpc_server.register_proto_endpoint<
+		recorder_pb::McapStatusRequest,
+		recorder_pb::McapStatusResponse>(
+		"recorder_mcap_status",
+		protocol::subject_recorder_mcap_status(input_endpoints->nats_target_key),
+		[&mcap_recorder](const recorder_pb::McapStatusRequest &request)
+			-> std::expected<recorder_pb::McapStatusResponse, protocol::RpcError> {
+			auto status = get_mcap_recording_status(mcap_recorder, request);
+			if (!status) {
+				return std::unexpected(status.error());
+			}
+			return make_ok_mcap_response<recorder_pb::McapStatusResponse>(*status);
+		});
+	if (!recorder_rpc_server.start()) {
+		spdlog::error("pipeline streamer recorder service failed on '{}'", config.input.nats_url);
 		nats_client.Stop();
 		return exit_code(PipelineExitCode::SubscriberError);
 	}
@@ -951,6 +1349,7 @@ int run_pipeline(const RuntimeConfig &config) {
 		}
 
 		update_mcap_stream_info(mcap_recorder, stream_info);
+		update_mp4_source_info(mp4_recorder, target_info, stream_fps(stream_info));
 		if (config.record.mcap.enabled) {
 			std::lock_guard lock(mcap_recorder.mutex);
 			if (!mcap_recorder.sink) {
@@ -962,7 +1361,6 @@ int run_pipeline(const RuntimeConfig &config) {
 				}
 				mcap_recorder.active_record_config = config;
 				mcap_recorder.sink.emplace(std::move(*created));
-				mcap_recorder.status.format = cvmmap::RecordingFormat::Mcap;
 				mcap_recorder.status.can_record = true;
 				mcap_recorder.status.is_recording = true;
 				mcap_recorder.status.last_frame_ok = true;
@@ -1187,6 +1585,10 @@ int run_pipeline(const RuntimeConfig &config) {
 			const bool want_encoded_input =
 				config.input.video_source == InputVideoSource::Encoded ||
 				(config.input.video_source == InputVideoSource::Auto && has_encoded_access_unit);
+			update_mp4_source_info(
+				mp4_recorder,
+				snapshot->metadata.info,
+				active_stream_info ? stream_fps(*active_stream_info) : 30.0f);
 			if (config.input.video_source == InputVideoSource::Encoded && !has_encoded_access_unit) {
 				spdlog::error("pipeline encoded input requested but SHM snapshot does not contain an encoded access unit");
 				return exit_code(PipelineExitCode::InitializationError);
@@ -1229,6 +1631,7 @@ int run_pipeline(const RuntimeConfig &config) {
 			}
 
 			latency_tracker.note_ingest();
+			write_mp4_frame(&mp4_recorder, *snapshot);
 
 			if (want_encoded_input) {
 				auto access_unit = make_access_unit_from_snapshot(*snapshot);
@@ -1350,8 +1753,15 @@ int run_pipeline(const RuntimeConfig &config) {
 	if (backend) {
 		(*backend)->shutdown();
 	}
-	std::ignore = stop_mcap_recording(mcap_recorder);
-	recorder_service.Stop();
+	auto stop_mp4 = stop_mp4_recording(mp4_recorder, recorder_pb::Mp4StopRequest{});
+	if (!stop_mp4) {
+		spdlog::warn("pipeline MP4 recorder stop during shutdown failed: {}", stop_mp4.error().message);
+	}
+	auto stop_mcap = stop_mcap_recording(mcap_recorder, recorder_pb::McapStopRequest{});
+	if (!stop_mcap) {
+		spdlog::warn("pipeline MCAP recorder stop during shutdown failed: {}", stop_mcap.error().message);
+	}
+	recorder_rpc_server.stop();
 
 	spdlog::info(
 		"PIPELINE_METRICS codec={} backend={} sync_messages={} status_messages={} torn_frames={} pushed_frames={} encoded_access_units={} resets={} format_rebuilds={} supervised_restarts={}",