IOManager.hpp

Go to the documentation of this file.

#pragma once
 
#include "Image/ImageWriter.hpp"
 
#include "Model/ModelWriter.hpp"
 
#include "MayaFlux/Kakshya/NDData/Composite.hpp"
#include "MayaFlux/Kakshya/Region/Region.hpp"
 
#include "Spatial/SpatialTransfer.hpp"
 
#include "Video/VideoFileReader.hpp"
 
#include "Volume/VolumeReader.hpp"
#include "Volume/VolumeWriter.hpp"
 
#include "MayaFlux/Transitive/Parallel/AsyncGroup.hpp"
 
#include <future>
 
#include <libavcodec/avcodec.h>
#include <libavutil/pixfmt.h>
 
namespace MayaFlux::Core {
class VKImage;
struct GlobalStreamInfo;
}
 
namespace MayaFlux::Nodes::Network {
class MeshNetwork;
}
 
namespace MayaFlux::Kakshya {
class SignalSourceContainer;
class VideoStreamContainer;
class VideoFileContainer;
class SoundFileContainer;
class CameraContainer;
class AudioOutputContainer;
class CompositeContainer;
}
 
namespace MayaFlux::Buffers {
class VideoContainerBuffer;
class SoundContainerBuffer;
class TextureBuffer;
class TextBuffer;
class MeshBuffer;
class ComputeMeshBuffer;
class VolumeGridBuffer;
class BufferManager;
}
 
namespace MayaFlux::Registry::Service {
class IOService;
class AudioBackendService;
}
 
namespace MayaFlux::IO {
 
using TextureResolver = std::function<std::shared_ptr<Core::VKImage>(const std::string& path)>;
 
class ImageReader;
class ModelReader;
class SoundFileWriter;
class CameraSource;
struct CameraConfig;
class CompositeReader;
class CompositeWriter;
class VideoFileWriter;
class VolumeCapture;
 
struct VideoLoadConfig {
    FileReadOptions file_options { FileReadOptions::EXTRACT_METADATA | FileReadOptions::EXTRACT_REGIONS };
    AudioReadOptions audio_options { AudioReadOptions::DEINTERLEAVE };
    VideoReadOptions video_options {};
    uint32_t target_width { 1920 };
    uint32_t target_height { 1080 };
};
 
/**
 * @brief Result of load_video(path, config): the video container, then its audio.
 *
 * Read it as `auto [video, audio] = load_video(path, config);`. The audio is
 * nullptr if the file has no audio track or VideoReadOptions::EXTRACT_AUDIO was
 * not set in the config. Both are nullptr on failure.
 */
using VideoLoadResult = std::pair<
    std::shared_ptr<Kakshya::VideoFileContainer>,
    std::shared_ptr<Kakshya::SoundFileContainer>>;
 
/**
 * @class IOManager
 * @brief Optional orchestration layer for IO reader lifetime and IOService dispatch.
 *
 * MayaFlux IO readers are fully self-sufficient — VideoFileReader, SoundFileReader,
 * and ImageReader can all be used standalone without this class.  IOManager exists
 * to solve problems that only arise at scale:
 *
 *  - Globally unique reader_ids across concurrent VideoFileReader instances, so
 *    IOService::request_decode routes to the correct decode thread.
 *  - A single IOService registration backed by a keyed dispatch map, rather than
 *    each reader overwriting the previous registration.
 *  - Shared ownership of VideoFileReader instances whose background decode thread
 *    must outlive the scope in which they were created.
 *  - High-level load/hook functions that collapse the full open→load→processor
 *    setup→buffer wiring sequence into single calls, mirroring what Depot does
 *    for audio and eliminating the boilerplate currently required for video.
 *
 * VideoFileReader detects the IOService registration generically via BackendRegistry
 * and skips its own registration — no IOManager-specific knowledge in the reader.
 *
 * Levels of integration (all valid):
 *  1. VideoFileReader standalone, single stream — self-registers, self-ids.
 *  2. IOManager constructed directly — multi-stream safe, no Engine required.
 *  3. Engine constructs IOManager in Init() — fully orchestrated, zero user wiring.
 */
class MAYAFLUX_API IOManager {
public:
    /**
     * @brief Construct IOManager and register the IOService into BackendRegistry.
     *
     * Must be constructed before any VideoFileReader::load_into_container() call
     * that should participate in managed dispatch.
     */
    IOManager(Core::GlobalStreamInfo& stream_info, uint32_t frame_rate, const std::shared_ptr<Buffers::BufferManager>& buffer_manager, const std::shared_ptr<Vruta::TaskScheduler>& scheduler);
 
    /**
     * @brief Calls shutdown() if it has not already run.
     *
     * Both stop() and shutdown() are safe to call earlier and explicitly;
     * the destructor exists only to catch a caller that never did.
     */
    ~IOManager();
 
    IOManager(const IOManager&) = delete;
    IOManager& operator=(const IOManager&) = delete;
    IOManager(IOManager&&) = delete;
    IOManager& operator=(IOManager&&) = delete;
 
    /**
     * @brief Finalizes all in-flight output while consumers are still running.
     *
     * Stops every audio, video, volume and spatial capture through its
     * stop_*_capture() call, closes the sound writers and waits for pending
     * saves. A capture the caller never stopped is still finalized (an Ogawa
     * archive or video container left open is not a valid file), and
     * volume/spatial captures release their scheduler reference here instead
     * of reaching into an already-destroyed scheduler.
     *
     * The IOService stays registered and readers stay alive, so threads that
     * consume them keep working. Idempotent. Call before the subsystems stop
     * (see Engine::End()), while backend services and logging sinks are alive.
     */
    void stop();
 
