konjure_lang/

live.rs

//! Deterministic, replayable command journal for a live language session.
//!
//! The journal is an in-memory protocol. It never reads files, invokes host
//! registries, or repeats external effects while replaying accepted commands.
//!
//! ```
//! use konjure_lang::{LiveCommand, LiveRuntime};
//! use std::collections::BTreeMap;
//!
//! let mut session = LiveRuntime::new();
//! let event = session.apply(LiveCommand::Load {
//!     source: "let ball = spawn(Sphere { radius: 1 });".into(),
//!     modules: BTreeMap::new(),
//! })?;
//! assert_eq!(event.sequence, 1);
//! let history = session.history().clone();
//! let mut replay = LiveRuntime::new();
//! replay.restore(history)?;
//! assert_eq!(replay.snapshot(), session.snapshot());
//! # Ok::<(), konjure_lang::LiveError>(())
//! ```

pub mod pairing;

use crate::{
    Snapshot, Value,
    spatial::{RenderOutput, SpatialRuntime},
};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::io::{self, Write};

/// Source text and explicitly supplied modules for one language revision.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct ProgramSource {
    /// Main module source text.
    pub source: String,
    /// Imported module source, keyed by its explicit module name.
    pub modules: BTreeMap<String, String>,
}

/// An ordered, serializable state change accepted by a live language session.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum LiveCommand {
    /// Compile and initialize a new source revision.
    Load {
        /// Main module source text.
        source: String,
        /// Explicitly supplied imported modules.
        modules: BTreeMap<String, String>,
    },
    /// Advance fixed simulation ticks.
    Step {
        /// Fixed ticks to advance.
        count: u32,
    },
    /// Invoke a callback stored in a component field.
    InvokeComponent {
        /// Stable entity identity.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        entity: u64,
        /// Nominal component key.
        component: String,
        /// Callback field name.
        field: String,
        /// Typed callback arguments.
        arguments: Vec<Value>,
    },
    /// Invoke a typed method on a live entity.
    InvokeEntity {
        /// Stable entity identity.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        entity: u64,
        /// Typed method name.
        method: String,
        /// Typed method arguments.
        arguments: Vec<Value>,
    },
    /// Run terminal lifecycle callbacks once.
    Finish,
}

/// A command with its globally monotonic session sequence number.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct LiveEvent {
    /// Sequence assigned when the command was accepted.
    #[cfg_attr(feature = "typescript", ts(type = "number"))]
    pub sequence: u64,
    /// The accepted command.
    pub command: LiveCommand,
}

/// A bounded retained suffix of a live session's event stream.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct LiveHistory {
    /// Sequence immediately before the first retained event.
    #[cfg_attr(feature = "typescript", ts(type = "number"))]
    pub base_sequence: u64,
    /// Accepted events in exact sequence order.
    pub events: Vec<LiveEvent>,
}

/// Authority granted to one connected live-session peer.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum LiveRole {
    /// May submit state-changing commands.
    Owner,
    /// May receive and replay accepted events.
    Participant,
}

/// Runtime state reported by a connected live-session client.
///
/// This is observational telemetry only. It never changes the accepted command
/// journal, playback state, or generated scene.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum LiveDebugStatus {
    /// The client is advancing an unfinished runtime.
    Running,
    /// The client has a loaded runtime that is not advancing.
    Paused,
    /// The client has run its terminal lifecycle callbacks.
    Finished,
    /// The client can report transport progress but not runtime state.
    Unavailable,
}

/// Bounded runtime and render telemetry supplied by one connected client.
///
/// Hosts validate this report before retaining it. It is not a debugger and
/// does not carry source, traces, logs, commands, or scene state.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct LiveDebugReport {
    /// The client's accepted runtime tick, when it can observe one.
    #[cfg_attr(feature = "typescript", ts(type = "number"))]
    pub runtime_tick: u64,
    /// The latest command sequence submitted to that client's renderer.
    #[cfg_attr(feature = "typescript", ts(type = "number"))]
    pub presented_sequence: u64,
    /// Recent local render rate, if the client measures one.
    pub render_fps: Option<f32>,
    /// The extent to which this client can observe its runtime.
    pub status: LiveDebugStatus,
}

