Module compose 

Declarative incremental composition for live frame sources.

crate::declare::Recon runs a finite, pre-discovered folder to completion. This module is the live counterpart: a retained node (ReconNode) owns a Session and its Pipeline, and the caller only declares the current input state (FrameState). The node reconciles each declaration against what it has already accepted and does the minimal incremental work:

declared state	node action
new synchronized set (next sequence)	ingest one position
same key and revision again	nothing (Outcome::Duplicate)
set with cameras missing, or FrameState::Pending	wait, ingest nothing (Outcome::Waiting)
sequence jump (dropped frames)	ingest, report the gap (Outcome::Appended{ skipped })
late arrival of an older sequence	per ReconcilePolicy
higher revision of an accepted sequence	per ReconcilePolicy
FrameState::End / ReconNode::close	close open zones, drain refinement, final events

Frames are identified by FrameKey (source_id, sequence, revision), not by path. Each accepted sequence becomes one position, in sequence order. Payloads can be files under the image folder or in-memory images (FramePayload); in-memory images are written atomically to a spool folder inside the image folder, because the reconstruction core reads images by name.

Events are the ordinary pipeline events (hooks registered on the builder, or ReconNode::subscribe). Reconciliation decisions that drop or reorder input are also reported as Event::Warning.

Example

use cumulus3d_cli::compose::{reconstruct, CompositionHost, FrameKey, FrameState, LiveFrameSet, ReconNode, ReconcilePolicy};
use cumulus3d_cli::declare::Dense;

fn main() -> Result<(), cumulus3d_cli::compose::ComposeError> {
    let host = CompositionHost::new();
    let recon = ReconNode::declare()
        .images("live/images")
        .cameras(["camF", "camR", "camL"])
        .zones(12, 2)
        .dense(Dense::profile("fast").fusion_min_views(3))
        .reconcile(ReconcilePolicy::ReplayWindow { positions: 4 })
        .on_zone_preview(|range, cloud| println!("preview {}: {} points", range.zone, cloud.len()))
        .build(&host)?;

    let set = |seq: u64| {
        LiveFrameSet::new(FrameKey::new("drone-1", seq, 0))
            .file("camF", format!("camF/camF_{:04}.jpg", seq * 3))
            .file("camR", format!("camR/camR_{:04}.jpg", seq * 3))
            .file("camL", format!("camL/camL_{:04}.jpg", seq * 3))
    };
    recon.compose(FrameState::ready(set(0)))?;
    recon.compose(FrameState::ready(set(1)))?;
    recon.compose(FrameState::ready(set(1)))?; // no-op: same identity and revision
    // The same node, found by id (positional-memoization style):
    reconstruct(&host, "default", FrameState::ready(set(2)))?;
    let closed = recon.close()?;
    println!("{} events, {:?}", closed.summary.events, closed.stats);
    Ok(())
}

For a producer thread (decoder, network receiver) use ReconNode::spawn: the node then runs on its own thread behind a bounded queue with an explicit Backpressure policy, and the producer only calls ComposeHandle::send / ComposeHandle::try_send (both usable from async code; try_send never blocks).

Structs

Closed

What ReconNode::close returns.

ComposeHandle

Producer side of a spawned node (ReconNode::spawn).

ComposeStats

Reconciliation counters.

CompositionHost

Owner of retained nodes, addressed by id. Cheap to clone (shared).

FrameKey

Stable identity of one synchronized frame set.

LiveFrameSet

One synchronized frame set (one camera frame per camera) with its identity.

LiveImage

One camera frame of a synchronized set.

NodeBuilder

Records a node configuration; NodeBuilder::build checks it and creates the node on a host.

QueueStats

Queue counters of a spawned node.

ReconNode

A retained reconstruction node: session, pipeline, accepted-frame journal and lifecycle. Cheap to clone (shared); all clones address the same node.

Enums

Backpressure

Queue policy between a producer and a spawned node (ReconNode::spawn).

ComposeError

Errors.

EncodedFormat

Encoded image container for FramePayload::Encoded.

FramePayload

Where the pixels of one camera frame come from.

FrameState

Declared input state.

IgnoreReason

Why a declared state was not applied.

Outcome

Result of one declaration.

ReconcilePolicy

What to do with input that is not a plain append (late arrivals and corrected revisions).

SendError

Why a state could not be queued.

Functions

reconstruct

Declares state on the host’s node id: the declarative entry point (reconstruct(&host, "main", FrameState::ready(set)) on every new input).