    /**
     * @brief Unregisters IOService and releases readers, writers, buffers and caches.
     *
     * Calls stop() first. Call after the consuming threads have been joined and
     * before the graphics backend is cleaned up, so objects whose destructors
     * touch the backend are released while it is still alive. Idempotent.
     */
    void shutdown();
 
    // ─────────────────────────────────────────────────────────────────────────
    // Video — load
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Load a video file into a VideoFileContainer.
     *
     * Performs can_read check, opens the
     * file with metadata extraction, creates the container, registers the reader
     * with IOManager (assigning a globally unique reader_id), calls
     * setup_io_service(), loads into container, and configures the default
     * FrameAccessProcessor with auto_advance enabled.
     *
     * Audio is NOT extracted. Use load_video(path, options) for that.
     *
     * @param filepath Path to the video file.
     * @return Loaded VideoFileContainer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::VideoFileContainer> load_video(const std::string& filepath);
 
    /**
     * @brief Load a video file into a VideoFileContainer.
     *
     * Performs can_read check, opens with
     * metadata and region extraction, registers the reader (assigning a globally
     * unique reader_id), calls setup_io_service(), loads into container, and
     * configures the default FrameAccessProcessor with auto_advance enabled.
     *
     * Passing VideoReadOptions::EXTRACT_AUDIO also extracts the embedded
     * SoundFileContainer, configures its ContiguousAccessProcessor identically
     * to load_audio_file() in Depot, and fills the audio half of the result.
     *
     * @param filepath Path to the video file.
     * @param config VideoLoadConfig containing options and target dimensions.
     * @return The video container and, if requested and present, its audio container.
     */
    [[nodiscard]] VideoLoadResult load_video(
        const std::string& filepath, VideoLoadConfig config);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Video — hook
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Wire a video container to the graphics buffer system.
     *
     * Creates a VideoContainerBuffer via BufferManager with GRAPHICS_BACKEND token,
     * stores it keyed by container pointer, and returns it. A VideoFileContainer
     * or a DynamicVideoStream can be wired. A DynamicVideoStream needs frames or
     * an enabled ring before it can be hooked, and gets its seek processor.
     *
     * @param container Video container to wire.
     * @return Created VideoContainerBuffer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Buffers::VideoContainerBuffer>
    hook_video_container_to_buffer(
        const std::shared_ptr<Kakshya::VideoStreamContainer>& container);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Video — retrieve
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Retrieve the VideoContainerBuffer created for a container.
     * @param container Container previously passed to hook_video_container_to_buffer().
     * @return Stored buffer, or nullptr if not found.
     */
    [[nodiscard]] std::shared_ptr<Buffers::VideoContainerBuffer>
    get_video_buffer(
        const std::shared_ptr<Kakshya::VideoStreamContainer>& container) const;
 
    // ─────────────────────────────────────────────────────────────────────────
    // Audio — load
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Load an audio file into a SoundFileContainer.
     *
     * Performs can_read check, opens the file, creates and populates the
     * container, and configures the default ContiguousAccessProcessor
     * with auto_advance enabled.
     *
     * @param filepath Path to the audio file.
     * @param config LoadConfig struct containing audio read options.
     * @return Loaded SoundFileContainer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::SoundFileContainer> load_audio(const std::string& filepath, VideoLoadConfig config = {});
 
    /**
     * @brief Load an audio file into a fully resident, size-bounded DynamicSoundStream.
     *
     * Applies the engine sample rate, ROW_MAJOR layout, and the engine buffer
     * size as the CursorAccessProcessor block size. The result is ready for use
     * with StreamSliceProcessor without any further configuration.
     *
     * @param filepath   Path to the audio file.
     * @param max_frames Upper bound on frame count. 0 defaults to 5 s at the
     *                   engine sample rate inside SoundFileReader::load_bounded.
     * @param truncate   If true, silently truncate files exceeding max_frames.
     * @return Configured DynamicSoundStream, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::DynamicSoundStream> load_audio_bounded(
        const std::string& filepath,
        uint64_t max_frames = 0,
        bool truncate = false);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Audio — write
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Construct an open SoundFileWriter the caller drives manually.
     *
     * Allocates, opens, and returns a writer ready to accept write() calls.
     * IOManager does not track this writer; the caller is responsible for
     * calling close() when done.
     *
     * @param filepath    Output file path. Extension determines container format.
     * @param channels    Number of channels in all submitted data.
     * @param sample_rate PCM sample rate in Hz.
     * @param codec_id    Encoder override; AV_CODEC_ID_NONE = container default.
     * @return Open SoundFileWriter, or nullptr if open() failed.
     */
    [[nodiscard]] std::shared_ptr<SoundFileWriter>
    create_writer(const std::string& filepath,
        uint32_t channels,
        uint32_t sample_rate = 48000,
        AVCodecID codec_id = AV_CODEC_ID_NONE);
 
    /**
     * @brief Write a SoundStreamContainer to file in one shot.
     *
     * Opens a writer, posts the container's full data, closes, and stores
     * the encode future in m_save_tasks. Non-blocking; returns immediately.
     * Container metadata (sample_rate, channels) drives the encoder parameters.
     *
     * @param container Source container; any SoundStreamContainer child.
     * @param filepath  Output file path.
     * @param codec_id  Encoder override; AV_CODEC_ID_NONE = container default.
     */
    void write(const std::shared_ptr<Kakshya::SoundStreamContainer>& container,
        const std::string& filepath,
        AVCodecID codec_id = AV_CODEC_ID_NONE);
 