/// Public progress information for one live-session connection.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct LivePeer {
    /// Host-assigned stable connection identifier.
    pub id: String,
    /// Display name supplied during hello.
    pub name: String,
    /// Authority assigned by the host, independent of the client-supplied name.
    pub role: LiveRole,
    /// Last known accepted event sequence, when the peer has loaded a revision.
    #[cfg_attr(feature = "typescript", ts(type = "number | null"))]
    pub sequence: Option<u64>,
    /// The latest admitted runtime and render telemetry for this connection.
    pub debug: Option<LiveDebugReport>,
}

/// Client-to-host live-session wire message.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum LiveClientMessage {
    /// Authenticates and names a new connection.
    Hello {
        /// Session access token.
        token: String,
        /// Human-readable connection name.
        name: String,
    },
    /// Requests an owner-authorized live command.
    Command {
        /// Client-scoped identifier echoed by the host response.
        request_id: String,
        /// Requested runtime command.
        command: LiveCommand,
    },
    /// Replaces the program only when the caller based it on the current source revision.
    EditSource {
        /// Client-scoped identifier echoed by the host response.
        request_id: String,
        /// Sequence of the last accepted [`LiveCommand::Load`], not a simulation tick.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        base_revision: u64,
        /// Complete source replacement candidate.
        source: ProgramSource,
    },
    /// Requests a checked semantic UI action from an authenticated peer.
    Ui {
        /// Client-scoped identifier echoed by the host response.
        request_id: String,
        /// Event emitted against the view's displayed semantic UI revision.
        event: konjure_sdk::presentation::UiEvent,
    },
    /// Starts or stops host-coordinated playback.
    Playback {
        /// Client-scoped identifier echoed by the host response.
        request_id: String,
        /// Whether playback should advance.
        playing: bool,
    },
    /// Reports the latest event sequence applied by this client.
    Ack {
        /// Last exact event sequence applied.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        sequence: u64,
    },
    /// Publishes bounded runtime and render telemetry without changing the session.
    Debug {
        /// Latest client-local runtime and render progress.
        report: LiveDebugReport,
    },
    /// Requests a clean connection close.
    Close {
        /// Client-scoped identifier echoed by the close response.
        request_id: String,
    },
}

/// Host-to-client live-session wire message.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum LiveServerMessage {
    /// Establishes role, retained journal, playback state, and peers.
    Welcome {
        /// Authority assigned to this connection.
        role: LiveRole,
        /// Replayable retained journal suffix.
        history: LiveHistory,
        /// Current host playback state.
        playing: bool,
        /// Visible connected peers.
        peers: Vec<LivePeer>,
    },
    /// Broadcasts one accepted event, optionally replying to a request.
    Event {
        /// Accepted runtime event.
        event: LiveEvent,
        /// Originating client request, when applicable.
        request_id: Option<String>,
    },
    /// Broadcasts a playback state change.
    Playback {
        /// Current host playback state.
        playing: bool,
    },
    /// Broadcasts the visible peer list.
    Peers {
        /// Connected peers and reported progress.
        peers: Vec<LivePeer>,
    },
    /// Reports a rejected request without changing accepted runtime state.
    Error {
        /// Human-readable rejection reason.
        message: String,
        /// Originating client request, when applicable.
        request_id: Option<String>,
    },
    /// Confirms connection closure.
    Closed,
}

/// A rejected live command or replay history. Rejections preserve committed state.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum LiveError {
    /// Source plus imported modules exceeds the revision input budget.
    SourceLimit,
    /// Encoded callback or method arguments exceed the command budget.
    ArgumentsLimit,
    /// A session exceeds its total fixed-tick budget.
    TickLimit,
    /// The retained event count has reached its fixed bound.
    EventLimit,
    /// The retained journal cannot fit its fixed byte budget.
    JournalLimit,
    /// Replay input was not the next exact sequence number.
    Sequence {
        /// Required next global sequence number.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        expected: u64,
        /// Sequence supplied by the rejected event.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        actual: u64,
    },
    /// A source replacement was based on a different accepted source revision.
    SourceRevision {
        /// The current accepted source revision.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        expected: u64,
        /// The source revision supplied by the editor.
        #[cfg_attr(feature = "typescript", ts(type = "number"))]
        actual: u64,
    },
    /// A runtime compile, validation, or lifecycle action failed.
    Runtime {
        /// Compact compiler or runtime error details for protocol consumers.
        diagnostics: Vec<LiveDiagnostic>,
    },
    /// A state-dependent operation was requested before a source revision loaded.
    NotLoaded,
}

