executeCoordinated

Batched-execution path for Draft pipelines (G5).

Walks the coordinated nodes of a Draft in topological order and issues one call to propose per node. A DraftStage.BatchedEmbroider — the product of G3 consolidation carries multiple agreements in a single proposal, so it costs one call, not one per op.

Round-count reduction at execution

For an unplanned draft with K independent DraftStage.Embroider nodes, executeCoordinated issues K proposals. For the same draft after Draft.plan — which fuses independent embroiders at the same dependency level into DraftStage.BatchedEmbroider nodes — it issues only depth proposals, where depth is the number of dependency levels in the coordination DAG.

This matches the analytical CoordinationCost.rounds model:

roundsAtExecution == coordinationCost(plan(draft), stats).rounds

Payload encoding

Each proposal carries the embroider opId(s) of its node as a comma-separated UTF-8 string — the symbolic agreements that ride that Raft round-trip. A DraftStage.BatchedEmbroider encodes all its opIds in one payload; a lone DraftStage.Embroider encodes one opId. Callers interpret the bytes; this function only produces them.

Ordering and dependencies

Nodes are visited in topological order (the order they appear in Draft.nodes). A sequential dependency (level N depends on level N−1) is automatically enforced because propose suspends until the proposal commits before the next proposal is issued.

Fully-monotone drafts

When Draft.isMonotone is true, no coordinated node exists and executeCoordinated issues zero proposals and returns 0 immediately.

Parameters

See also

Samples

val branchA = Warp.shuttle(OpId("source.a")).embroider(OpId("embroider.a"))
val branchB = Warp.shuttle(OpId("source.b")).embroider(OpId("embroider.b"))
val unplanned: Draft<Unit> = branchA.combine(branchB)
val planned: Draft<Unit> = unplanned.plan(WarpStats.empty())

val unplannedProposals = mutableListOf<String>()
unplanned.executeCoordinated { payload ->
    unplannedProposals.add(payload.decodeToString())
}

val plannedProposals = mutableListOf<String>()
planned.executeCoordinated { payload ->
    plannedProposals.add(payload.decodeToString())
}

// Unplanned: 2 separate Embroider nodes → 2 proposals (one Raft round each).
check(unplannedProposals.size == 2)
// Planned: BatchedEmbroider(A,B) → 1 proposal (both agreements in one Raft round-trip).
check(plannedProposals.size == 1)
check(plannedProposals.single().contains("embroider.a"))
check(plannedProposals.single().contains("embroider.b"))