    /**
     * @brief Begin continuous capture of live audio output to a file.
     *
     * Registers an observer with AudioBackendService. Each output cycle the
     * observer drives an AudioOutputContainer and posts its processed_data
     * to a SoundFileWriter. Returns an opaque capture id for use with
     * stop_audio_capture().
     *
     * Returns 0 and logs an error if AudioBackendService is unavailable.
     *
     * @param filepath    Output file path.
     * @param codec_id    Encoder override; AV_CODEC_ID_NONE = container default.
     * @return Capture handle; pass to stop_audio_capture() to finalise.
     */
    [[nodiscard]] uint32_t capture_output(const std::string& filepath,
        AVCodecID codec_id = AV_CODEC_ID_NONE);
 
    /**
     * @brief Stop a running audio capture and finalise the file.
     *
     * Unregisters the AudioBackendService observer, calls writer->close(),
     * and stores the encode future in m_save_tasks. Non-blocking.
     *
     * @param capture_id Handle returned by capture_output().
     */
    void stop_audio_capture(uint32_t capture_id);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Audio — hook
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Wire a SoundFileContainer to the audio buffer system.
     *
     * Creates per-channel SoundContainerBuffers with AUDIO_BACKEND token,
     * stores them keyed by container pointer, and returns them.
     *
     * @param container Loaded SoundFileContainer to wire.
     * @return Per-channel SoundContainerBuffers, or empty on failure.
     */
    [[nodiscard]] std::vector<std::shared_ptr<Buffers::SoundContainerBuffer>>
    hook_audio_container_to_buffers(
        const std::shared_ptr<Kakshya::SoundFileContainer>& container);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Audio — retrieve
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Retrieve the SoundContainerBuffers created for a container.
     * @param container Container previously passed to hook_audio_container_to_buffers().
     * @return Stored buffer vector, or empty if not found.
     */
    [[nodiscard]] std::vector<std::shared_ptr<Buffers::SoundContainerBuffer>>
    get_audio_buffers(
        const std::shared_ptr<Kakshya::SoundFileContainer>& container) const;
 
    /**
     * @brief Retrieve the SoundFileContainer extracted from a video file.
     * @param container VideoFileContainer whose audio was extracted during load_video().
     * @return Extracted SoundFileContainer, or nullptr if not found.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::SoundFileContainer>
    get_extracted_audio(const std::shared_ptr<Kakshya::VideoFileContainer>& container) const;
 
    // ─────────────────────────────────────────────────────────────────────────
    // Camera — open
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Open an FFmpeg camera device and create a CameraContainer.
     *
     * Convenience wrapper over the backend-agnostic hosting path: constructs
     * an FFmpegCameraReader, opens the device, creates the container,
     * configures its FrameAccessProcessor (auto_advance disabled), then
     * calls register_camera_source() to assign a reader_id and wire the
     * container's IOService callback via CameraContainer::setup_io().
     *
     * A non-FFmpeg backend does not go through this call at all — it opens
     * itself however its own backend requires, then calls
     * register_camera_source() directly with the resulting CameraSource.
     *
     * After this call, the normal processing pipeline drives frame pulls:
     * CameraContainer::process_default() → IOService::request_frame(reader_id)
     * → dispatch_frame_request() → CameraSource::pull_frame_all()
     * → decode thread → pull_frame() → mutable_frame_ptr() write → READY.
     *
     * @param config Device and resolution configuration.
     * @return Opened CameraContainer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::CameraContainer>
    open_camera(const CameraConfig& config);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Camera — hook
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Wire a CameraContainer to the graphics buffer system.
     *
     * Creates a VideoContainerBuffer via BufferManager with GRAPHICS_BACKEND
     * token, stores it keyed by container pointer, and returns it. Identical
     * to hook_video_container_to_buffer() but accepts CameraContainer.
     *
     * @param container Opened CameraContainer from open_camera().
     * @return Created VideoContainerBuffer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Buffers::VideoContainerBuffer>
    hook_camera_to_buffer(const std::shared_ptr<Kakshya::CameraContainer>& container);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Camera — retrieve
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Retrieve the VideoContainerBuffer created for a camera container.
     */
    [[nodiscard]] std::shared_ptr<Buffers::VideoContainerBuffer>
    get_camera_buffer(const std::shared_ptr<Kakshya::CameraContainer>& container) const;
 
    // ─────────────────────────────────────────────────────────────────────────
    // Camera source management
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Assign a globally unique reader_id and take ownership of a
     *        CameraSource that is already open.
     *
     * The backend-agnostic entry point for camera hosting: any CameraSource
     * implementation, from any backend, is registered identically. open()
     * and configuration are entirely the caller's concern beforehand — this
     * call never touches either.
     *
     * @param source Fully-opened CameraSource.
     * @return Globally unique reader_id assigned to this source.
     */
    [[nodiscard]] uint64_t register_camera_source(std::shared_ptr<CameraSource> source);
 
    /**
     * @brief Release ownership of the camera source identified by reader_id.
     *
     * No-op with a warning if reader_id is unknown.
     *
     * @param reader_id Id returned by register_camera_source().
     */
    void release_camera_source(uint64_t reader_id);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Video reader management
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Assign a globally unique reader_id and take ownership of a reader.
     *
     * Called internally by load_video(). Advanced users who open a
     * VideoFileReader manually can call this before load_into_container() to
     * participate in managed dispatch and lifetime tracking.
     *
     * Calls reader->set_reader_id(id) before returning.
     *
     * @param reader Fully-opened VideoFileReader.
     * @return Globally unique reader_id assigned to this reader.
     */
    [[nodiscard]] uint64_t register_video_reader(std::shared_ptr<VideoFileReader> reader);
 