impl std::fmt::Display for LiveError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::SourceLimit => formatter.write_str("live source budget exceeded"),
            Self::ArgumentsLimit => formatter.write_str("live argument budget exceeded"),
            Self::TickLimit => formatter.write_str("live simulation tick budget exceeded"),
            Self::EventLimit => formatter.write_str("live event budget exceeded"),
            Self::JournalLimit => formatter.write_str("live journal budget exceeded"),
            Self::NotLoaded => formatter.write_str("no live program is loaded"),
            Self::Sequence { expected, actual } => {
                write!(formatter, "expected event {expected}, received {actual}")
            }
            Self::SourceRevision { expected, actual } => write!(
                formatter,
                "source revision conflict: current {expected}, draft based on {actual}"
            ),
            Self::Runtime { diagnostics } => {
                for (index, diagnostic) in diagnostics.iter().enumerate() {
                    if index > 0 {
                        formatter.write_str("; ")?;
                    }
                    write!(formatter, "{}: {}", diagnostic.code, diagnostic.message)?;
                }
                Ok(())
            }
        }
    }
}

impl std::error::Error for LiveError {}

/// Serializable diagnostic detail carried by [`LiveError::Runtime`].
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct LiveDiagnostic {
    /// Stable language diagnostic code.
    pub code: String,
    /// Human-readable diagnostic message.
    pub message: String,
}

/// Largest integer shared exactly by the Rust and JavaScript wire formats.
const MAX_EXACT_INTEGER: u64 = 9_007_199_254_740_991;

/// Maximum combined UTF-8 source and module bytes in one revision.
pub const MAX_SOURCE_BYTES: usize = 1024 * 1024;
/// Maximum serialized arguments in one invocation command.
pub const MAX_ARGUMENT_BYTES: usize = 64 * 1024;
/// Maximum events retained in one journal suffix.
pub const MAX_EVENTS: usize = 100_000;
/// Maximum fixed ticks accepted across one loaded revision.
pub const MAX_CUMULATIVE_TICKS: u64 = 216_000;
/// Maximum serialized retained journal bytes.
pub const MAX_JOURNAL_BYTES: usize = 16 * 1024 * 1024;

/// An explicitly stepped language runtime with a bounded, replayable journal.
pub struct LiveRuntime {
    runtime: Option<SpatialRuntime>,
    sequence: u64,
    source_revision: u64,
    history: LiveHistory,
    cumulative_ticks: u64,
    journal_bytes: usize,
}

impl Default for LiveRuntime {
    fn default() -> Self {
        Self::new()
    }
}

impl LiveRuntime {
    /// Creates an empty session. The first accepted command must be [`LiveCommand::Load`].
    #[must_use]
    pub fn new() -> Self {
        Self {
            runtime: None,
            sequence: 0,
            source_revision: 0,
            history: LiveHistory::default(),
            cumulative_ticks: 0,
            journal_bytes: history_envelope_bytes(0),
        }
    }

    /// Applies a command transactionally and appends its exact event to the journal.
    ///
    /// # Errors
    /// Rejects oversized inputs, invalid runtime actions, or a full journal without
    /// changing the accepted runtime, sequence, or history.
    pub fn apply(&mut self, command: LiveCommand) -> Result<LiveEvent, LiveError> {
        self.apply_next(command)
    }

    /// Accepts one replay event only when its sequence is exactly next.
    ///
    /// # Errors
    /// Rejects gaps, duplicates, and invalid commands without changing this session.
    pub fn accept(&mut self, event: LiveEvent) -> Result<(), LiveError> {
        let expected = self.sequence.checked_add(1).ok_or(LiveError::EventLimit)?;
        if event.sequence != expected {
            return Err(LiveError::Sequence {
                expected,
                actual: event.sequence,
            });
        }
        self.apply_next(event.command)?;
        Ok(())
    }

