ConvergentExecution
A minimal local runtime that ties a Draft pipeline to a converging IncrementalResult.
Because every monotone stage in a Draft commutes, the query never "finishes" — it converges, refining as contributions arrive from distributed peers. This is the execution side of that: callers submit lattice deltas via submit, which are queued to a channel and processed asynchronously on the provided scope, then accumulated into result via the lattice join.
Scope discipline
The scope parameter is required with no default. The runtime's lifecycle is explicitly tied to whatever scope the caller provides — a service scope in production, backgroundScope in tests (sharing the test scheduler). Defaulting to GlobalScope or a real Dispatchers.* would silently decouple contributions from the test's virtual clock, breaking determinism.
Thread safety
submit is non-blocking: it sends to an UNLIMITED channel using trySend (which never fails for unlimited channels). A single coroutine launched on scope drains the channel, maintaining FIFO ordering for lattice joins. IncrementalResult.contribute is CAS-safe, so concurrent submits from multiple threads converge correctly.
Under a kotlinx.coroutines.test.StandardTestDispatcher, call runCurrent() after submit to drain the channel and observe the updated result.
Draft connection
The draft is stored and exposed for inspection (E-2/E-3 rewrite and cost-model integration). This runtime operates on the monotone path only — DraftStage.Embroider stages (the coordinated path) are not executed here; WarpNode handles them.
Parameters
the dataflow graph whose monotone stages guide convergent execution.
the coroutine scope on which contributions are processed. Required — no default.
the lattice bottom — the result before any contributions arrive.
Type Parameters
the lattice type — must be Quilted.
Samples
val alice = ReplicaId("alice")
val bob = ReplicaId("bob")
val draft = Warp.shuttle(OpId("source"))
.map(OpId("map.score"))
.filter(OpId("filter.threshold"))
// scope is a service scope in production; backgroundScope in tests.
val exec = ConvergentExecution(draft = draft, scope = scope, initial = GCounter.ZERO)
// submit is non-blocking — queued to an UNLIMITED channel, processed on scope.
exec.submit(GCounter.of(alice to 5L))
exec.submit(GCounter.of(bob to 3L))
// After scope runs: exec.result.state.value.value == 8L
// The draft is exposed for inspection and cost-model integration.
check(exec.draft.isMonotone)
check(exec.draft.stages.size == 3)