    /**
     * @brief Release ownership of the reader identified by reader_id.
     *
     * When the last shared_ptr to the reader is released its decode thread
     * is joined.  Safe to call from a container state-change callback.
     * No-op with a warning if reader_id is unknown.
     *
     * @param reader_id Id returned by register_video_reader().
     */
    void release_video_reader(uint64_t reader_id);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Video — write
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Construct an open VideoFileWriter the caller drives manually.
     *
     * Allocates and opens the writer. IOManager does not track this writer;
     * the caller is responsible for calling close() or stop_recording() when
     * done.
     *
     * @param filepath         Output file path. Extension determines container.
     * @param width            Frame width in pixels.
     * @param height           Frame height in pixels.
     * @param frame_rate       Average frame rate in frames per second.
     * @param src_pixel_format AVPixelFormat of submitted pixel data.
     * @param codec_id         Encoder override; AV_CODEC_ID_NONE = container default.
     * @return Open VideoFileWriter, or nullptr if open() failed.
     */
    [[nodiscard]] std::shared_ptr<VideoFileWriter>
    create_writer(const std::string& filepath,
        uint32_t width,
        uint32_t height,
        double frame_rate,
        AVPixelFormat src_pixel_format,
        AVCodecID codec_id = AV_CODEC_ID_NONE);
 
    /**
     * @brief Begin continuous capture of a window's rendered frames to a file.
     *
     * Delegates to VideoFileWriter::record(). The encoder is opened lazily on
     * the first delivered frame so pixel format and dimensions come from the
     * live swapchain. Returns an opaque capture id for use with
     * stop_video_capture().
     *
     * Returns 0 if the window is null or record() fails.
     *
     * @param window     Window whose swapchain frames will be captured.
     * @param filepath   Output file path. Extension determines container format.
     * @param frame_rate Nominal frame rate written into the container header.
     * @param codec_id   Encoder override; AV_CODEC_ID_NONE = container default.
     * @return Capture handle; pass to stop_video_capture() to finalise.
     */
    [[nodiscard]] uint32_t capture_window(
        const std::shared_ptr<Core::Window>& window,
        const std::string& filepath,
        double frame_rate,
        AVCodecID codec_id = AV_CODEC_ID_NONE);
 
    /**
     * @brief Stop a running video capture and finalise the file.
     *
     * Calls writer->stop_recording(), the counterpart to the record() call
     * capture_window() made, and stores the encode future in m_save_tasks.
     * Non-blocking.
     *
     * @param capture_id Handle returned by capture_window().
     */
    void stop_video_capture(uint32_t capture_id);
 
    /**
     * @brief Stop the active window capture and finalise the file.
     *
     * Convenience overload that resolves the capture id by window pointer,
     * then calls stop_video_capture(). No-op with a warning if no capture
     * is active for this window.
     *
     * @param window Window previously passed to capture_window().
     */
    void stop_capture(const std::shared_ptr<Core::Window>& window);
 
    /**
     * @brief Returns all active video capture IDs (capture_window()).
     */
    [[nodiscard]] std::vector<uint32_t> get_video_capture_ids() const;
 
    /**
     * @brief Returns all VideoFileWriters created via create_writer().
     */
    [[nodiscard]] std::vector<std::shared_ptr<VideoFileWriter>> get_video_writers() const;
 
    // ─────────────────────────────────────────────────────────────────────────
    // Image — load
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Load an image file into a TextureBuffer.
     *
     * Opens the file via ImageReader, decodes to the requested channel layout,
     * and creates a TextureBuffer containing the pixel data. The reader is retained
     * for the lifetime of IOManager.
     *
     * @param filepath Path to the image file (PNG, JPG, BMP, TGA, etc.).
     * @param desired_channels Channel count passed to ImageReader (0 keeps the source layout).
     * @param region Optional source-pixel rectangle with inclusive end coordinates.
     * @return TextureBuffer with pixel data, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Buffers::TextureBuffer>
    load_image(const std::string& filepath, int desired_channels = 4,
        std::optional<Kakshya::Region> region = std::nullopt);
 
    /**
     * @brief Open a schema-bearing reader selected by file format.
     *
     * CSV and TSV currently use DelimitedTextReader. The returned reader is
     * owned by the caller and can inspect its layout or read bounded batches.
     * IOManager does not retain it or register it with IOService.
     *
     * @param filepath Path to a supported Composite file.
     * @param layout Optional exact field layout. Without one, CSV/TSV fields
     * are preserved as UTF-8 text.
     * @return Open reader, or nullptr for an unsupported or invalid file.
     */
    [[nodiscard]] std::shared_ptr<CompositeReader>
    open_composite_reader(
        const std::string& filepath,
        std::optional<Kakshya::CompositeLayout> layout = std::nullopt);
 
    /**
     * @brief Load a complete Composite file into an owning container.
     *
     * Dispatches by supported file format. The reader is local to this call;
     * use open_composite_reader() for bounded reading of large files.
     *
     * @param filepath Path to a supported Composite file.
     * @param layout Optional exact field layout for typed CSV/TSV values.
     * @param batch_size Default container materialization batch size.
     * @return Owning CompositeContainer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Kakshya::CompositeContainer>
    load_composite(
        const std::string& filepath,
        std::optional<Kakshya::CompositeLayout> layout = std::nullopt,
        size_t batch_size = 1);
 
    /**
     * @brief Open a Composite writer selected by the output file format.
     *
     * CSV and TSV currently use DelimitedTextWriter. The caller owns the
     * returned writer and must call write_rows() and close(). IOManager
     * does not retain it or enqueue its writes.
     *
     * @param filepath Destination path with a supported extension.
     * @param layout Field layout for every slice written to the file.
     * @return Open writer, or nullptr if the format or layout is unsupported
     * or the file could not be opened.
     */
    [[nodiscard]] std::shared_ptr<CompositeWriter>
    open_composite_writer(
        const std::string& filepath,
        const Kakshya::CompositeLayout& layout);
 