    /// Replays a complete retained history transactionally, then replaces this session.
    ///
    /// # Errors
    /// Rejects non-contiguous, oversized, or invalid histories while preserving state.
    pub fn restore(&mut self, history: LiveHistory) -> Result<(), LiveError> {
        validate_history(&history)?;
        let mut candidate = Self {
            sequence: history.base_sequence,
            history: LiveHistory {
                base_sequence: history.base_sequence,
                events: Vec::new(),
            },
            journal_bytes: history_envelope_bytes(history.base_sequence),
            ..Self::new()
        };
        for event in history.events {
            candidate.accept(event)?;
        }
        *self = candidate;
        Ok(())
    }

    /// Returns the accepted language state, if a revision has loaded.
    #[must_use]
    pub fn snapshot(&self) -> Option<Snapshot> {
        self.runtime.as_ref().map(SpatialRuntime::snapshot)
    }

    /// Returns the accepted runtime tick and terminal lifecycle state without
    /// cloning the scene or snapshot.
    #[must_use]
    pub fn execution_state(&self) -> Option<(u64, bool)> {
        self.runtime.as_ref().map(SpatialRuntime::execution_state)
    }

    /// Returns validated generated render data for the accepted language state.
    pub fn render(&self) -> Result<RenderOutput, LiveError> {
        let mut output = self
            .runtime
            .as_ref()
            .ok_or(LiveError::NotLoaded)?
            .render()
            .map_err(runtime_error)?;
        output.ui.revision = self.source_revision();
        Ok(output)
    }

    /// Returns the bounded retained journal suffix.
    #[must_use]
    pub fn history(&self) -> &LiveHistory {
        &self.history
    }

    /// Returns the last globally assigned session sequence number.
    #[must_use]
    pub const fn sequence(&self) -> u64 {
        self.sequence
    }

    /// Returns the sequence of the last accepted source load.
    ///
    /// Simulation ticks and invocations do not change this revision, so it can
    /// safely be used as an optimistic-concurrency precondition for source edits.
    #[must_use]
    pub const fn source_revision(&self) -> u64 {
        self.source_revision
    }

    /// Compiles and accepts a complete source replacement at one exact base revision.
    ///
    /// # Errors
    /// Rejects stale or invalid candidates without changing the accepted runtime,
    /// journal, source revision, or playback state owned by the host.
    pub fn edit_source(
        &mut self,
        base_revision: u64,
        source: ProgramSource,
    ) -> Result<LiveEvent, LiveError> {
        if base_revision != self.source_revision {
            return Err(LiveError::SourceRevision {
                expected: self.source_revision,
                actual: base_revision,
            });
        }
        self.apply(LiveCommand::Load {
            source: source.source,
            modules: source.modules,
        })
    }

    fn apply_next(&mut self, command: LiveCommand) -> Result<LiveEvent, LiveError> {
        validate_command(&command)?;
        let event = LiveEvent {
            sequence: self
                .sequence
                .checked_add(1)
                .filter(|value| *value <= MAX_EXACT_INTEGER)
                .ok_or(LiveError::EventLimit)?,
            command,
        };
        let event_bytes = encoded_len(&event, MAX_JOURNAL_BYTES)?;
        let (base_sequence, journal_bytes, resets_history) =
            if matches!(event.command, LiveCommand::Load { .. }) {
                let base_sequence = self.sequence;
                (
                    base_sequence,
                    history_envelope_bytes(base_sequence)
                        .checked_add(event_bytes)
                        .ok_or(LiveError::JournalLimit)?,
                    true,
                )
            } else {
                if self.history.events.len() >= MAX_EVENTS {
                    return Err(LiveError::EventLimit);
                }
                let separator = usize::from(!self.history.events.is_empty());
                (
                    self.history.base_sequence,
                    self.journal_bytes
                        .checked_add(separator)
                        .and_then(|bytes| bytes.checked_add(event_bytes))
                        .ok_or(LiveError::JournalLimit)?,
                    false,
                )
            };
        if journal_bytes > MAX_JOURNAL_BYTES {
            return Err(LiveError::JournalLimit);
        }
        self.execute(&event.command)?;
        if resets_history {
            self.history = LiveHistory {
                base_sequence,
                events: vec![event.clone()],
            }
        } else {
            self.history.events.push(event.clone());
        }
        self.sequence = event.sequence;
        if matches!(event.command, LiveCommand::Load { .. }) {
            self.source_revision = event.sequence;
        }
        self.journal_bytes = journal_bytes;
        Ok(event)
    }

