diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs
index 267b330a..789a0b24 100644
--- a/src/GameCult.Mesh/CultMeshPrimitives.cs
+++ b/src/GameCult.Mesh/CultMeshPrimitives.cs
@@ -1052,6 +1052,26 @@ namespace GameCult.Mesh
return next;
}
+ ///
+ /// Creates a managed reactive document mirror whose local edits can be coalesced into one prediction or replacement.
+ ///
+ public async Task> ReactiveAsync(
+ CultMeshReactiveDocumentOptions? options = null)
+ {
+ var current = CloneDocument(await LatestAsync().ConfigureAwait(false));
+ var reactive = new CultMeshReactiveDocument(this, current, options);
+ reactive.Start();
+ return reactive;
+ }
+
+ ///
+ /// Creates a managed reactive document mirror synchronously for host APIs that cannot be async.
+ ///
+ public CultMeshReactiveDocument Reactive(CultMeshReactiveDocumentOptions? options = null)
+ {
+ return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult();
+ }
+
/// Creates a same-schema alias presentation for another CLR document type.
public CultMeshDocumentHandle AsSchemaAlias() where TAlias : class
{
@@ -1085,7 +1105,17 @@ namespace GameCult.Mesh
return new CultMeshDocumentHandle(aliasFeed.Bind(Context), replace, submitPrediction);
}
- private static TTarget ConvertDocument(TSource document)
+ internal static TDocumentValue CloneDocument(TDocumentValue document)
+ where TDocumentValue : class
+ {
+ if (document == null) throw new ArgumentNullException(nameof(document));
+ var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TDocumentValue));
+ return (TDocumentValue)CultDocumentMessagePackSerialization.DeserializeUntyped(
+ typeof(TDocumentValue),
+ payload);
+ }
+
+ internal static TTarget ConvertDocument(TSource document)
where TSource : class
where TTarget : class
{
@@ -1100,6 +1130,256 @@ namespace GameCult.Mesh
}
}
+ ///
+ /// Configures a managed CultMesh reactive document mirror.
+ ///
+ public sealed class CultMeshReactiveDocumentOptions
+ {
+ ///
+ /// Gets or sets the write coalescing window. The default approximates a frame boundary for non-Unity hosts.
+ ///
+ public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16);
+
+ /// Gets or sets whether canonical snapshots replace dirty local predictions immediately.
+ public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; }
+ }
+
+ ///
+ /// Captures a canonical snapshot that arrived while the local reactive document had an outstanding prediction.
+ ///
+ public sealed class CultMeshReactiveDocumentReconciliation
+ where TDocument : class
+ {
+ internal CultMeshReactiveDocumentReconciliation(
+ TDocument canonical,
+ TDocument predicted,
+ DateTimeOffset receivedAt)
+ {
+ Canonical = canonical;
+ Predicted = predicted;
+ ReceivedAt = receivedAt;
+ }
+
+ /// Gets the authoritative canonical document snapshot.
+ public TDocument Canonical { get; }
+
+ /// Gets the locally predicted document snapshot that was active when the canonical value arrived.
+ public TDocument Predicted { get; }
+
+ /// Gets when the canonical snapshot was received.
+ public DateTimeOffset ReceivedAt { get; }
+ }
+
+ ///
+ /// Managed typed document mirror that keeps an editable current value and coalesces local edits into CultMesh mutations.
+ ///
+ public sealed class CultMeshReactiveDocument : IDisposable
+ where TDocument : class
+ {
+ private readonly CultMeshDocumentHandle _document;
+ private readonly CultMeshReactiveDocumentOptions _options;
+ private readonly object _gate = new();
+ private IDisposable? _subscription;
+ private Timer? _timer;
+ private bool _dirty;
+ private bool _flushQueued;
+ private bool _flushing;
+ private bool _disposed;
+
+ internal CultMeshReactiveDocument(
+ CultMeshDocumentHandle document,
+ TDocument current,
+ CultMeshReactiveDocumentOptions? options)
+ {
+ _document = document ?? throw new ArgumentNullException(nameof(document));
+ Current = current ?? throw new ArgumentNullException(nameof(current));
+ _options = options ?? new CultMeshReactiveDocumentOptions();
+ }
+
+ /// Gets the underlying document handle.
+ public CultMeshDocumentHandle Document => _document;
+
+ /// Gets the editable local document value.
+ public TDocument Current { get; private set; }
+
+ /// Gets whether local edits are waiting to be sent.
+ public bool IsDirty
+ {
+ get
+ {
+ lock (_gate)
+ return _dirty;
+ }
+ }
+
+ /// Gets the most recent reconciliation snapshot, when a canonical value arrived during local prediction.
+ public CultMeshReactiveDocumentReconciliation? Reconciliation { get; private set; }
+
+ internal void Start()
+ {
+ _subscription = _document.Watch(ApplyCanonicalSnapshot);
+ }
+
+ /// Marks the current value dirty and schedules a coalesced prediction or replacement.
+ public void MarkDirty()
+ {
+ ThrowIfDisposed();
+ lock (_gate)
+ {
+ _dirty = true;
+ ScheduleFlushLocked();
+ }
+ }
+
+ /// Mutates the current value and schedules a coalesced prediction or replacement.
+ public TDocument Update(Action update)
+ {
+ if (update == null) throw new ArgumentNullException(nameof(update));
+ ThrowIfDisposed();
+ lock (_gate)
+ {
+ update(Current);
+ _dirty = true;
+ ScheduleFlushLocked();
+ return Current;
+ }
+ }
+
+ /// Replaces the current local value and schedules a coalesced prediction or replacement.
+ public TDocument SetCurrent(TDocument value)
+ {
+ if (value == null) throw new ArgumentNullException(nameof(value));
+ ThrowIfDisposed();
+ lock (_gate)
+ {
+ Current = CultMeshDocumentHandle.CloneDocument(value);
+ _dirty = true;
+ ScheduleFlushLocked();
+ return Current;
+ }
+ }
+
+ /// Reads a fresh canonical snapshot and adopts it as the current value.
+ public async Task RefreshAsync()
+ {
+ ThrowIfDisposed();
+ var latest = CultMeshDocumentHandle.CloneDocument(
+ await _document.LatestAsync().ConfigureAwait(false));
+ lock (_gate)
+ {
+ Current = latest;
+ _dirty = false;
+ Reconciliation = null;
+ return Current;
+ }
+ }
+
+ /// Immediately sends the latest local dirty value, if any, through the document authority shape.
+ public async Task FlushAsync()
+ {
+ TDocument predicted;
+ lock (_gate)
+ {
+ ThrowIfDisposed();
+ if (!_dirty)
+ return;
+ if (_flushing)
+ {
+ _flushQueued = true;
+ return;
+ }
+
+ _timer?.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
+ _flushing = true;
+ _dirty = false;
+ predicted = CultMeshDocumentHandle.CloneDocument(Current);
+ }
+
+ try
+ {
+ await _document.SetAsync(predicted).ConfigureAwait(false);
+ }
+ finally
+ {
+ var shouldFlushAgain = false;
+ lock (_gate)
+ {
+ _flushing = false;
+ shouldFlushAgain = _flushQueued || _dirty;
+ _flushQueued = false;
+ }
+
+ if (shouldFlushAgain)
+ await FlushAsync().ConfigureAwait(false);
+ }
+ }
+
+ /// Clears the most recent reconciliation metadata.
+ public void ClearReconciliation()
+ {
+ ThrowIfDisposed();
+ Reconciliation = null;
+ }
+
+ ///
+ public void Dispose()
+ {
+ lock (_gate)
+ {
+ if (_disposed)
+ return;
+ _disposed = true;
+ }
+
+ _subscription?.Dispose();
+ _timer?.Dispose();
+ }
+
+ private void ApplyCanonicalSnapshot(TDocument canonical)
+ {
+ if (canonical == null)
+ return;
+
+ lock (_gate)
+ {
+ if (_disposed)
+ return;
+
+ var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical);
+ if (_dirty || _flushing)
+ {
+ Reconciliation = new CultMeshReactiveDocumentReconciliation(
+ nextCanonical,
+ CultMeshDocumentHandle.CloneDocument(Current),
+ DateTimeOffset.UtcNow);
+ if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot)
+ return;
+ }
+
+ Current = nextCanonical;
+ Reconciliation = null;
+ }
+ }
+
+ private void ScheduleFlushLocked()
+ {
+ if (_options.FlushDelay <= TimeSpan.Zero)
+ {
+ _ = Task.Run(FlushAsync);
+ return;
+ }
+
+ _timer ??= new Timer(_ => _ = Task.Run(FlushAsync));
+ _timer.Change(_options.FlushDelay, Timeout.InfiniteTimeSpan);
+ }
+
+ private void ThrowIfDisposed()
+ {
+ if (_disposed)
+ throw new ObjectDisposedException(nameof(CultMeshReactiveDocument));
+ }
+ }
+
///
/// Schema-aware lookup table for typed CultMesh document handles.
///
@@ -1285,6 +1565,22 @@ namespace GameCult.Mesh
{
return Document().Watch(onNext);
}
+
+ /// Creates a managed reactive document mirror by CLR type or same-schema alias.
+ public Task> ReactiveAsync(
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ return Document().ReactiveAsync(options);
+ }
+
+ /// Creates a managed reactive document mirror synchronously by CLR type or same-schema alias.
+ public CultMeshReactiveDocument Reactive(
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ return Document().Reactive(options);
+ }
}
///
diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
index d228788a..48812792 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
@@ -892,6 +892,112 @@ public sealed class CultMeshStreamingTests
cache.Get(key)!.Revision.Should().Be(6);
}
+ [Test]
+ public async Task ReactiveDocument_CoalescesLocalEditsIntoOnePrediction()
+ {
+ var subject = new Subject();
+ var current = new MeshNoteDocument
+ {
+ Schema = "tests.mesh_note.v1",
+ Text = "initial",
+ Revision = 1
+ };
+ var predictions = new List();
+ var handle = CultMesh.Document(
+ "mesh.note.reactive",
+ CultMesh.Verse("starbridge", "pilot-a"),
+ _ => Task.FromResult(current),
+ _ => subject,
+ value =>
+ {
+ current = value;
+ subject.OnNext(value);
+ return Task.CompletedTask;
+ },
+ value =>
+ {
+ predictions.Add(value);
+ current = value;
+ subject.OnNext(value);
+ return Task.CompletedTask;
+ });
+ var catalog = CultMesh.Documents(handle);
+
+ using var reactive = await catalog.ReactiveAsync(
+ new CultMeshReactiveDocumentOptions
+ {
+ FlushDelay = TimeSpan.FromMinutes(1)
+ });
+
+ reactive.Current.Text.Should().Be("initial");
+ reactive.Update(document =>
+ {
+ document.Text = "first-local-edit";
+ document.Revision++;
+ });
+ reactive.Update(document =>
+ {
+ document.Text = "second-local-edit";
+ document.Revision++;
+ });
+
+ predictions.Should().BeEmpty();
+
+ await reactive.FlushAsync();
+
+ predictions.Should().ContainSingle();
+ predictions[0].Text.Should().Be("second-local-edit");
+ predictions[0].Revision.Should().Be(3);
+ current.Text.Should().Be("second-local-edit");
+ }
+
+ [Test]
+ public async Task ReactiveDocument_TracksCanonicalSnapshotDuringDirtyPrediction()
+ {
+ var subject = new Subject();
+ var current = new MeshNoteDocument
+ {
+ Schema = "tests.mesh_note.v1",
+ Text = "initial",
+ Revision = 1
+ };
+ var handle = CultMesh.Document(
+ "mesh.note.reconciliation",
+ CultMesh.Verse("starbridge", "pilot-a"),
+ _ => Task.FromResult(current),
+ _ => subject,
+ value =>
+ {
+ current = value;
+ subject.OnNext(value);
+ return Task.CompletedTask;
+ });
+
+ using var reactive = await handle.ReactiveAsync(
+ new CultMeshReactiveDocumentOptions
+ {
+ FlushDelay = TimeSpan.FromMinutes(1)
+ });
+
+ reactive.Update(document =>
+ {
+ document.Text = "local-prediction";
+ document.Revision = 2;
+ });
+
+ subject.OnNext(new MeshNoteDocument
+ {
+ Schema = "tests.mesh_note.v1",
+ Text = "canonical-correction",
+ Revision = 7
+ });
+
+ reactive.Current.Text.Should().Be("local-prediction");
+ reactive.Reconciliation.Should().NotBeNull();
+ reactive.Reconciliation!.Canonical.Text.Should().Be("canonical-correction");
+ reactive.Reconciliation.Predicted.Text.Should().Be("local-prediction");
+ }
+
[Test]
public async Task DocumentHandle_ReadsSchemaPublicationsFromSingleFileStores()
{