    /**
     * @brief Save a complete CompositeContainer in a supported format.
     *
     * The write is synchronous, unlike the queued save_image, save_mesh, and
     * save_volume. For bounded writes of large data, use
     * open_composite_writer() and supply slices directly. A partial file
     * may remain on failure.
     *
     * @param container Source container.
     * @param filepath Destination CSV or TSV path.
     * @return True only if every element was written and the file closed.
     */
    bool save_composite(
        const std::shared_ptr<Kakshya::CompositeContainer>& container,
        const std::string& filepath);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Mesh — load
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Load all meshes from a 3D model file into MeshBuffer instances.
     *
     * Opens the file via ModelReader, extracts all aiMesh entries as MeshData,
     * constructs one MeshBuffer per mesh, calls setup_processors() on each,
     * and returns them. setup_rendering() is left to the caller.
     *
     * If resolver is null, the default resolver is used: paths resolved
     * relative to the model file's directory via ImageReader::load_texture.
     * Unresolvable textures are logged and skipped.
     *
     * @param filepath Path to the model file (glTF, FBX, OBJ, PLY, etc.).
     * @param resolver Optional texture resolver.
     * @return One MeshBuffer per mesh in scene order, or empty on failure.
     */
    [[nodiscard]] std::vector<std::shared_ptr<Buffers::MeshBuffer>>
    load_mesh(
        const std::string& filepath,
        TextureResolver resolver = nullptr);
 
    /**
     * @brief Load a 3D model file as a MeshNetwork.
     *
     * One MeshSlot per aiMesh. Per-slot diffuse textures resolved via resolver.
     * Null resolver uses the default: ImageReader::load_texture relative to the
     * model file's directory.
     *
     * @param filepath Path to the model file.
     * @param resolver Optional texture resolver.
     * @return Populated MeshNetwork, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Nodes::Network::MeshNetwork>
    load_mesh_network(
        const std::string& filepath,
        TextureResolver resolver = nullptr);
 
    // ─────────────────────────────────────────────────────────────────────────
    // Mesh - save
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Save a MeshBuffer's mesh data to disk asynchronously.
     *
     * MeshData is always CPU-authoritative for MeshBuffer, so there is no
     * download to wait on, but the write itself (AssimpModelWriter ->
     * Assimp::Exporter -> disk) is still blocking I/O; queuing it here keeps
     * that off the caller's thread, the same reason save_image/save_volume
     * do. Returns once the task is queued; write failure is logged from the
     * task and does not surface here.
     *
     * @param mesh_buffer Source mesh. Must have valid MeshData.
     * @param filepath    Destination path with extension.
     * @param options     Format-specific writer options.
     * @return True once the task is queued.
     */
    bool save_mesh(
        const std::shared_ptr<Buffers::MeshBuffer>& mesh_buffer,
        const std::string& filepath,
        const IO::ModelWriteOptions& options = {});
 
    /**
     * @brief Save one or more meshes to disk asynchronously.
     *
     * For callers that already have MeshData in hand (e.g. from
     * IO::download_compute_mesh) rather than a MeshBuffer. Every entry must
     * satisfy Kakshya::MeshData::is_valid(). Returns once the task is queued.
     */
    bool save_mesh(
        const std::vector<Kakshya::MeshData>& meshes,
        const std::string& filepath,
        const IO::ModelWriteOptions& options = {});
 
    /**
     * @brief Save a MeshNetwork's current slots to disk asynchronously.
     *
     * Slot data lives on MeshWriterNode, CPU-resident the same way
     * MeshBuffer's is, whether or not the network is wrapped in a
     * MeshNetworkBuffer or has ever been rendered. Each slot's vertices are
     * baked into world space from its current world_transform. Returns
     * once the task is queued.
     *
     * @param network  Source network.
     * @param filepath Destination path with extension.
     * @param options  Format-specific writer options.
     * @return True once the task is queued.
     */
    bool save_mesh(
        const std::shared_ptr<Nodes::Network::MeshNetwork>& network,
        const std::string& filepath,
        const IO::ModelWriteOptions& options = {});
 
    /**
     * @brief Save a ComputeMeshBuffer's current live geometry to disk asynchronously.
     *
     * ComputeMeshBuffer's geometry lives GPU-only, so this queues a real
     * device-to-host transfer (IO::download_compute_mesh) on top of the
     * disk write every overload above also queues. Returns once the task
     * is queued; readback or write failure is logged from the task and
     * does not surface here.
     *
     * @param buffer   Source buffer. setup_processors() must have run.
     * @param filepath Destination path with extension.
     * @param options  Format-specific writer options.
     * @return True once the task is queued.
     */
    bool save_mesh(
        const std::shared_ptr<Buffers::ComputeMeshBuffer>& buffer,
        const std::string& filepath,
        const IO::ModelWriteOptions& options = {});
 
    /**
     * @brief Save a ComputeMeshBuffer with a millisecond epoch timestamp
     *        spliced into the path, queued asynchronously.
     *
     * The timestamp is computed on the worker thread at the moment the
     * readback actually runs, not at the moment this call returns, so
     * repeated presses in quick succession still each land in their own
     * file rather than racing on a single path computed up front.
     */
    bool save_mesh_snapshot(
        const std::shared_ptr<Buffers::ComputeMeshBuffer>& buffer,
        const std::string& path_pattern,
        const IO::ModelWriteOptions& options = {});
 