    fn execute(&mut self, command: &LiveCommand) -> Result<(), LiveError> {
        match command {
            LiveCommand::Load { source, modules } => {
                self.runtime = Some(SpatialRuntime::new(source, modules).map_err(runtime_errors)?);
                self.cumulative_ticks = 0;
            }
            LiveCommand::Step { count } => {
                let ticks = self
                    .cumulative_ticks
                    .checked_add(u64::from(*count))
                    .ok_or(LiveError::TickLimit)?;
                if ticks > MAX_CUMULATIVE_TICKS {
                    return Err(LiveError::TickLimit);
                }
                self.runtime
                    .as_mut()
                    .ok_or(LiveError::NotLoaded)?
                    .step(*count)
                    .map_err(runtime_error)?;
                self.cumulative_ticks = ticks;
            }
            LiveCommand::InvokeComponent {
                entity,
                component,
                field,
                arguments,
            } => {
                self.runtime
                    .as_mut()
                    .ok_or(LiveError::NotLoaded)?
                    .invoke_component(*entity, component, field, arguments)
                    .map_err(runtime_error)?;
            }
            LiveCommand::InvokeEntity {
                entity,
                method,
                arguments,
            } => {
                self.runtime
                    .as_mut()
                    .ok_or(LiveError::NotLoaded)?
                    .invoke_entity(*entity, method, arguments)
                    .map_err(runtime_error)?;
            }
            LiveCommand::Finish => {
                self.runtime
                    .as_mut()
                    .ok_or(LiveError::NotLoaded)?
                    .finish()
                    .map_err(runtime_error)?;
            }
        }
        Ok(())
    }
}

fn validate_command(command: &LiveCommand) -> Result<(), LiveError> {
    match command {
        LiveCommand::Load { source, modules } => {
            let size = modules
                .iter()
                .try_fold(source.len(), |total, (name, value)| {
                    total.checked_add(name.len())?.checked_add(value.len())
                })
                .ok_or(LiveError::SourceLimit)?;
            if size > MAX_SOURCE_BYTES {
                return Err(LiveError::SourceLimit);
            }
        }
        LiveCommand::InvokeComponent {
            entity, arguments, ..
        }
        | LiveCommand::InvokeEntity {
            entity, arguments, ..
        } => {
            if *entity == 0 || *entity > MAX_EXACT_INTEGER {
                return Err(LiveError::ArgumentsLimit);
            }
            let argument_bytes = encoded_len(arguments, MAX_ARGUMENT_BYTES)
                .map_err(|_| LiveError::ArgumentsLimit)?;
            if argument_bytes > MAX_ARGUMENT_BYTES {
                return Err(LiveError::ArgumentsLimit);
            }
        }
        LiveCommand::Step { .. } | LiveCommand::Finish => {}
    }
    Ok(())
}

fn validate_history(history: &LiveHistory) -> Result<(), LiveError> {
    if history.base_sequence > MAX_EXACT_INTEGER {
        return Err(LiveError::EventLimit);
    }
    if history.events.len() > MAX_EVENTS {
        return Err(LiveError::EventLimit);
    }
    if encoded_len(history, MAX_JOURNAL_BYTES)? > MAX_JOURNAL_BYTES {
        return Err(LiveError::JournalLimit);
    }
    let mut expected = history.base_sequence;
    let mut cumulative_ticks = 0_u64;
    for event in &history.events {
        expected = expected
            .checked_add(1)
            .filter(|value| *value <= MAX_EXACT_INTEGER)
            .ok_or(LiveError::EventLimit)?;
        if event.sequence != expected {
            return Err(LiveError::Sequence {
                expected,
                actual: event.sequence,
            });
        }
        validate_command(&event.command)?;
        match &event.command {
            LiveCommand::Load { .. } => cumulative_ticks = 0,
            LiveCommand::Step { count } => {
                cumulative_ticks = cumulative_ticks
                    .checked_add(u64::from(*count))
                    .ok_or(LiveError::TickLimit)?;
                if cumulative_ticks > MAX_CUMULATIVE_TICKS {
                    return Err(LiveError::TickLimit);
                }
            }
            LiveCommand::InvokeComponent { .. }
            | LiveCommand::InvokeEntity { .. }
            | LiveCommand::Finish => {}
        }
    }
    Ok(())
}