    // ─────────────────────────────────────────────────────────────────────────
    // Image — save
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Save image data to disk asynchronously.
     *
     * The source is downloaded from the GPU on the calling thread (must have
     * command queue access) and the encode + disk flush is dispatched to a
     * worker. Returns once the download completes and the encode task is
     * queued; encode/flush failures are logged but do not block the caller.
     *
     * For synchronous semantics (e.g. shutdown flush, tests) use the
     * IO::save_image / IO::save_texture_buffer free functions in
     * ImageExport.hpp.
     *
     * The extension of @p filepath selects the writer via
     * ImageWriterRegistry. Format-variant compatibility is the writer's
     * responsibility (PNG wants uint8, EXR wants float, etc.).
     *
     * @return True if the download succeeded and the encode task was queued.
     */
    bool save_image(
        const std::shared_ptr<Core::VKImage>& image,
        const std::string& filepath,
        const IO::ImageWriteOptions& options = {});
 
    bool save_image(
        const std::shared_ptr<Buffers::TextureBuffer>& buffer,
        const std::string& filepath,
        const IO::ImageWriteOptions& options = {});
 
    bool save_image(
        const std::shared_ptr<Buffers::TextBuffer>& buffer,
        const std::string& filepath,
        const IO::ImageWriteOptions& options = {});
 
    /**
     * @brief Save an already-downloaded ImageData to disk asynchronously.
     *
     * For callers that have an ImageData in hand (e.g. from
     * IO::download_image) and want to avoid the GPU download step. Purely
     * CPU-bound from here, so no thread restrictions on the caller.
     */
    bool save_image(
        IO::ImageData data,
        const std::string& filepath,
        const IO::ImageWriteOptions& options = {});
 
    /**
     * @brief Wait for all in-flight save operations to complete.
     *
     * Called automatically by the destructor. Invoke explicitly to flush
     * pending saves before a checkpoint.
     */
    void wait_for_pending_saves();
 
    // ─────────────────────────────────────────────────────────────────────────
    // Volume — load
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Load a volume file into a newly constructed VolumeGridBuffer.
     *
     * Opens the file via IO::VolumeReader, reads every selected grid into
     * VolumeData, constructs a VolumeGridBuffer with one FieldDecl per
     * field — name and semantics from the file, stride sizeof(float) for a
     * scalar field or sizeof(glm::vec4) for a vector one, the GPU layout
     * every vector field uses regardless of what VolumeData stores — over
     * the file's own lattice, and seeds it via IO::upload_volume. Mirrors
     * load_mesh: construction and GPU registration only, nothing about
     * processing chain or rendering.
     *
     * setup_processors() and any simulation wiring via setup_flow() are
     * left to the caller, as VolumeGridBuffer's own usage example assumes.
     * Nothing about an imported file says whether the caller wants to keep
     * simulating what it loaded, run it once as a mask, or read it once and
     * discard the buffer — so nothing here decides that for them.
     *
     * The returned buffer is retained for the lifetime of IOManager; see
     * get_loaded_volumes().
     *
     * @param filepath Path to the volume file (.vdb).
     * @param options  Grid selection, forwarded to VolumeReader::load().
     * @return Constructed and seeded VolumeGridBuffer, or nullptr on failure.
     */
    [[nodiscard]] std::shared_ptr<Buffers::VolumeGridBuffer>
    load_volume(
        const std::string& filepath,
        const IO::VolumeReadOptions& options = {});
 
    /**
     * @brief Returns all VolumeGridBuffers created via load_volume().
     */
    [[nodiscard]] std::vector<std::shared_ptr<Buffers::VolumeGridBuffer>>
    get_loaded_volumes() const { return m_loaded_volumes; }
 
    // ─────────────────────────────────────────────────────────────────────────
    // Volume — save
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Save an already-downloaded VolumeData to disk asynchronously.
     *
     * Purely CPU-bound from here, so no thread restrictions on the caller.
     * Encode and disk flush are dispatched to a worker; failures are logged
     * but do not block.
     *
     * The extension of @p filepath selects the writer via
     * VolumeWriterRegistry. For synchronous semantics use the
     * IO::save_volume free function in VolumeTransfer.hpp, and for a numbered
     * frame sequence use IO::VolumeCapture.
     *
     * @return True if the encode task was queued.
     */
    bool save_volume(
        Kakshya::VolumeData data,
        const std::string& filepath,
        const IO::VolumeWriteOptions& options = {});
 
    /**
     * @brief Download and save a GPU-resident volume to disk asynchronously.
     *
     * Mirrors save_image(VKImage): download, writer lookup and write all
     * happen inside the dispatched task, so this returns as soon as the
     * task is queued rather than blocking the caller. The future is tracked
     * in the same pool as every other save task; wait_for_pending_saves()
     * waits on it like any other.
     *
     * @param volume      Volume to read from.
     * @param filepath    Destination path; extension selects the writer.
     * @param options     Writer options.
     * @param field_names Fields to save. Empty means every declared field.
     * @return True once the task is queued. Download or write failure is
     *         logged from the task and does not surface here.
     */
    bool save_volume(
        const std::shared_ptr<Buffers::VolumeGridBuffer>& volume,
        const std::string& filepath,
        const IO::VolumeWriteOptions& options = {},
        const std::vector<std::string>& field_names = {});
 