fn history_envelope_bytes(base_sequence: u64) -> usize {
    serde_json::to_vec(&LiveHistory {
        base_sequence,
        events: Vec::new(),
    })
    .expect("live history envelope is serializable")
    .len()
}

fn encoded_len(value: &impl Serialize, limit: usize) -> Result<usize, LiveError> {
    let mut writer = LimitedWriter { written: 0, limit };
    serde_json::to_writer(&mut writer, value).map_err(|_| LiveError::JournalLimit)?;
    Ok(writer.written)
}

struct LimitedWriter {
    written: usize,
    limit: usize,
}

impl Write for LimitedWriter {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        let remaining = self.limit.saturating_sub(self.written);
        if buffer.len() > remaining {
            return Err(io::Error::new(
                io::ErrorKind::WriteZero,
                "live journal exceeds limit",
            ));
        }
        self.written += buffer.len();
        Ok(buffer.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

fn runtime_error(diagnostic: crate::Diagnostic) -> LiveError {
    LiveError::Runtime {
        diagnostics: vec![LiveDiagnostic {
            code: diagnostic.code,
            message: diagnostic.message,
        }],
    }
}

fn runtime_errors(diagnostics: Vec<crate::Diagnostic>) -> LiveError {
    LiveError::Runtime {
        diagnostics: diagnostics
            .into_iter()
            .map(|diagnostic| LiveDiagnostic {
                code: diagnostic.code,
                message: diagnostic.message,
            })
            .collect(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const SOURCE: &str = include_str!("../examples/lifecycle.kj");

    fn load(runtime: &mut LiveRuntime) {
        runtime
            .apply(LiveCommand::Load {
                source: SOURCE.into(),
                modules: BTreeMap::new(),
            })
            .unwrap();
    }

    #[test]
    fn journal_replay_produces_the_same_snapshot_and_render() {
        let mut original = LiveRuntime::new();
        assert_eq!(original.execution_state(), None);
        load(&mut original);
        assert_eq!(original.execution_state(), Some((0, false)));
        original.apply(LiveCommand::Step { count: 2 }).unwrap();
        assert_eq!(original.execution_state(), Some((2, false)));
        original.apply(LiveCommand::Finish).unwrap();
        assert_eq!(original.execution_state(), Some((2, true)));
        let history = original.history().clone();
        let snapshot = original.snapshot();
        let render = serde_json::to_value(original.render().unwrap()).unwrap();

        let mut replay = LiveRuntime::new();
        replay.restore(history).unwrap();

        assert_eq!(replay.snapshot(), snapshot);
        assert_eq!(
            serde_json::to_value(replay.render().unwrap()).unwrap(),
            render
        );
    }

    #[test]
    fn rejected_load_action_and_sequence_preserve_accepted_state() {
        let mut runtime = LiveRuntime::new();
        load(&mut runtime);
        let before = runtime.snapshot();
        let history = runtime.history().clone();
        let sequence = runtime.sequence();

        assert!(matches!(
            runtime.apply(LiveCommand::Load {
                source: "x".repeat(MAX_SOURCE_BYTES + 1),
                modules: BTreeMap::new(),
            }),
            Err(LiveError::SourceLimit)
        ));
        assert!(matches!(
            runtime.apply(LiveCommand::InvokeEntity {
                entity: 999,
                method: "missing".into(),
                arguments: Vec::new(),
            }),
            Err(LiveError::Runtime { .. })
        ));
        assert!(matches!(
            runtime.accept(LiveEvent {
                sequence: sequence + 2,
                command: LiveCommand::Step { count: 1 },
            }),
            Err(LiveError::Sequence { .. })
        ));

        assert_eq!(runtime.snapshot(), before);
        assert_eq!(runtime.history(), &history);
        assert_eq!(runtime.sequence(), sequence);
    }

    #[test]
    fn source_edits_use_the_last_load_sequence_not_the_latest_event() {
        let mut runtime = LiveRuntime::new();
        load(&mut runtime);
        let base_revision = runtime.source_revision();
        runtime.apply(LiveCommand::Step { count: 2 }).unwrap();
        assert_eq!(runtime.source_revision(), base_revision);

        let first = runtime
            .edit_source(
                base_revision,
                ProgramSource {
                    source: SOURCE.into(),
                    modules: BTreeMap::new(),
                },
            )
            .unwrap();
        assert_eq!(runtime.source_revision(), first.sequence);
        assert!(matches!(
            runtime.edit_source(
                base_revision,
                ProgramSource {
                    source: SOURCE.into(),
                    modules: BTreeMap::new(),
                },
            ),
            Err(LiveError::SourceRevision {
                expected,
                actual,
            }) if expected == first.sequence && actual == base_revision
        ));
    }

    #[test]
    fn invalid_source_edit_preserves_runtime_and_source_revision() {
        let mut runtime = LiveRuntime::new();
        load(&mut runtime);
        runtime.apply(LiveCommand::Step { count: 1 }).unwrap();
        let before_snapshot = runtime.snapshot();
        let before_history = runtime.history().clone();
        let before_sequence = runtime.sequence();
        let before_revision = runtime.source_revision();

        assert!(matches!(
            runtime.edit_source(
                before_revision,
                ProgramSource {
                    source: "not valid Konjure".into(),
                    modules: BTreeMap::new(),
                },
            ),
            Err(LiveError::Runtime { .. })
        ));
        assert_eq!(runtime.snapshot(), before_snapshot);
        assert_eq!(runtime.history(), &before_history);
        assert_eq!(runtime.sequence(), before_sequence);
        assert_eq!(runtime.source_revision(), before_revision);
    }

    #[test]
    fn caps_reject_replay_and_cumulative_ticks_without_mutation() {
        let mut runtime = LiveRuntime::new();
        load(&mut runtime);
        runtime.cumulative_ticks = MAX_CUMULATIVE_TICKS - 1;
        let before = runtime.snapshot();
        assert_eq!(
            runtime.apply(LiveCommand::Step { count: 2 }),
            Err(LiveError::TickLimit)
        );
        assert_eq!(runtime.snapshot(), before);

        let oversized = LiveHistory {
            base_sequence: 0,
            events: vec![LiveEvent {
                sequence: 1,
                command: LiveCommand::Load {
                    source: "x".repeat(MAX_SOURCE_BYTES + 1),
                    modules: BTreeMap::new(),
                },
            }],
        };
        assert_eq!(runtime.restore(oversized), Err(LiveError::SourceLimit));
        assert_eq!(runtime.snapshot(), before);

        let too_many_events = LiveHistory {
            base_sequence: 0,
            events: vec![
                LiveEvent {
                    sequence: 1,
                    command: LiveCommand::Step { count: 0 },
                };
                MAX_EVENTS + 1
            ],
        };
        assert_eq!(runtime.restore(too_many_events), Err(LiveError::EventLimit));
        assert_eq!(runtime.snapshot(), before);
    }

    #[test]
    fn cached_journal_size_matches_serialized_history_after_append_restore_and_load() {
        let mut runtime = LiveRuntime::new();
        let check = |runtime: &LiveRuntime| {
            assert_eq!(
                runtime.journal_bytes,
                serde_json::to_vec(runtime.history()).unwrap().len()
            );
        };
        check(&runtime);
        load(&mut runtime);
        check(&runtime);
        for _ in 0..12 {
            runtime.apply(LiveCommand::Step { count: 1 }).unwrap();
            check(&runtime);
        }
        let mut restored = LiveRuntime::new();
        restored.restore(runtime.history().clone()).unwrap();
        check(&restored);
        load(&mut restored);
        check(&restored);
    }

    #[test]
    fn finish_runs_lifecycle_once_even_when_the_command_repeats() {
        let mut runtime = LiveRuntime::new();
        load(&mut runtime);
        runtime.apply(LiveCommand::Step { count: 2 }).unwrap();
        runtime.apply(LiveCommand::Finish).unwrap();
        runtime.apply(LiveCommand::Finish).unwrap();
        let snapshot = runtime.snapshot().unwrap();
        assert!(snapshot.finished);
        assert_eq!(snapshot.logs, ["Started", "Finished"]);
    }
}