    /**
     * @brief Begin recording a numbered .vdb sequence from a volume.
     *
     * Constructs a VolumeCapture on the engine scheduler and retains it.
     * The capture spawns a GraphicsRoutine that reads one frame and
     * suspends on a FrameDelay, so it advances on the frame clock and
     * resumes on the thread that has command queue access. Nothing polls
     * it and nothing drives it.
     *
     * The readback stays on the graphics thread because it must. The
     * encode is dispatched to the save task pool, so the per-frame cost on
     * the graphics thread is the transfer alone.
     *
     * For a capture the caller owns and drives itself, construct
     * IO::VolumeCapture directly.
     *
     * @param volume         Volume to record.
     * @param path_pattern   Destination with one index field, such as
     *                       "smoke.{:04}.vdb". Zero padding matters.
     * @param field_names    Fields to record. Empty means all.
     * @param options        Writer options, applied to every frame.
     * @param max_frames     Stop after this many frames. Zero is unbounded.
     * @param frame_interval Frames between captures.
     * @return Capture handle for stop_volume_capture, or 0 on failure.
     */
    [[nodiscard]] uint32_t capture_volume(
        const std::shared_ptr<Buffers::VolumeGridBuffer>& volume,
        const std::string& path_pattern,
        const std::vector<std::string>& field_names = {},
        const IO::VolumeWriteOptions& options = {},
        uint32_t max_frames = 0,
        uint64_t frame_interval = 1);
 
    /**
     * @brief Stop a running volume capture and release it.
     *
     * Frames already queued for encode still complete. No-op with a
     * warning if the id is unknown.
     */
    void stop_volume_capture(uint32_t capture_id);
 
    /**
     * @brief Returns all active volume capture IDs.
     */
    [[nodiscard]] std::vector<uint32_t> get_volume_capture_ids() const;
 
    // ─────────────────────────────────────────────────────────────────────────
    // Spatial — capture
    // ─────────────────────────────────────────────────────────────────────────
 
    /**
     * @brief Begin (or join) recording time-sampled streams into an Alembic
     *        archive.
     *
     * Resolves @p filepath to a shared SpatialCache: the first call for a
     * given path opens it, and a later call passing the same path reuses the
     * cache already open, so several independent captures can each record
     * their own streams into one archive. Unlike capture_volume, nothing is
     * numbered per frame: Alembic carries time natively, so every sample of
     * every stream lands in the one file across the whole recording.
     *
     * Spawns a SpatialCapture on the engine scheduler and retains it, the
     * same GraphicsRoutine/FrameDelay shape capture_volume uses. For a
     * capture the caller owns and drives itself, construct IO::SpatialCapture
     * directly against a SpatialCache of its own.
     *
     * @param filepath       Destination .abc path.
     * @param sources        Streams to tick together every frame. Build one
     *                       with IO::make_network_geometry_source() or a
     *                       hand-written IO::SpatialCaptureSource.
     * @param max_frames     Stop after this many frames. Zero is unbounded.
     * @param frame_interval Frames between captures.
     * @return Capture handle for stop_spatial_capture, or 0 on failure
     *         (no sources, no scheduler, or the archive failed to open).
     */
    [[nodiscard]] uint32_t capture_spatial(
        const std::string& filepath,
        std::vector<IO::SpatialCaptureSource> sources,
        uint32_t max_frames = 0,
        uint64_t frame_interval = 1);
 
    /**
     * @brief Stop a running spatial capture and release it.
     *
     * Frames already written are kept. The underlying archive stays open and
     * registered under its filepath as long as any other capture still
     * shares it. No-op with a warning if the id is unknown.
     */
    void stop_spatial_capture(uint32_t capture_id);
 
    /**
     * @brief Returns all active spatial capture IDs.
     */
    [[nodiscard]] std::vector<uint32_t> get_spatial_capture_ids() const;
 
    /**
     * @brief Save every source once, as a single time sample, with a
     *        millisecond epoch timestamp spliced into the path, queued
     *        asynchronously.
     *
     * The timestamp is computed on the worker thread at the moment the
     * write actually runs, mirroring save_mesh_snapshot, so repeated calls
     * in quick succession still each land in their own file.
     *
     * @return True once the task is queued; open or write failure is
     *         logged from the task and does not surface here.
     */
    bool save_spatial_snapshot(
        std::vector<IO::SpatialCaptureSource> sources,
        const std::string& path_pattern);
 
    /**
     * @brief Returns all active video reader IDs.
     */
    [[nodiscard]] std::vector<uint64_t> get_video_reader_ids() const;
 
    /**
     * @brief Returns the video reader assigned to the given ID, or nullptr.
     */
    [[nodiscard]] std::shared_ptr<VideoFileReader> get_video_reader(uint64_t id) const;
 
    /**
     * @brief Returns all active camera reader IDs.
     */
    [[nodiscard]] std::vector<uint64_t> get_camera_reader_ids() const;
 
    /**
     * @brief Returns the camera source assigned to the given ID, or nullptr.
     */
    [[nodiscard]] std::shared_ptr<CameraSource> get_camera_reader(uint64_t id) const;
 
    /**
     * @brief Returns all retained audio readers.
     */
    [[nodiscard]] std::vector<std::shared_ptr<SoundFileReader>> get_audio_readers() const { return m_audio_readers; };
 
    /**
     * @brief Returns all retained image readers.
     */
    [[nodiscard]] std::vector<std::shared_ptr<ImageReader>> get_image_readers() const { return m_image_readers; };
 
    /**
     * @brief Returns all retained model readers.
     */
    [[nodiscard]] std::vector<std::shared_ptr<ModelReader>> get_model_readers() const { return m_model_readers; };
 
    /**
     * @brief Returns all active audio capture IDs.
     */
    [[nodiscard]] std::vector<uint32_t> get_audio_capture_ids() const;
 
    /**
     * @brief Returns all registered SoundFileWriters created via create_writer() or write().
     */
    [[nodiscard]] std::vector<std::shared_ptr<SoundFileWriter>> get_sound_writers() const { return m_writers; };
 
private:
    Core::GlobalStreamInfo& m_stream_info;
    uint32_t m_frame_rate;
 
    std::atomic<bool> m_stop_done { false };
    std::atomic<bool> m_shutdown_done { false };
 
    /**
     * @brief IOService::request_decode target — shared-lock lookup + signal_decode().
     * Non-blocking. Safe from any thread.
     */
    void dispatch_decode_request(uint64_t reader_id);
 
    /**
     * @brief IOService::request_frame target — shared-lock lookup + pull_frame_all().
     * Non-blocking. Safe from any thread.
     */
    void dispatch_frame_request(uint64_t reader_id);
 
    /**
     * @brief Retain a save task in m_save_tasks and opportunistically reclaim
     *        finished ones.
     *
     * Lock-free: delegates to Parallel::AsyncGroup, which never blocks one
     * caller on another's future. Failed saves are logged through the
     * group's exception handler set up in the constructor, which reuses
     * Journal::error_rethrow for the catch(const std::exception&) vs
     * catch(...) message dispatch and then swallows the rethrow, since
     * AsyncGroup's contract is to report a task's exception, not propagate
     * it into whichever thread happens to call submit()/drain() next.
     * Nothing about a task's own work (already running on its own thread
     * via std::async before this is called) is affected by tracking it here.
     */
    void track_save_task(std::future<bool> fut);
 
    void configure_frame_processor(
        const std::shared_ptr<Kakshya::VideoFileContainer>& container) const;
 
    void configure_audio_processor(
        const std::shared_ptr<Kakshya::SoundFileContainer>& container) const;
 
    // ── Audio capture ──────────────────────────────────────────────────────
 
    struct AudioCaptureState {
        std::shared_ptr<Kakshya::AudioOutputContainer> container;
        std::shared_ptr<SoundFileWriter> writer;
        uint32_t observer_id {};
    };
 
    mutable std::mutex m_audio_captures_mutex;
    std::unordered_map<uint32_t, AudioCaptureState> m_audio_captures;
 
    // ── Audio capture ──────────────────────────────────────────────────────
 
    struct VideoCaptureState {
        std::shared_ptr<VideoFileWriter> writer;
        uint32_t capture_id {};
    };
 
    mutable std::mutex m_video_captures_mutex;
    std::unordered_map<uint32_t, VideoCaptureState> m_video_captures;
    std::atomic<uint32_t> m_next_video_capture_id { 1 };
 
    std::vector<std::shared_ptr<VideoFileWriter>> m_video_writers;
 
    // ── Volume ──────────────────────────────────────────────────────
 
    std::atomic<uint32_t> m_next_volume_capture_id { 1 };
    mutable std::mutex m_volume_captures_mutex;
    std::unordered_map<uint32_t, std::unique_ptr<IO::VolumeCapture>> m_volume_captures;
    std::vector<std::shared_ptr<Buffers::VolumeGridBuffer>> m_loaded_volumes;
 
    // ── Spatial ──────────────────────────────────────────────────────
 
    struct SpatialCaptureEntry {
        std::unique_ptr<IO::SpatialCapture> capture;
        std::string filepath; ///< Keys m_spatial_cache_refcounts.
    };
 
    std::atomic<uint32_t> m_next_spatial_capture_id { 1 };
    mutable std::mutex m_spatial_captures_mutex;
    std::unordered_map<uint32_t, SpatialCaptureEntry> m_spatial_captures;
    std::unordered_map<std::string, std::shared_ptr<IO::SpatialCache>> m_spatial_caches;
    std::unordered_map<std::string, int> m_spatial_cache_refcounts; ///< Closes+erases a cache at zero.
 
    // ── readers ──────────────────────────────────────────────────────
 
    std::atomic<uint64_t> m_next_reader_id { 1 };
    mutable std::shared_mutex m_readers_mutex;
    mutable std::shared_mutex m_camera_mutex;
    std::unordered_map<uint64_t, std::shared_ptr<VideoFileReader>> m_video_readers;
    std::unordered_map<uint64_t, std::shared_ptr<CameraSource>> m_camera_readers;
 
    std::vector<std::shared_ptr<SoundFileReader>> m_audio_readers;
 
    std::vector<std::shared_ptr<ImageReader>> m_image_readers;
 
    std::vector<std::shared_ptr<ModelReader>> m_model_readers;
 
    std::mutex m_writers_mutex;
    std::vector<std::shared_ptr<SoundFileWriter>> m_writers;
 
    // ── Stored buffers ─────────────────────────────────────────────────────
 
    mutable std::shared_mutex m_buffers_mutex;
    Parallel::AsyncGroup<bool> m_save_tasks;
 
    std::unordered_map<
        std::shared_ptr<Kakshya::VideoStreamContainer>,
        std::shared_ptr<Buffers::VideoContainerBuffer>>
        m_video_buffers;
 
    std::unordered_map<
        std::shared_ptr<Kakshya::CameraContainer>,
        std::shared_ptr<Buffers::VideoContainerBuffer>>
        m_camera_buffers;
 
    std::unordered_map<
        std::shared_ptr<Kakshya::VideoFileContainer>,
        std::shared_ptr<Kakshya::SoundFileContainer>>
        m_extracted_audio;
 
    std::unordered_map<
        std::shared_ptr<Kakshya::SoundFileContainer>,
        std::vector<std::shared_ptr<Buffers::SoundContainerBuffer>>>
        m_audio_buffers;
 
    std::shared_ptr<Buffers::BufferManager> m_buffer_manager;
 
    std::shared_ptr<Vruta::TaskScheduler> m_scheduler;
 
    // ── IOService ──────────────────────────────────────────────────────────
 
    std::shared_ptr<Registry::Service::IOService> m_io_service;
    Registry::Service::AudioBackendService* m_audio_backend_service {};
};
 
} // namespace MayaFlux::IO