module Runtime.NativeDataParallelOwned

import Learning.DataParallel
import Learning.Numerics
import Learning.Transition
import Std.Numeric.Profile
import Component.Pointwise
import Runtime.OrderedCombine
import Compiler.Planning.OrderedCombine
import Runtime.Protocol
import Compiler.Planning.Protocol
import Runtime.NativeProtocol
import Runtime.NativeCombine
import Runtime.NativeDataParallel
import Runtime.NativeBranch
import Runtime.RegionTransfer
import Runtime.RegionFrame
import Compiler.Planning.RegionFrame
import Runtime.NativeRegionFrame
import Runtime.NativePeerEndpoint
import Runtime.NativeFloatPlan
import Runtime.NativeMemoryPlan
import Runtime.NativeLaunchRecipeLoad
import Runtime.NativePhysicalProgram
import Runtime.NativePhysicalProgramComposer
import Runtime.NativePointwisePlan
import Runtime.NativeHostTarget
import Runtime.NativeResourcePlan
import Runtime.NativeSocketPlan
import Runtime.NativeIndependentState
import Runtime.NativeWordArithmetic
import Runtime.ArenaCertificate
import Checkpoint.Envelope
import Checkpoint.NativeEnvelope
import Checkpoint.NativeRegions
import Checkpoint.NativePeerPublication
import Checkpoint.RegionManifest
import Representation.Partition
import Data.Bytes
import Data.SHA256Digest
import Model.Word64
import Std.Foundation
import Std.List
import Std.Natural
import Std.Word

-- Per-coordinate ownership for a data-parallel learner whose update is
-- separable (Learning.DataParallel.dataParallelSeparable) with q words of
-- optimizer state per coordinate: coordinate i is owned by owner i mod n,
-- and only its owner holds that coordinate's optimizer state and updates it.
-- The shared state words stay replicated. `p` chooses the parameters'
-- residency: replicated on every owner (p = 0, optimizer-state
-- partitioning) or partitioned with the optimizer state (p = 1, parameter
-- partitioning: a coordinate's parameter is one more word its owner holds).
--
-- One update with replicated parameters, on a star of owners 0..n-1:
--   every owner: its samples' (numerator, mass) leaf;
--   sum: the declared left-deep binary64 sum at owner 0, replicated
--     (Runtime.NativeCombine);
--   every owner: means, then only the shared assignments and its own
--     coordinates' assignments of the update program
--     (dataParallelOwnedProgram), in program order;
--   assemble: every owner offers its parameters with an ownership mask; a
--     second ordered combine assembles them by selection, never by
--     arithmetic, refusing two owners for one coordinate, and replicates the
--     result; every owner checks that every coordinate was owned and installs
--     the assembled parameters.
-- With partitioned parameters the assembly gathers them before they are
-- read, and the gathered copy is an explicit, short-lived read replica:
--   assemble: every owner offers its own parameters (the others +0) with its
--     ownership mask; the assembly gathers and replicates them;
--   materialize: every coordinate was owned; the result becomes this owner's
--     read replica, tagged with the version it was gathered at
--     (MaterializeReadVersion);
--   every owner: its leaf, reading only a held replica of the current version;
--   release: the replica is cleared and no longer held (ReleaseReadReplica);
--   sum, then means and the owned assignments on the owner's own parameter
--     and state words; a replica still held when the version advances
--     refuses (`data-parallel-replica-retained`).
-- The replica is held across all of the owner's samples (one gathered version
-- per update) and is part of the owner's peak storage, not its persistent
-- state. Either way the results equal the replicated realization's bit for
-- bit: separability makes each owned coordinate's results independent of the
-- others.
--
-- Persistence: owner 0 publishes the replicated words (parameters when
-- replicated, shared state); every owned word is a shard of its owner (the
-- star publication prepares each owner's shards from that owner's capture).
-- Restore reads the replicated words and only the restoring owner's own
-- words. Publication requires every owner's capture prefix and replicated
-- words to agree. Partitioned checkpoints bind the learner's
-- placement-independent identity, not its split of samples or its owner
-- count, so a saved state resumes on another number of owners.
family DataParallelOwnedOperation : Type 0
constructor DataParallelOwnedContribute
constructor DataParallelOwnedSum
field unrestricted dataParallelOwnedSumOperation : (family OrderedCombineOperation (family RegionTransferIdentity))
constructor DataParallelOwnedUpdate
constructor DataParallelOwnedOffer
constructor DataParallelOwnedAssemble
field unrestricted dataParallelOwnedAssembleOperation : (family OrderedCombineOperation (family RegionTransferIdentity))
constructor DataParallelOwnedInstall
constructor DataParallelOwnedMaterialize
constructor DataParallelOwnedRelease
-- gradient partitioning: every owner's leaf projected onto each coordinate
-- owner's coordinates, then one reduction to that owner per coordinate owner
constructor DataParallelOwnedProject
constructor DataParallelOwnedGradient
field unrestricted dataParallelOwnedGradientOwner : Nat
field unrestricted dataParallelOwnedGradientOperation : (family OrderedCombineOperation (family RegionTransferIdentity))

end-family

-- What every topology shares: both combines' placements, where the owner
-- state begins and the membership epoch a fresh owner starts in.
family NativeDataParallelOwnedLayout : Type 0
constructor NativeDataParallelOwnedLayoutOf
field unrestricted dataParallelOwnedSumPlacement : (family NativeCombinePlacement)
field unrestricted dataParallelOwnedAssemblePlacement : (family NativeCombinePlacement)
field unrestricted dataParallelOwnedStateAt : Nat
field unrestricted dataParallelOwnedEpoch : Nat
-- the learner's identity digest, computed once per artifact
field unrestricted dataParallelOwnedDigest : Bytes
-- with gradient partitioning, one reduction per coordinate owner (owner r's
-- at position r); none otherwise
field unrestricted dataParallelOwnedGradients : (family StdList (family NativeCombinePlacement))

end-family

-- The star's publication: capture and release frames and the region
-- checkpoint.
family NativeDataParallelOwnedStarLayout : Type 0
constructor NativeDataParallelOwnedStarLayoutOf
field unrestricted dataParallelOwnedCapture : (family RegionFrameLayout)
field unrestricted dataParallelOwnedCommitted : (family RegionFrameLayout)
field unrestricted dataParallelOwnedCheckpoint : (family NativeRegionCheckpoint)

end-family

def dpoSum =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) .
        (family NativeCombinePlacement))
      layout
      (branch NativeDataParallelOwnedLayoutOf sum assemble stateAt epoch digest gradients . sum)))

def dpoAssemblePlacement =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) .
        (family NativeCombinePlacement))
      layout
      (branch
        NativeDataParallelOwnedLayoutOf
        sum
        assemble
        stateAt
        epoch
        digest
        gradients
        .
        assemble)))

def dpoStateAt =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) . Nat)
      layout
      (branch NativeDataParallelOwnedLayoutOf sum assemble stateAt epoch digest gradients . stateAt)))

def dpoEpoch =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) . Nat)
      layout
      (branch NativeDataParallelOwnedLayoutOf sum assemble stateAt epoch digest gradients . epoch)))

def dpoDigest =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) . Bytes)
      layout
      (branch NativeDataParallelOwnedLayoutOf sum assemble stateAt epoch digest gradients . digest)))

def dpoGradients =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (eliminate
      NativeDataParallelOwnedLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedLayout) .
        (family StdList (family NativeCombinePlacement)))
      layout
      (branch
        NativeDataParallelOwnedLayoutOf
        sum
        assemble
        stateAt
        epoch
        digest
        gradients
        .
        gradients)))

-- Owner r's reduction (the sum's placement when there is none).
def dpoGradientAt =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (lambda unrestricted owner : Nat .
      (stdListHeadOr
        (family NativeCombinePlacement)
        (dpoSum layout)
        (stdListDrop (family NativeCombinePlacement) owner (dpoGradients layout)))))

def dpoCapture =
  (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
    (eliminate
      NativeDataParallelOwnedStarLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedStarLayout) .
        (family RegionFrameLayout))
      star
      (branch NativeDataParallelOwnedStarLayoutOf capture committed checkpoint . capture)))

def dpoCommitted =
  (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
    (eliminate
      NativeDataParallelOwnedStarLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedStarLayout) .
        (family RegionFrameLayout))
      star
      (branch NativeDataParallelOwnedStarLayoutOf capture committed checkpoint . committed)))

def dpoCheckpoint =
  (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
    (eliminate
      NativeDataParallelOwnedStarLayout
      (lambda unrestricted current : (family NativeDataParallelOwnedStarLayout) .
        (family NativeRegionCheckpoint))
      star
      (branch NativeDataParallelOwnedStarLayoutOf capture committed checkpoint . checkpoint)))

-- The shared realization's view: the sum placement and the owner state.
def dpoGeneric =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (constructor
      NativeDataParallelLayout
      NativeDataParallelLayoutOf
      (dpoSum layout)
      (dpoStateAt layout)
      (dpoDigest layout)))

-- ---- ownership and shapes ----
-- Coordinate i belongs to owner i mod n.
def dpoOwns =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted role : Nat .
      (lambda unrestricted coordinate : Nat .
        (naturalEqual (naturalModuloUnchecked coordinate owners) role))))

-- Coordinate words per owner: the most any owner holds.
def dpoPerOwner =
  (lambda unrestricted k : Nat .
    (lambda unrestricted owners : Nat .
      (naturalDivideUnchecked (naturalSaturatingSubtract (naturalAdd k owners) 1) owners)))

def dpoShared =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (naturalSaturatingSubtract (dpM c u system) (naturalMultiply q (dpK c u system)))))))

-- Owner state: capture prefix, parameters (k), shared state (h), then this
-- owner's coordinate words, word j of its idx-th coordinate (coordinate
-- role + idx n) at j o + idx. Every owner's state has the same extent.
def dpoParametersAt =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (naturalAdd (dpoStateAt layout) nativeRegionCaptureBytes))

-- Replicated parameters precede the shared state; partitioned ones are owned
-- words.
def dpoReplicatedParameterBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (naturalMultiply (naturalIsZero p) (dpParameterBytes c u system))))))

def dpoSharedAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
            (naturalAdd (dpoParametersAt layout) (dpoReplicatedParameterBytes c u system p)))))))

def dpoOwnedAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (naturalAdd
                (dpoSharedAt c u system p layout)
                (naturalMultiply dpWord (dpoShared c u system q)))))))))

-- Offset, past the parameters and shared state, of word j of coordinate i
-- in its owner's state.
def dpoWordOffset =
  (lambda unrestricted k : Nat .
    (lambda unrestricted owners : Nat .
      (lambda unrestricted word : Nat .
        (lambda unrestricted coordinate : Nat .
          (naturalMultiply
            dpWord
            (naturalAdd
              (naturalMultiply word (dpoPerOwner k owners))
              (naturalDivideUnchecked coordinate owners)))))))

def dpoAgreedBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (naturalAdd
              nativeRegionCaptureBytes
              (naturalAdd
                (dpoReplicatedParameterBytes c u system p)
                (naturalMultiply dpWord (dpoShared c u system q)))))))))

def dpoStateBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (naturalAdd
              (dpoAgreedBytes c u system q p)
              (naturalMultiply
                dpWord
                (naturalMultiply
                  (naturalAdd q p)
                  (dpoPerOwner (dpK c u system) (dpOwners c u system))))))))))

-- The update-program slot of owned word j of a coordinate: its parameter
-- (when partitioned, word 0) or optimizer-state word.
def dpoSlotOf =
  (lambda unrestricted k : Nat .
    (lambda unrestricted p : Nat .
      (lambda unrestricted word : Nat .
        (lambda unrestricted coordinate : Nat .
          (naturalAdd (naturalMultiply (naturalSaturatingSubtract (succ word) p) k) coordinate)))))

-- The read replica of partitioned parameters: held flag, the version it was
-- gathered at, then the k parameters. None when parameters are replicated.
def dpoReplicaBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (naturalMultiply p (naturalMultiply dpWord (naturalAdd 2 (dpK c u system))))))))

-- Commands only when `flag` is nonzero, or only when it is zero.
def dpoWhen =
  (lambda unrestricted flag : Nat .
    (lambda unrestricted commands : (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)) .
      (lambda unrestricted tail : (family NativePhysicalCommands) .
        (nat-eliminate
          (lambda unrestricted current : Nat . (family NativePhysicalCommands))
          tail
          (lambda unrestricted previous : Nat .
            (lambda unrestricted ignored : (family NativePhysicalCommands) . (commands tail)))
          flag))))

def dpoUnless =
  (lambda unrestricted flag : Nat . (dpoWhen (naturalIsZero flag)))

-- `step coordinate word` for every coordinate this role owns and every word.
def dpoEachOwned =
  (lambda unrestricted k : Nat .
    (lambda unrestricted owners : Nat .
      (lambda unrestricted role : Nat .
        (lambda unrestricted q : Nat .
          (lambda unrestricted step : (pi unrestricted coordinate : Nat . (pi unrestricted word : Nat . (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)))) .
            (lambda unrestricted tail : (family NativePhysicalCommands) .
              (dpEach
                k
                (lambda unrestricted coordinate : Nat .
                  (lambda unrestricted rest : (family NativePhysicalCommands) .
                    (nat-eliminate
                      (lambda unrestricted owned : Nat . (family NativePhysicalCommands))
                      rest
                      (lambda unrestricted previous : Nat .
                        (lambda unrestricted ignored : (family NativePhysicalCommands) .
                          (dpEach q (step coordinate) rest)))
                      (dpoOwns owners role coordinate))))
                tail)))))))

-- ---- the assembly combine ----
-- An offered value is the k parameters followed by k ownership words (1 for
-- an owned coordinate). Combining selects, coordinate by coordinate, the
-- right operand's parameter where it owns the coordinate and the left
-- operand's otherwise; two owners for one coordinate refuse.
def dpoOfferBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalMultiply dpWord (naturalMultiply 2 (dpK c u system))))))

-- The read replica lies after this owner's offer.
def dpoReplicaAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (naturalAdd
            (nativeCombineRegionEnd (dpoAssemblePlacement layout))
            (dpoOfferBytes c u system))))))

def dpoReplicaParametersAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (nativePointwiseAt (dpoReplicaAt c u system layout) 2)))))

def dpoSelect =
  (lambda unrestricted k : Nat .
    (lambda unrestricted out : Nat .
      (lambda unrestricted left : Nat .
        (lambda unrestricted right : Nat .
          (dpEach
            k
            (lambda unrestricted index : Nat .
              (lambda unrestricted tail : (family NativePhysicalCommands) .
                (let unrestricted rightMask =
                  (nativePointwiseAt right (naturalAdd k index))
                  in
                  (let unrestricted leftMask =
                    (nativePointwiseAt left (naturalAdd k index))
                    in
                    (dpCopy
                      (nativePointwiseAt out index)
                      (nativePointwiseAt left index)
                      dpWord
                      (dpCopy
                        (nativePointwiseAt out (naturalAdd k index))
                        leftMask
                        dpWord
                        (nativeWhenEqual
                          (nativeMemoryLoad rightMask)
                          (nativeLaunchRecipeImmediate 1)
                          b"data-parallel-assembly-select"
                          (nativeLaunchRecipeAssertEqual
                            (nativeMemoryLoad leftMask)
                            (nativeLaunchRecipeImmediate 0)
                            b"data-parallel-assembly-overlap"
                            (dpCopy
                              (nativePointwiseAt out index)
                              (nativePointwiseAt right index)
                              dpWord
                              (nativeMemoryStoreWord64
                                b"data-parallel-assembly-owned"
                                (nativePointwiseAt out (naturalAdd k index))
                                (nativeLaunchRecipeImmediate 1)
                                dpEnd)))
                          tail))))))))))))

-- The composition of every assembly transfer identity.
def dpoAssembleComposition : Bytes =
  b"data-parallel-owned"

def dpoAssembleBaseAnchor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (constructor
            RegionTransferIdentity
            RegionTransferIdentityOf
            dpoAssembleComposition
            digest
            b"parameters-and-ownership"
            b"binary64-words-selected-by-owner"
            (constructor LearningRound LearningRoundOf 1 0)
            0
            0
            0
            (constructor LogicalInterval LogicalIntervalOf 0 (dpOwners c u system)))))))

-- Assembly performs no arithmetic: its contract lists no operation and names
-- the selection order.
def dpoAssembleContract =
  (lambda unrestricted owners : Nat .
    (constructor
      NumericalContract
      NumericalContractOf
      nativeFiniteFloat64Profile
      (constructor StdList StdListEmpty (family StdNumericOperation))
      (constructor
        NumericalRelation
        NumericalExactOrder
        (bytes-append b"data-parallel-assemble-owned-coordinates;owners=" (regionFrameWord owners)))))

def dpoAssembleAnchor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (combineCoverageAnchor
            (dpMembers (dpOwners c u system))
            (combineNumericalAnchor
              (dpoAssembleContract (dpOwners c u system))
              (dpoAssembleBaseAnchor c u system digest)))))))

-- The assembly combine of this role under `topology` (the same tree as the
-- sum's), from where the sum's storage ends.
def dpoAssemblePlacementFor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted topology : (family DataParallelTopology) .
            (lambda unrestricted role : Nat .
              (lambda unrestricted sum : (family NativeCombinePlacement) .
                (constructor
                  NativeCombinePlacement
                  NativeCombinePlacementOf
                  (nativeCombineContextFor
                    (dpoAssembleAnchor c u system digest)
                    role
                    (dpTree topology (dpOwners c u system)))
                  (dpPublicationBase c u system sum)
                  (dpoOfferBytes c u system)))))))))

-- ---- gradient partitioning ----
-- With gradient partitioning (ReduceToRegionOwners) each coordinate owner r
-- receives only the sums of its own coordinates and the mass: every owner's
-- leaf is projected onto r's coordinates (padded to the most any owner
-- holds) and reduced over the same tree in the same order as the full sum,
-- then delivered along the tree path to r alone (CombineDeliverTo). Each
-- coordinate's sum is the replicated sum's word bit for bit.
def dpoGradientOwners =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalMinimum (dpOwners c u system) (dpK c u system)))))

def dpoGradientWords =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (succ (dpoPerOwner (dpK c u system) (dpOwners c u system))))))

def dpoGradientBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalMultiply dpWord (dpoGradientWords c u system)))))

def dpoGradientComposition =
  (lambda unrestricted owner : Nat .
    (bytes-append b"data-parallel-gradient-owner;" (regionFrameWord owner)))

def dpoGradientBaseAnchor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted owner : Nat .
            (constructor
              RegionTransferIdentity
              RegionTransferIdentityOf
              (dpoGradientComposition owner)
              digest
              b"owned-numerator-mass"
              b"binary64-vector-and-mass"
              (constructor LearningRound LearningRoundOf 1 0)
              0
              0
              0
              (constructor LogicalInterval LogicalIntervalOf 0 (dpOwners c u system))))))))

def dpoGradientAnchor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted topology : (family DataParallelTopology) .
            (lambda unrestricted owner : Nat .
              (combineCoverageAnchor
                (dpMembers (dpOwners c u system))
                (combineNumericalAnchor
                  (dpContract topology (dpOwners c u system))
                  (dpoGradientBaseAnchor c u system digest owner)))))))))

-- Owner r's reduction is delivered to owner r.
def dpoGradientDelivery =
  (lambda unrestricted target : Nat . (constructor CombineDelivery CombineDeliverTo target))

-- One reduction per coordinate owner, from `base`: each placement's frames,
-- then its leaf value, then the next placement.
def dpoGradientPlacementsFor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted topology : (family DataParallelTopology) .
            (lambda unrestricted role : Nat .
              (lambda unrestricted base : Nat .
                (app
                  (app
                    (nat-eliminate
                      (lambda unrestricted current : Nat .
                        (pi unrestricted owner : Nat .
                          (pi unrestricted from : Nat .
                            (family StdList (family NativeCombinePlacement)))))
                      (lambda unrestricted owner : Nat .
                        (lambda unrestricted from : Nat .
                          (constructor StdList StdListEmpty (family NativeCombinePlacement))))
                      (lambda unrestricted previous : Nat .
                        (lambda unrestricted rest : (pi unrestricted owner : Nat . (pi unrestricted from : Nat . (family StdList (family NativeCombinePlacement)))) .
                          (lambda unrestricted owner : Nat .
                            (lambda unrestricted from : Nat .
                              (let unrestricted placement =
                                (constructor
                                  NativeCombinePlacement
                                  NativeCombinePlacementOf
                                  (nativeCombineDeliveringContextFor
                                    (dpoGradientAnchor c u system digest topology owner)
                                    (dpoGradientDelivery owner)
                                    role
                                    (dpTree topology (dpOwners c u system)))
                                  from
                                  (dpoGradientBytes c u system))
                                in
                                (constructor
                                  StdList
                                  StdListCons
                                  (family NativeCombinePlacement)
                                  placement
                                  (rest
                                    (succ owner)
                                    (frameAlign
                                      (naturalAdd
                                        (nativeCombineRegionEnd placement)
                                        (dpoGradientBytes c u system))))))))))
                      (dpoGradientOwners c u system))
                    0)
                  base))))))))

-- Where storage after the reductions may begin (`base` when there are none).
def dpoGradientsEnd =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted gradients : (family StdList (family NativeCombinePlacement)) .
          (lambda unrestricted base : Nat .
            (stdListFold
              (family NativeCombinePlacement)
              Nat
              (lambda unrestricted placement : (family NativeCombinePlacement) .
                (lambda unrestricted rest : Nat .
                  (let unrestricted ending =
                    (frameAlign
                      (naturalAdd (nativeCombineRegionEnd placement) (dpoGradientBytes c u system)))
                    in
                    (naturalSelect (naturalLess rest ending) ending rest))))
              base
              gradients))))))

def dpoCompositionOf =
  (lambda unrestricted identity : (family RegionTransferIdentity) .
    (eliminate
      RegionTransferIdentity
      (lambda unrestricted current : (family RegionTransferIdentity) . Bytes)
      identity
      (branch
        RegionTransferIdentityOf
        composition
        run
        region
        representation
        round
        operation
        sender
        receiver
        interval
        .
        composition)))

-- 1 + r for a transfer of owner r's reduction, 0 for any other.
def dpoGradientIndex =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (lambda unrestricted identity : (family RegionTransferIdentity) .
      (let unrestricted composition =
        (dpoCompositionOf identity)
        in
        (stdListFold
          Nat
          Nat
          (lambda unrestricted owner : Nat .
            (lambda unrestricted rest : Nat .
              (naturalSelect
                (bytes-equal composition (dpoGradientComposition owner))
                (succ owner)
                rest)))
          0
          (stdListNaturalsBelow
            (stdListLength (family NativeCombinePlacement) (dpoGradients layout)))))))

-- Elementwise binary64 addition of `count` words.
def dpoApplyWords =
  (lambda unrestricted count : Nat .
    (lambda unrestricted out : Nat .
      (lambda unrestricted left : Nat .
        (lambda unrestricted right : Nat .
          (dpEach
            count
            (lambda unrestricted index : Nat .
              (dpAdd
                (nativePointwiseAt out index)
                (nativePointwiseAt left index)
                (nativePointwiseAt right index))))))))

-- Every reduction's leaf: this owner's numerators of that owner's
-- coordinates (+0 past k), then its mass.
def dpoProject =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (let unrestricted k =
              (dpK c u system)
              in
              (let unrestricted owners =
                (dpOwners c u system)
                in
                (let unrestricted leaf =
                  (dpLeafAt (dpoGeneric layout))
                  in
                  (let unrestricted per =
                    (dpoPerOwner k owners)
                    in
                    (stdListFold
                      Nat
                      (family NativePhysicalCommands)
                      (lambda unrestricted owner : Nat .
                        (lambda unrestricted rest : (family NativePhysicalCommands) .
                          (let unrestricted target =
                            (nativeCombineRegionEnd (dpoGradientAt layout owner))
                            in
                            (dpEach
                              per
                              (lambda unrestricted index : Nat .
                                (lambda unrestricted more : (family NativePhysicalCommands) .
                                  (let unrestricted coordinate =
                                    (naturalAdd owner (naturalMultiply index owners))
                                    in
                                    (nat-eliminate
                                      (lambda unrestricted inside : Nat .
                                        (family NativePhysicalCommands))
                                      (nativeMemoryStoreWord64
                                        b"data-parallel-gradient-padding"
                                        (nativePointwiseAt target index)
                                        (nativeLaunchRecipeImmediate 0)
                                        more)
                                      (lambda unrestricted previous : Nat .
                                        (lambda unrestricted ignored : (family NativePhysicalCommands) .
                                        (dpCopy
                                        (nativePointwiseAt target index)
                                        (nativePointwiseAt leaf coordinate)
                                        dpWord
                                        more)))
                                      (naturalLess coordinate k)))))
                              (dpCopy
                                (nativePointwiseAt target per)
                                (nativePointwiseAt leaf k)
                                dpWord
                                rest)))))
                      tail
                      (stdListNaturalsBelow
                        (stdListLength (family NativeCombinePlacement) (dpoGradients layout)))))))))))))

-- ---- frames of both combines ----
-- Every identity either combine derives keeps its base anchor's
-- composition, so the composition names the combine for any transfer --
-- including transfers between other owners, which admission also checks.
def dpoIsAssembly =
  (lambda unrestricted identity : (family RegionTransferIdentity) .
    (eliminate
      RegionTransferIdentity
      (lambda unrestricted current : (family RegionTransferIdentity) . Nat)
      identity
      (branch
        RegionTransferIdentityOf
        composition
        run
        region
        representation
        round
        operation
        sender
        receiver
        interval
        .
        (bytes-equal composition dpoAssembleComposition))))

def dpoPlacementOf =
  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
    (lambda unrestricted identity : (family RegionTransferIdentity) .
      (nat-eliminate
        (lambda unrestricted assembly : Nat . (family NativeCombinePlacement))
        (nat-eliminate
          (lambda unrestricted index : Nat . (family NativeCombinePlacement))
          (dpoSum layout)
          (lambda unrestricted owner : Nat .
            (lambda unrestricted ignored : (family NativeCombinePlacement) .
              (dpoGradientAt layout owner)))
          (dpoGradientIndex layout identity))
        (lambda unrestricted previous : Nat .
          (lambda unrestricted ignored : (family NativeCombinePlacement) .
            (dpoAssemblePlacement layout)))
        (dpoIsAssembly identity))))

-- Partitioned parameters: no read replica may outlive the version it was
-- gathered at; then the version advances.
def dpoRetire =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (nativeLaunchRecipeAssertEqual
              (nativeMemoryLoad (dpoReplicaAt c u system layout))
              (nativeLaunchRecipeImmediate 0)
              b"data-parallel-replica-retained"
              (niAdvanceAt (dpoStateAt layout) tail)))))))

-- The means the update reads: every coordinate's from the replicated sum,
-- or, with gradient partitioning, only this owner's coordinates' from its
-- reduction (word idx of owner r's value is coordinate r + idx n).
def dpoMeans =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted role : Nat .
            (lambda unrestricted at : Nat .
              (lambda unrestricted tail : (family NativePhysicalCommands) .
                (let unrestricted k =
                  (dpK c u system)
                  in
                  (let unrestricted m =
                    (dpM c u system)
                    in
                    (let unrestricted owners =
                      (dpOwners c u system)
                      in
                      (nat-eliminate
                        (lambda unrestricted partitioned : Nat . (family NativePhysicalCommands))
                        (let unrestricted total =
                          (nativeCombineResultAt (dpoSum layout))
                          in
                          (dpEach
                            k
                            (lambda unrestricted index : Nat .
                              (dpDivide
                                (nativePointwiseAt at (naturalAdd (naturalAdd k m) index))
                                (nativePointwiseAt total index)
                                (nativePointwiseAt total k)))
                            tail))
                        (lambda unrestricted previous : Nat .
                          (lambda unrestricted ignored : (family NativePhysicalCommands) .
                            (let unrestricted owned =
                              (nativeCombineResultAt (dpoGradientAt layout role))
                              in
                              (dpEach
                                k
                                (lambda unrestricted index : Nat .
                                  (lambda unrestricted rest : (family NativePhysicalCommands) .
                                    (nat-eliminate
                                      (lambda unrestricted held : Nat .
                                        (family NativePhysicalCommands))
                                      rest
                                      (lambda unrestricted previous : Nat .
                                        (lambda unrestricted ignored : (family NativePhysicalCommands) .
                                        (dpDivide
                                        (nativePointwiseAt at (naturalAdd (naturalAdd k m) index))
                                        (nativePointwiseAt
                                        owned
                                        (naturalDivideUnchecked index owners))
                                        (nativePointwiseAt owned (dpoPerOwner k owners))
                                        rest)))
                                      (dpoOwns owners role index))))
                                tail))))
                        (naturalNonzero
                          (stdListLength (family NativeCombinePlacement) (dpoGradients layout))))))))))))))

-- ---- local work ----
-- Means, then this owner's assignments of the update program on its
-- parameters, shared state and owned coordinate words.
def dpoUpdate =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted role : Nat .
                (lambda unrestricted update : (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)) .
                  (lambda unrestricted tail : (family NativePhysicalCommands) .
                    (let unrestricted k =
                      (dpK c u system)
                      in
                      (let unrestricted m =
                        (dpM c u system)
                        in
                        (let unrestricted owners =
                          (dpOwners c u system)
                          in
                          (let unrestricted shared =
                            (dpoShared c u system q)
                            in
                            (let unrestricted words =
                              (naturalAdd q p)
                              in
                              (let unrestricted total =
                                (nativeCombineResultAt (dpoSum layout))
                                in
                                (let unrestricted at =
                                  (dpUpdateAt c u system (dpoGeneric layout))
                                  in
                                  (let unrestricted sharedSlot =
                                    (nativePointwiseAt at (naturalAdd k (naturalMultiply q k)))
                                    in
                                    (let unrestricted owned =
                                      (dpoOwnedAt c u system q p layout)
                                      in
                                      (dpoMeans
                                        c
                                        u
                                        system
                                        layout
                                        role
                                        at
                                        (dpoUnless
                                        p
                                        (dpCopy
                                        at
                                        (dpoParametersAt layout)
                                        (dpParameterBytes c u system))
                                        (dpCopy
                                        sharedSlot
                                        (dpoSharedAt c u system p layout)
                                        (naturalMultiply dpWord shared)
                                        (dpoEachOwned
                                        k
                                        owners
                                        role
                                        words
                                        (lambda unrestricted coordinate : Nat .
                                        (lambda unrestricted word : Nat .
                                        (dpCopy
                                        (nativePointwiseAt at (dpoSlotOf k p word coordinate))
                                        (naturalAdd owned (dpoWordOffset k owners word coordinate))
                                        dpWord)))
                                        (nativeMemoryStoreWord64
                                        b"data-parallel-rate"
                                        (nativePointwiseAt at (naturalAdd (naturalMultiply 2 k) m))
                                        (nativeLaunchRecipeImmediate
                                        (modelWord64Natural (dataParallelRateWord c u system)))
                                        (update
                                        (dpoUnless
                                        p
                                        (dpCopy
                                        (dpoParametersAt layout)
                                        at
                                        (dpParameterBytes c u system))
                                        (dpCopy
                                        (dpoSharedAt c u system p layout)
                                        sharedSlot
                                        (naturalMultiply dpWord shared)
                                        (dpoEachOwned
                                        k
                                        owners
                                        role
                                        words
                                        (lambda unrestricted coordinate : Nat .
                                        (lambda unrestricted word : Nat .
                                        (dpCopy
                                        (naturalAdd owned (dpoWordOffset k owners word coordinate))
                                        (nativePointwiseAt at (dpoSlotOf k p word coordinate))
                                        dpWord)))
                                        (dpoWhen p (dpoRetire c u system layout) tail))))))))))))))))))))))))))))

-- This owner's offer: its parameters (when partitioned, its own and +0 for
-- the others) and its ownership words.
def dpoOffer =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted role : Nat .
                (lambda unrestricted tail : (family NativePhysicalCommands) .
                  (let unrestricted k =
                    (dpK c u system)
                    in
                    (let unrestricted owners =
                      (dpOwners c u system)
                      in
                      (let unrestricted leaf =
                        (nativeCombineRegionEnd (dpoAssemblePlacement layout))
                        in
                        (dpoUnless
                          p
                          (dpCopy leaf (dpoParametersAt layout) (dpParameterBytes c u system))
                          (dpoWhen
                            p
                            (dpEach
                              k
                              (lambda unrestricted coordinate : Nat .
                                (lambda unrestricted rest : (family NativePhysicalCommands) .
                                  (nat-eliminate
                                    (lambda unrestricted held : Nat .
                                      (family NativePhysicalCommands))
                                    (nativeMemoryStoreWord64
                                      b"data-parallel-unowned-parameter"
                                      (nativePointwiseAt leaf coordinate)
                                      (nativeLaunchRecipeImmediate 0)
                                      rest)
                                    (lambda unrestricted previous : Nat .
                                      (lambda unrestricted ignored : (family NativePhysicalCommands) .
                                        (dpCopy
                                        (nativePointwiseAt leaf coordinate)
                                        (naturalAdd
                                        (dpoOwnedAt c u system q p layout)
                                        (dpoWordOffset k owners 0 coordinate))
                                        dpWord
                                        rest)))
                                    (dpoOwns owners role coordinate)))))
                            (dpEach
                              k
                              (lambda unrestricted coordinate : Nat .
                                (nativeMemoryStoreWord64
                                  b"data-parallel-ownership"
                                  (nativePointwiseAt leaf (naturalAdd k coordinate))
                                  (nativeLaunchRecipeImmediate (dpoOwns owners role coordinate))))
                              tail))))))))))))))

-- Every coordinate of the assembled value was owned.
def dpoCovered =
  (lambda unrestricted k : Nat .
    (lambda unrestricted result : Nat .
      (lambda unrestricted tail : (family NativePhysicalCommands) .
        (dpEach
          k
          (lambda unrestricted coordinate : Nat .
            (nativeLaunchRecipeAssertEqual
              (nativeMemoryLoad (nativePointwiseAt result (naturalAdd k coordinate)))
              (nativeLaunchRecipeImmediate 1)
              b"data-parallel-assembly-coverage"))
          tail))))

-- Every coordinate was owned; install the assembled parameters and advance.
def dpoInstall =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (let unrestricted result =
              (nativeCombineResultAt (dpoAssemblePlacement layout))
              in
              (dpoCovered
                (dpK c u system)
                result
                (dpCopy
                  (dpoParametersAt layout)
                  result
                  (dpParameterBytes c u system)
                  (niAdvanceAt (dpoStateAt layout) tail)))))))))

-- MaterializeReadVersion: every coordinate was owned; the gathered
-- parameters become this owner's read replica of the current version.
def dpoMaterialize =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (let unrestricted result =
              (nativeCombineResultAt (dpoAssemblePlacement layout))
              in
              (let unrestricted replica =
                (dpoReplicaAt c u system layout)
                in
                (dpoCovered
                  (dpK c u system)
                  result
                  (dpCopy
                    (dpoReplicaParametersAt c u system layout)
                    result
                    (dpParameterBytes c u system)
                    (dpCopy
                      (nativePointwiseAt replica 1)
                      (dpoStateAt layout)
                      dpWord
                      (nativeMemoryStoreWord64
                        b"data-parallel-replica-held"
                        replica
                        (nativeLaunchRecipeImmediate 1)
                        tail)))))))))))

-- A read of the replica: it is held and was gathered at the current version.
def dpoReading =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (let unrestricted replica =
              (dpoReplicaAt c u system layout)
              in
              (nativeLaunchRecipeAssertEqual
                (nativeMemoryLoad replica)
                (nativeLaunchRecipeImmediate 1)
                b"data-parallel-replica-released"
                (nativeLaunchRecipeAssertEqual
                  (nativeMemoryLoad (nativePointwiseAt replica 1))
                  (nativeMemoryLoad (dpoStateAt layout))
                  b"data-parallel-replica-version"
                  tail))))))))

-- ReleaseReadReplica: the replica is no longer held and its parameters are
-- cleared.
def dpoRelease =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted tail : (family NativePhysicalCommands) .
            (nativeMemoryStoreWord64
              b"data-parallel-replica-release"
              (dpoReplicaAt c u system layout)
              (nativeLaunchRecipeImmediate 0)
              (dpEach
                (dpK c u system)
                (lambda unrestricted coordinate : Nat .
                  (nativeMemoryStoreWord64
                    b"data-parallel-replica-clear"
                    (nativePointwiseAt (dpoReplicaParametersAt c u system layout) coordinate)
                    (nativeLaunchRecipeImmediate 0)))
                tail)))))))

def dpoLocal =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted owner : Nat .
                (lambda unrestricted contribution : (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)) .
                  (lambda unrestricted update : (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)) .
                    (lambda unrestricted operation : (family DataParallelOwnedOperation) .
                      (eliminate
                        DataParallelOwnedOperation
                        (lambda unrestricted current : (family DataParallelOwnedOperation) .
                          (pi unrestricted tail : (family NativePhysicalCommands) .
                            (family NativePhysicalCommands)))
                        operation
                        (branch
                          DataParallelOwnedContribute
                          .
                          (lambda unrestricted tail : (family NativePhysicalCommands) .
                            (dpoWhen
                              p
                              (dpoReading c u system layout)
                              (dpContribute
                                c
                                u
                                system
                                (dpoGeneric layout)
                                (naturalSelect
                                  p
                                  (dpoReplicaParametersAt c u system layout)
                                  (dpoParametersAt layout))
                                owner
                                contribution
                                tail))))
                        (branch
                          DataParallelOwnedSum
                          value
                          .
                          (nativeCombineLocal
                            (dpoSum layout)
                            (dpLeafAt (dpoGeneric layout))
                            (dpApply c u system)
                            value))
                        (branch
                          DataParallelOwnedUpdate
                          .
                          (dpoUpdate c u system q p layout owner update))
                        (branch DataParallelOwnedOffer . (dpoOffer c u system q p layout owner))
                        (branch
                          DataParallelOwnedAssemble
                          value
                          .
                          (nativeCombineLocal
                            (dpoAssemblePlacement layout)
                            (nativeCombineRegionEnd (dpoAssemblePlacement layout))
                            (dpoSelect (dpK c u system))
                            value))
                        (branch DataParallelOwnedInstall . (dpoInstall c u system layout))
                        (branch DataParallelOwnedMaterialize . (dpoMaterialize c u system layout))
                        (branch DataParallelOwnedRelease . (dpoRelease c u system layout))
                        (branch DataParallelOwnedProject . (dpoProject c u system layout))
                        (branch
                          DataParallelOwnedGradient
                          target
                          value
                          .
                          (nativeCombineLocal
                            (dpoGradientAt layout target)
                            (nativeCombineRegionEnd (dpoGradientAt layout target))
                            (dpoApplyWords (dpoGradientWords c u system))
                            value)))))))))))))

-- ---- the round protocol ----
def dpoEnd =
  (constructor
    GlobalProtocol
    GlobalProtocolEnd
    (family DataParallelOwnedOperation)
    (family RegionTransferIdentity)
    (family ProtocolNoDecision))

def dpoEachRole =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted operation : (family DataParallelOwnedOperation) .
      (lambda unrestricted tail : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (stdListFold
          Nat
          (family
            GlobalProtocol
            (family DataParallelOwnedOperation)
            (family RegionTransferIdentity)
            (family ProtocolNoDecision))
          (lambda unrestricted role : Nat .
            (lambda unrestricted rest : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
              (protocolLocal
                (family DataParallelOwnedOperation)
                (family RegionTransferIdentity)
                (family ProtocolNoDecision)
                role
                operation
                rest)))
          tail
          (stdListNaturalsBelow owners)))))

def dpoSpliced =
  (lambda unrestricted wrap : (pi unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) . (family DataParallelOwnedOperation)) .
    (lambda unrestricted combine : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
      (lambda unrestricted tail : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (appendGlobalProtocol
          (family DataParallelOwnedOperation)
          (family RegionTransferIdentity)
          (family ProtocolNoDecision)
          (mapGlobalProtocolOperations
            (family OrderedCombineOperation (family RegionTransferIdentity))
            (family DataParallelOwnedOperation)
            (family RegionTransferIdentity)
            (family ProtocolNoDecision)
            wrap
            combine)
          tail))))

-- The sum phase: the replicated sum, or, with gradient partitioning, every
-- owner's projection and then one reduction per coordinate owner.
def dpoSumPhase =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted sum : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
      (lambda unrestricted gradients : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
        (lambda unrestricted tail : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
          (nat-eliminate
            (lambda unrestricted partitioned : Nat .
              (family
                GlobalProtocol
                (family DataParallelOwnedOperation)
                (family RegionTransferIdentity)
                (family ProtocolNoDecision)))
            (dpoSpliced
              (lambda unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) .
                (constructor DataParallelOwnedOperation DataParallelOwnedSum value))
              sum
              tail)
            (lambda unrestricted previous : Nat .
              (lambda unrestricted ignored : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                (dpoEachRole
                  owners
                  (constructor DataParallelOwnedOperation DataParallelOwnedProject)
                  (stdListFold
                    Nat
                    (family
                      GlobalProtocol
                      (family DataParallelOwnedOperation)
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))
                    (lambda unrestricted target : Nat .
                      (lambda unrestricted rest : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                        (dpoSpliced
                          (lambda unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) .
                            (constructor
                              DataParallelOwnedOperation
                              DataParallelOwnedGradient
                              target
                              value))
                          (stdListHeadOr
                            (family
                              GlobalProtocol
                              (family OrderedCombineOperation (family RegionTransferIdentity))
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision))
                            sum
                            (stdListDrop
                              (family
                                GlobalProtocol
                                (family OrderedCombineOperation (family RegionTransferIdentity))
                                (family RegionTransferIdentity)
                                (family ProtocolNoDecision))
                              target
                              gradients))
                          rest)))
                    tail
                    (stdListNaturalsBelow
                      (stdListLength
                        (family
                          GlobalProtocol
                          (family OrderedCombineOperation (family RegionTransferIdentity))
                          (family RegionTransferIdentity)
                          (family ProtocolNoDecision))
                        gradients))))))
            (naturalNonzero
              (stdListLength
                (family
                  GlobalProtocol
                  (family OrderedCombineOperation (family RegionTransferIdentity))
                  (family RegionTransferIdentity)
                  (family ProtocolNoDecision))
                gradients)))))))

def dpoReplicatedRound =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted sum : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
      (lambda unrestricted assemble : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (lambda unrestricted gradients : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
          (dpoEachRole
            owners
            (constructor DataParallelOwnedOperation DataParallelOwnedContribute)
            (dpoSumPhase
              owners
              sum
              gradients
              (dpoEachRole
                owners
                (constructor DataParallelOwnedOperation DataParallelOwnedUpdate)
                (dpoEachRole
                  owners
                  (constructor DataParallelOwnedOperation DataParallelOwnedOffer)
                  (dpoSpliced
                    (lambda unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) .
                      (constructor DataParallelOwnedOperation DataParallelOwnedAssemble value))
                    assemble
                    (dpoEachRole
                      owners
                      (constructor DataParallelOwnedOperation DataParallelOwnedInstall)
                      dpoEnd))))))))))

-- Partitioned parameters: gather, materialize, read, release, then sum and
-- update the owned words.
def dpoPartitionedRound =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted sum : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
      (lambda unrestricted assemble : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (lambda unrestricted gradients : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
          (dpoEachRole
            owners
            (constructor DataParallelOwnedOperation DataParallelOwnedOffer)
            (dpoSpliced
              (lambda unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) .
                (constructor DataParallelOwnedOperation DataParallelOwnedAssemble value))
              assemble
              (dpoEachRole
                owners
                (constructor DataParallelOwnedOperation DataParallelOwnedMaterialize)
                (dpoEachRole
                  owners
                  (constructor DataParallelOwnedOperation DataParallelOwnedContribute)
                  (dpoEachRole
                    owners
                    (constructor DataParallelOwnedOperation DataParallelOwnedRelease)
                    (dpoSumPhase
                      owners
                      sum
                      gradients
                      (dpoEachRole
                        owners
                        (constructor DataParallelOwnedOperation DataParallelOwnedUpdate)
                        dpoEnd)))))))))))

def dpoRound =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted p : Nat .
      (lambda unrestricted sum : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (lambda unrestricted assemble : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
          (lambda unrestricted gradients : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
            (nat-eliminate
              (lambda unrestricted current : Nat .
                (family
                  GlobalProtocol
                  (family DataParallelOwnedOperation)
                  (family RegionTransferIdentity)
                  (family ProtocolNoDecision)))
              (dpoReplicatedRound owners sum assemble gradients)
              (lambda unrestricted previous : Nat .
                (lambda unrestricted ignored : (family GlobalProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                  (dpoPartitionedRound owners sum assemble gradients)))
              p))))))

def dpoLocalAdmitted =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
      (lambda unrestricted role : Nat .
        (lambda unrestricted operation : (family DataParallelOwnedOperation) .
          (eliminate
            DataParallelOwnedOperation
            (lambda unrestricted current : (family DataParallelOwnedOperation) . Nat)
            operation
            (branch DataParallelOwnedContribute . (naturalLess role owners))
            (branch
              DataParallelOwnedSum
              value
              .
              (nativeCombineLocalAdmitted (dpoSum layout) role value))
            (branch DataParallelOwnedUpdate . (naturalLess role owners))
            (branch DataParallelOwnedOffer . (naturalLess role owners))
            (branch
              DataParallelOwnedAssemble
              value
              .
              (nativeCombineLocalAdmitted (dpoAssemblePlacement layout) role value))
            (branch DataParallelOwnedInstall . (naturalLess role owners))
            (branch DataParallelOwnedMaterialize . (naturalLess role owners))
            (branch DataParallelOwnedRelease . (naturalLess role owners))
            (branch DataParallelOwnedProject . (naturalLess role owners))
            (branch
              DataParallelOwnedGradient
              target
              value
              .
              (nativeCombineLocalAdmitted (dpoGradientAt layout target) role value)))))))

-- Plan owner r's reduction for every coordinate owner, in owner order.
def dpoGradientGlobals =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted topology : (family DataParallelTopology) .
          (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
            (let unrestricted owners =
              (dpOwners c u system)
              in
              (stdListFold
                Nat
                (family
                  StdResult
                  (family OrderedCombineFailure)
                  (family
                    StdList
                    (family
                      GlobalProtocol
                      (family OrderedCombineOperation (family RegionTransferIdentity))
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))))
                (lambda unrestricted target : Nat .
                  (lambda unrestricted rest : (family StdResult (family OrderedCombineFailure) (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)))) .
                    (stdResultBind
                      (family OrderedCombineFailure)
                      (family
                        GlobalProtocol
                        (family OrderedCombineOperation (family RegionTransferIdentity))
                        (family RegionTransferIdentity)
                        (family ProtocolNoDecision))
                      (family
                        StdList
                        (family
                          GlobalProtocol
                          (family OrderedCombineOperation (family RegionTransferIdentity))
                          (family RegionTransferIdentity)
                          (family ProtocolNoDecision)))
                      (planNativeCombineGlobalDelivering
                        (dpoGradientAt layout target)
                        (dpoGradientDelivery target)
                        (dpContract topology owners)
                        owners
                        (dpMembers owners)
                        (dpoGradientBaseAnchor c u system (dpoDigest layout) target))
                      (lambda unrestricted planned : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                        (stdResultBind
                          (family OrderedCombineFailure)
                          (family
                            StdList
                            (family
                              GlobalProtocol
                              (family OrderedCombineOperation (family RegionTransferIdentity))
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision)))
                          (family
                            StdList
                            (family
                              GlobalProtocol
                              (family OrderedCombineOperation (family RegionTransferIdentity))
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision)))
                          rest
                          (lambda unrestricted others : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                            (constructor
                              StdResult
                              StdSuccess
                              (family OrderedCombineFailure)
                              (family
                                StdList
                                (family
                                  GlobalProtocol
                                  (family OrderedCombineOperation (family RegionTransferIdentity))
                                  (family RegionTransferIdentity)
                                  (family ProtocolNoDecision)))
                              (constructor
                                StdList
                                StdListCons
                                (family
                                  GlobalProtocol
                                  (family OrderedCombineOperation (family RegionTransferIdentity))
                                  (family RegionTransferIdentity)
                                  (family ProtocolNoDecision))
                                planned
                                others))))))))
                (constructor
                  StdResult
                  StdSuccess
                  (family OrderedCombineFailure)
                  (family
                    StdList
                    (family
                      GlobalProtocol
                      (family OrderedCombineOperation (family RegionTransferIdentity))
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision)))
                  (constructor
                    StdList
                    StdListEmpty
                    (family
                      GlobalProtocol
                      (family OrderedCombineOperation (family RegionTransferIdentity))
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))))
                (stdListNaturalsBelow
                  (stdListLength (family NativeCombinePlacement) (dpoGradients layout))))))))))

def dpoProjected =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (lambda unrestricted topology : (family DataParallelTopology) .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted steps : Nat .
                (let unrestricted owners =
                  (dpOwners c u system)
                  in
                  (stdResultBind
                    (family OrderedCombineFailure)
                    (family
                      GlobalProtocol
                      (family OrderedCombineOperation (family RegionTransferIdentity))
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))
                    (family
                      ParticipantProtocol
                      (family DataParallelOwnedOperation)
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))
                    (planNativeCombineGlobal
                      (dpoSum layout)
                      (dpContract topology owners)
                      owners
                      (dpMembers owners)
                      (dpBaseAnchor c u system (dpoDigest layout)))
                    (lambda unrestricted sum : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                      (stdResultBind
                        (family OrderedCombineFailure)
                        (family
                          GlobalProtocol
                          (family OrderedCombineOperation (family RegionTransferIdentity))
                          (family RegionTransferIdentity)
                          (family ProtocolNoDecision))
                        (family
                          ParticipantProtocol
                          (family DataParallelOwnedOperation)
                          (family RegionTransferIdentity)
                          (family ProtocolNoDecision))
                        (planNativeCombineGlobal
                          (dpoAssemblePlacement layout)
                          (dpoAssembleContract owners)
                          owners
                          (dpMembers owners)
                          (dpoAssembleBaseAnchor c u system (dpoDigest layout)))
                        (lambda unrestricted assemble : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                          (stdResultBind
                            (family OrderedCombineFailure)
                            (family
                              StdList
                              (family
                                GlobalProtocol
                                (family OrderedCombineOperation (family RegionTransferIdentity))
                                (family RegionTransferIdentity)
                                (family ProtocolNoDecision)))
                            (family
                              ParticipantProtocol
                              (family DataParallelOwnedOperation)
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision))
                            (dpoGradientGlobals c u system topology layout)
                            (lambda unrestricted gradients : (family StdList (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                              (eliminate
                                StdResult
                                (lambda unrestricted result : (family StdResult (family ProtocolFailure) (family ParticipantProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                                  (family
                                    StdResult
                                    (family OrderedCombineFailure)
                                    (family
                                      ParticipantProtocol
                                      (family DataParallelOwnedOperation)
                                      (family RegionTransferIdentity)
                                      (family ProtocolNoDecision))))
                                (planParticipantProtocol
                                  (family DataParallelOwnedOperation)
                                  (family RegionTransferIdentity)
                                  (family ProtocolNoDecision)
                                  owners
                                  (nativeCombineRole
                                    (nativeCombinePlacementContext (dpoSum layout)))
                                  (dpoLocalAdmitted owners layout)
                                  (lambda unrestricted identity : (family RegionTransferIdentity) .
                                    (nativeCombineTransferAdmitted
                                      (dpoPlacementOf layout identity)
                                      identity))
                                  (protocolNoDecision (family RegionTransferIdentity))
                                  (protocolRepeat
                                    (family DataParallelOwnedOperation)
                                    (family RegionTransferIdentity)
                                    (family ProtocolNoDecision)
                                    steps
                                    (dpoRound owners p sum assemble gradients)
                                    dpoEnd))
                                (branch
                                  StdFailure
                                  failure
                                  .
                                  (constructor
                                    StdResult
                                    StdFailure
                                    (family OrderedCombineFailure)
                                    (family
                                      ParticipantProtocol
                                      (family DataParallelOwnedOperation)
                                      (family RegionTransferIdentity)
                                      (family ProtocolNoDecision))
                                    (constructor
                                      OrderedCombineFailure
                                      CombineProtocolRefused
                                      failure)))
                                (branch
                                  StdSuccess
                                  projected
                                  .
                                  (constructor
                                    StdResult
                                    StdSuccess
                                    (family OrderedCombineFailure)
                                    (family
                                      ParticipantProtocol
                                      (family DataParallelOwnedOperation)
                                      (family RegionTransferIdentity)
                                      (family ProtocolNoDecision))
                                    projected)))))))))))))))))

-- ---- persistence ----
-- Regions, by ordinal: parameters; shared optimizer state (when the learner
-- has shared words); coordinate optimizer state (when it has per-coordinate
-- words), word j of coordinate i at byte 8 (j k + i).
def dpoSharedOrdinal : Nat =
  1

def dpoCoordinateOrdinal =
  (lambda unrestricted shared : Nat . (succ (naturalNonzero shared)))

def dpoRegions =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (let unrestricted shared =
            (dpoShared c u system q)
            in
            (constructor
              StdList
              StdListCons
              (family CheckpointLogicalRegion)
              (constructor
                CheckpointLogicalRegion
                CheckpointLogicalRegionOf
                b"parameters"
                b"f64-le"
                (dpParameterBytes c u system))
              (stdListAppend
                (family CheckpointLogicalRegion)
                (nat-eliminate
                  (lambda unrestricted words : Nat .
                    (family StdList (family CheckpointLogicalRegion)))
                  (constructor StdList StdListEmpty (family CheckpointLogicalRegion))
                  (lambda unrestricted previous : Nat .
                    (lambda unrestricted ignored : (family StdList (family CheckpointLogicalRegion)) .
                      (constructor
                        StdList
                        StdListCons
                        (family CheckpointLogicalRegion)
                        (constructor
                          CheckpointLogicalRegion
                          CheckpointLogicalRegionOf
                          b"shared-optimizer-state"
                          b"f64-le"
                          (naturalMultiply dpWord shared))
                        (constructor StdList StdListEmpty (family CheckpointLogicalRegion)))))
                  shared)
                (nat-eliminate
                  (lambda unrestricted words : Nat .
                    (family StdList (family CheckpointLogicalRegion)))
                  (constructor StdList StdListEmpty (family CheckpointLogicalRegion))
                  (lambda unrestricted previous : Nat .
                    (lambda unrestricted ignored : (family StdList (family CheckpointLogicalRegion)) .
                      (constructor
                        StdList
                        StdListCons
                        (family CheckpointLogicalRegion)
                        (constructor
                          CheckpointLogicalRegion
                          CheckpointLogicalRegionOf
                          b"coordinate-optimizer-state"
                          b"f64-le"
                          (naturalMultiply dpWord (naturalMultiply q (dpK c u system))))
                        (constructor StdList StdListEmpty (family CheckpointLogicalRegion)))))
                  q))))))))

def dpoInterval =
  (lambda unrestricted begin : Nat .
    (lambda unrestricted bytes : Nat .
      (constructor LogicalInterval LogicalIntervalOf begin (naturalAdd begin bytes))))

def dpoWordIntervals =
  (lambda unrestricted count : Nat .
    (stdListMap
      Nat
      (family LogicalInterval)
      (lambda unrestricted index : Nat . (dpoInterval (naturalMultiply dpWord index) dpWord))
      (stdListNaturalsBelow count)))

-- One shard per owned word (each parameter word too when partitioned), so
-- each lies wholly in its owner's state.
def dpoPartitions =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (let unrestricted k =
              (dpK c u system)
              in
              (let unrestricted shared =
                (dpoShared c u system q)
                in
                (constructor
                  StdList
                  StdListCons
                  (family CheckpointRegionPartition)
                  (constructor
                    CheckpointRegionPartition
                    CheckpointRegionPartitionOf
                    0
                    (nat-eliminate
                      (lambda unrestricted current : Nat .
                        (family StdList (family LogicalInterval)))
                      (constructor
                        StdList
                        StdListCons
                        (family LogicalInterval)
                        (dpoInterval 0 (dpParameterBytes c u system))
                        (constructor StdList StdListEmpty (family LogicalInterval)))
                      (lambda unrestricted previous : Nat .
                        (lambda unrestricted ignored : (family StdList (family LogicalInterval)) .
                          (dpoWordIntervals k)))
                      p))
                  (stdListAppend
                    (family CheckpointRegionPartition)
                    (nat-eliminate
                      (lambda unrestricted words : Nat .
                        (family StdList (family CheckpointRegionPartition)))
                      (constructor StdList StdListEmpty (family CheckpointRegionPartition))
                      (lambda unrestricted previous : Nat .
                        (lambda unrestricted ignored : (family StdList (family CheckpointRegionPartition)) .
                          (constructor
                            StdList
                            StdListCons
                            (family CheckpointRegionPartition)
                            (constructor
                              CheckpointRegionPartition
                              CheckpointRegionPartitionOf
                              dpoSharedOrdinal
                              (constructor
                                StdList
                                StdListCons
                                (family LogicalInterval)
                                (dpoInterval 0 (naturalMultiply dpWord shared))
                                (constructor StdList StdListEmpty (family LogicalInterval))))
                            (constructor StdList StdListEmpty (family CheckpointRegionPartition)))))
                      shared)
                    (nat-eliminate
                      (lambda unrestricted words : Nat .
                        (family StdList (family CheckpointRegionPartition)))
                      (constructor StdList StdListEmpty (family CheckpointRegionPartition))
                      (lambda unrestricted previous : Nat .
                        (lambda unrestricted ignored : (family StdList (family CheckpointRegionPartition)) .
                          (constructor
                            StdList
                            StdListCons
                            (family CheckpointRegionPartition)
                            (constructor
                              CheckpointRegionPartition
                              CheckpointRegionPartitionOf
                              (dpoCoordinateOrdinal shared)
                              (dpoWordIntervals (naturalMultiply q k)))
                            (constructor StdList StdListEmpty (family CheckpointRegionPartition)))))
                      q))))))))))

def dpoShardCount =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (naturalAdd
              (naturalAdd
                (naturalSelect p (dpK c u system) 1)
                (naturalNonzero (dpoShared c u system q)))
              (naturalMultiply q (dpK c u system))))))))

-- argv with h shards (parameters (one per word when partitioned), shared
-- state when present, then one per coordinate word in byte order): 1
-- previous manifest, 2..h+1 previous
-- shards, each shard's candidate and committed file, the candidate and
-- committed manifest, the directory.
def dpoPaths =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (constructor NativeRegionPaths NativeRegionPathsOf 1 (dpoShardCount c u system q p)))))))

-- The word of a coordinate-state shard, and its coordinate.
def dpoShardWord =
  (lambda unrestricted shard : (family NativeRegionShard) .
    (naturalDivideUnchecked (intervalBegin (nativeRegionInterval shard)) dpWord))

-- Who holds a shard, and where it lies in that owner's state.
def dpoShardOwner =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted shard : (family NativeRegionShard) .
              (let unrestricted owner =
                (naturalModuloUnchecked
                  (naturalModuloUnchecked (dpoShardWord shard) (dpK c u system))
                  (dpOwners c u system))
                in
                (naturalSelect
                  (naturalEqual
                    (nativeRegionOrdinal shard)
                    (dpoCoordinateOrdinal (dpoShared c u system q)))
                  owner
                  (naturalSelect (naturalAnd p (naturalIsZero (nativeRegionOrdinal shard))) owner 0)))))))))

def dpoShardOffset =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted shard : (family NativeRegionShard) .
              (let unrestricted k =
                (dpK c u system)
                in
                (let unrestricted owners =
                  (dpOwners c u system)
                  in
                  (let unrestricted shared =
                    (dpoShared c u system q)
                    in
                    (let unrestricted ordinal =
                      (nativeRegionOrdinal shard)
                      in
                      (let unrestricted replicated =
                        (dpoReplicatedParameterBytes c u system p)
                        in
                        (let unrestricted owned =
                          (naturalAdd replicated (naturalMultiply dpWord shared))
                          in
                          (naturalAdd
                            nativeRegionCaptureBytes
                            (naturalSelect
                              (naturalIsZero ordinal)
                              (naturalSelect
                                p
                                (naturalAdd owned (dpoWordOffset k owners 0 (dpoShardWord shard)))
                                (intervalBegin (nativeRegionInterval shard)))
                              (naturalSelect
                                (naturalEqual ordinal (dpoCoordinateOrdinal shared))
                                (naturalAdd
                                  owned
                                  (dpoWordOffset
                                    k
                                    owners
                                    (naturalAdd (naturalDivideUnchecked (dpoShardWord shard) k) p)
                                    (naturalModuloUnchecked (dpoShardWord shard) k)))
                                (naturalAdd replicated (intervalBegin (nativeRegionInterval shard))))))))))))))))))

-- A hierarchy's checkpoint is a manifest tree, not the star's region
-- manifest: partitioned checkpoints name it (the star's names nothing).
def dpoPublicationTag =
  (lambda unrestricted topology : (family DataParallelTopology) .
    (eliminate
      DataParallelTopology
      (lambda unrestricted current : (family DataParallelTopology) . Bytes)
      topology
      (branch DataParallelStar . b"")
      (branch DataParallelHierarchy fanout . (bytes-append b"hierarchy;" (regionFrameWord fanout)))))

-- The identity a checkpoint context binds: with replicated parameters the
-- placed learner's digest; with partitioned ones the digest of its
-- placement-independent identity.
def dpoContextDigest =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (lambda unrestricted digest : Bytes .
            (nat-eliminate
              (lambda unrestricted current : Nat . Bytes)
              digest
              (lambda unrestricted previous : Nat .
                (lambda unrestricted ignored : Bytes .
                  (sha256RawDigestOrEmpty (dataParallelLearnerIdentity c u system))))
              p))))))

-- Replicated parameters bind the star's combine order and the split of
-- samples among owners. Partitioned parameters bind only the learner's
-- placement-independent identity: the state means the same under any owner
-- count and any contiguous split of the same input sequence, and restore
-- onto another placement is a repartitioning, not a different learner.
def dpoContext =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted topology : (family DataParallelTopology) .
              (lambda unrestricted identity : Bytes .
                (nat-eliminate
                  (lambda unrestricted current : Nat . (family CheckpointEnvelopeContract))
                  (constructor
                    CheckpointEnvelopeContract
                    CheckpointEnvelopeContractValue
                    (bytes-append
                      b"data-parallel-owned-optimizer-state-f64-finite-rne-separate-v1;"
                      (bytes-append
                        (regionFrameWord q)
                        (bytes-append
                          (numericalExactContractDigest (dpContract topology (dpOwners c u system)))
                          identity)))
                    b"typed-sample-contributions;ordered-combine;owned-coordinate-update;assembly"
                    b"immutable-partitioned-samples;one-pass-per-update"
                    b"declared-initial-parameters;no-randomness"
                    1
                    1
                    1)
                  (lambda unrestricted previous : Nat .
                    (lambda unrestricted ignored : (family CheckpointEnvelopeContract) .
                      (constructor
                        CheckpointEnvelopeContract
                        CheckpointEnvelopeContractValue
                        (bytes-append
                          b"data-parallel-partitioned-parameters-f64-finite-rne-separate-v1;"
                          (bytes-append
                            (dpoPublicationTag topology)
                            (bytes-append (regionFrameWord q) identity)))
                        b"typed-sample-contributions;ordered-combine;owned-coordinate-update;materialized-read-replica"
                        b"immutable-samples-any-contiguous-split;one-pass-per-update"
                        b"declared-initial-parameters;no-randomness"
                        1
                        1
                        1)))
                  p))))))))

-- This owner's destinations on restore: replicated parameters, shared state
-- and its own words (its parameters when partitioned, its coordinates'
-- optimizer state).
def dpoDestinations =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted role : Nat .
                (let unrestricted k =
                  (dpK c u system)
                  in
                  (let unrestricted owners =
                    (dpOwners c u system)
                    in
                    (let unrestricted shared =
                      (dpoShared c u system q)
                      in
                      (let unrestricted owned =
                        (dpoOwnedAt c u system q p layout)
                        in
                        (let unrestricted ownedWords =
                          (lambda unrestricted ordinal : Nat .
                            (lambda unrestricted count : Nat .
                              (lambda unrestricted first : Nat .
                                (stdListFold
                                  Nat
                                  (family StdList (family NativeRegionDestination))
                                  (lambda unrestricted index : Nat .
                                    (lambda unrestricted rest : (family StdList (family NativeRegionDestination)) .
                                      (let unrestricted word =
                                        (naturalDivideUnchecked index k)
                                        in
                                        (let unrestricted coordinate =
                                        (naturalModuloUnchecked index k)
                                        in
                                        (nat-eliminate
                                        (lambda unrestricted held : Nat .
                                        (family StdList (family NativeRegionDestination)))
                                        rest
                                        (lambda unrestricted previous : Nat .
                                        (lambda unrestricted ignored : (family StdList (family NativeRegionDestination)) .
                                        (constructor
                                        StdList
                                        StdListCons
                                        (family NativeRegionDestination)
                                        (constructor
                                        NativeRegionDestination
                                        NativeRegionDestinationOf
                                        ordinal
                                        (dpoInterval (naturalMultiply dpWord index) dpWord)
                                        (naturalAdd
                                        owned
                                        (dpoWordOffset k owners (naturalAdd word first) coordinate)))
                                        rest)))
                                        (dpoOwns owners role coordinate))))))
                                  (constructor
                                    StdList
                                    StdListEmpty
                                    (family NativeRegionDestination))
                                  (stdListNaturalsBelow count)))))
                          in
                          (stdListAppend
                            (family NativeRegionDestination)
                            (nat-eliminate
                              (lambda unrestricted current : Nat .
                                (family StdList (family NativeRegionDestination)))
                              (constructor
                                StdList
                                StdListCons
                                (family NativeRegionDestination)
                                (constructor
                                  NativeRegionDestination
                                  NativeRegionDestinationOf
                                  0
                                  (dpoInterval 0 (dpParameterBytes c u system))
                                  (dpoParametersAt layout))
                                (constructor StdList StdListEmpty (family NativeRegionDestination)))
                              (lambda unrestricted previous : Nat .
                                (lambda unrestricted ignored : (family StdList (family NativeRegionDestination)) .
                                  (ownedWords 0 k 0)))
                              p)
                            (stdListAppend
                              (family NativeRegionDestination)
                              (nat-eliminate
                                (lambda unrestricted words : Nat .
                                  (family StdList (family NativeRegionDestination)))
                                (constructor StdList StdListEmpty (family NativeRegionDestination))
                                (lambda unrestricted previous : Nat .
                                  (lambda unrestricted ignored : (family StdList (family NativeRegionDestination)) .
                                    (constructor
                                      StdList
                                      StdListCons
                                      (family NativeRegionDestination)
                                      (constructor
                                        NativeRegionDestination
                                        NativeRegionDestinationOf
                                        dpoSharedOrdinal
                                        (dpoInterval 0 (naturalMultiply dpWord shared))
                                        (dpoSharedAt c u system p layout))
                                      (constructor
                                        StdList
                                        StdListEmpty
                                        (family NativeRegionDestination)))))
                                shared)
                              (ownedWords (dpoCoordinateOrdinal shared) (naturalMultiply q k) p)))))))))))))))

-- Where a topology's publication storage may begin: after the assembly's
-- storage, this owner's offer and its read replica.
def dpoPublicationStart =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (lambda unrestricted assemble : (family NativeCombinePlacement) .
            (frameAlign
              (naturalAdd
                (naturalAdd (nativeCombineRegionEnd assemble) (dpoOfferBytes c u system))
                (dpoReplicaBytes c u system p))))))))

-- With gradient partitioning (g = 1) the coordinate owners' reductions,
-- placed where the publication storage would begin; none otherwise.
def dpoGradientsFor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted g : Nat .
          (lambda unrestricted digest : Bytes .
            (lambda unrestricted topology : (family DataParallelTopology) .
              (lambda unrestricted role : Nat .
                (lambda unrestricted base : Nat .
                  (nat-eliminate
                    (lambda unrestricted current : Nat .
                      (family StdList (family NativeCombinePlacement)))
                    (constructor StdList StdListEmpty (family NativeCombinePlacement))
                    (lambda unrestricted previous : Nat .
                      (lambda unrestricted ignored : (family StdList (family NativeCombinePlacement)) .
                        (dpoGradientPlacementsFor c u system digest topology role base)))
                    g)))))))))

-- ---- layout ----
-- The sum combine (the shared realization's placement), the assembly
-- combine and this owner's offer, the read replica (partitioned parameters),
-- the publication frames and checkpoint, then the owner state.
def dpoWithStarLayout =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted g : Nat .
              (lambda unrestricted role : Nat .
                (lambda unrestricted epoch : Nat .
                  (lambda unrestricted available : Nat .
                    (lambda unrestricted emit : (pi unrestricted layout : (family NativeDataParallelOwnedLayout) . (pi unrestricted star : (family NativeDataParallelOwnedStarLayout) . Bytes)) .
                      (let unrestricted digest =
                        (dpIdentityDigest c u system)
                        in
                        (let unrestricted contextDigest =
                          (dpoContextDigest c u system p digest)
                          in
                          (let unrestricted sum =
                            (dpPlacementFor
                              c
                              u
                              system
                              digest
                              (constructor DataParallelTopology DataParallelStar)
                              role)
                            in
                            (let unrestricted assemble =
                              (dpoAssemblePlacementFor
                                c
                                u
                                system
                                digest
                                (constructor DataParallelTopology DataParallelStar)
                                role
                                sum)
                              in
                              (let unrestricted gradients =
                                (dpoGradientsFor
                                  c
                                  u
                                  system
                                  g
                                  digest
                                  (constructor DataParallelTopology DataParallelStar)
                                  role
                                  (dpoPublicationStart c u system p assemble))
                                in
                                (let unrestricted frames =
                                  (dpoGradientsEnd
                                    c
                                    u
                                    system
                                    gradients
                                    (dpoPublicationStart c u system p assemble))
                                  in
                                  (eliminate
                                    StdResult
                                    (lambda unrestricted result : (family StdResult (family RegionFrameLayoutFailure) (family RegionFrameLayout)) .
                                      Bytes)
                                    (planRegionFrameLayout
                                      (dpPublicationIdentity c u system digest 0 1 0)
                                      (dpoStateBytes c u system q p)
                                      512
                                      frames
                                      available)
                                    (branch StdFailure error . b"")
                                    (branch
                                      StdSuccess
                                      capture
                                      .
                                      (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family RegionFrameLayoutFailure) (family RegionFrameLayout)) .
                                        Bytes)
                                        (planRegionFrameLayout
                                        (dpPublicationIdentity c u system digest 1 0 1)
                                        dpWord
                                        512
                                        (frameAlign (frameEnd capture))
                                        available)
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        committed
                                        .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family NativeCheckpointFailure) (family NativeRegionCheckpoint)) .
                                        Bytes)
                                        (planNativeRegionCheckpoint
                                        (dpoContext
                                        c
                                        u
                                        system
                                        q
                                        p
                                        (constructor DataParallelTopology DataParallelStar)
                                        contextDigest)
                                        (dpoRegions c u system q)
                                        (dpoPartitions c u system q p)
                                        epoch
                                        (frameAlign (frameEnd committed))
                                        available
                                        (dpoShardCount c u system q p))
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        checkpoint
                                        .
                                        (emit
                                        (constructor
                                        NativeDataParallelOwnedLayout
                                        NativeDataParallelOwnedLayoutOf
                                        sum
                                        assemble
                                        (nativeRegionEnd checkpoint)
                                        (nativeRegionEpoch checkpoint)
                                        digest
                                        gradients)
                                        (constructor
                                        NativeDataParallelOwnedStarLayout
                                        NativeDataParallelOwnedStarLayoutOf
                                        capture
                                        committed
                                        checkpoint))))))))))))))))))))))))

-- ---- start, restore, publication ----
-- Fresh: the declared replicated words and this owner's own words (owned
-- word j of coordinate i is the initial value at j k + i of the parameters,
-- when partitioned, followed by the optimizer state). Resume: the verified
-- restore of exactly those, then the restored prefix and a new membership
-- epoch.
def dpoFresh =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted role : Nat .
                (lambda unrestricted tail : (family NativePhysicalCommands) .
                  (let unrestricted k =
                    (dpK c u system)
                    in
                    (let unrestricted owners =
                      (dpOwners c u system)
                      in
                      (let unrestricted initial =
                        (dataParallelInitialStateWords c u system)
                        in
                        (let unrestricted ownedInitial =
                          (stdListAppend
                            StdU64
                            (stdListTake
                              StdU64
                              (naturalMultiply p k)
                              (dataParallelInitialParameters c u system))
                            initial)
                          in
                          (nativeMemoryStoreWord64
                            b"data-parallel-version"
                            (dpoStateAt layout)
                            (nativeLaunchRecipeImmediate 0)
                            (nativeMemoryStoreWord64
                              b"data-parallel-cursor"
                              (nativePointwiseAt (dpoStateAt layout) 1)
                              (nativeLaunchRecipeImmediate 0)
                              (nativeMemoryStoreWord64
                                b"data-parallel-epoch"
                                (nativePointwiseAt (dpoStateAt layout) 2)
                                (nativeLaunchRecipeImmediate (dpoEpoch layout))
                                (dpoUnless
                                  p
                                  (dpStoreWords
                                    (dpoParametersAt layout)
                                    (dataParallelInitialParameters c u system))
                                  (dpStoreWords
                                    (dpoSharedAt c u system p layout)
                                    (stdListDrop StdU64 (naturalMultiply q k) initial)
                                    (dpoEachOwned
                                      k
                                      owners
                                      role
                                      (naturalAdd q p)
                                      (lambda unrestricted coordinate : Nat .
                                        (lambda unrestricted word : Nat .
                                        (nativeMemoryStoreWord64
                                        b"data-parallel-owned-initial"
                                        (naturalAdd
                                        (dpoOwnedAt c u system q p layout)
                                        (dpoWordOffset k owners word coordinate))
                                        (nativeLaunchRecipeImmediate
                                        (modelWord64Natural
                                        (stdListHeadOr
                                        StdU64
                                        (modelWord64FromNaturalTruncated 0)
                                        (stdListDrop
                                        StdU64
                                        (naturalAdd (naturalMultiply word k) coordinate)
                                        ownedInitial)))))))
                                      tail))))))))))))))))))

def dpoStart =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
                (lambda unrestricted role : Nat .
                  (lambda unrestricted extent : Nat .
                    (lambda unrestricted resume : Nat .
                      (lambda unrestricted tail : (family NativePhysicalCommands) .
                        (nat-eliminate
                          (lambda unrestricted mode : Nat .
                            (family
                              StdResult
                              (family NativeCheckpointFailure)
                              (family NativePhysicalCommands)))
                          (constructor
                            StdResult
                            StdSuccess
                            (family NativeCheckpointFailure)
                            (family NativePhysicalCommands)
                            (dpoFresh c u system q p layout role tail))
                          (lambda unrestricted previous : Nat .
                            (lambda unrestricted ignored : (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)) .
                              (planNativeRegionRestore
                                (dpoCheckpoint star)
                                (nativeRegionPathArgument
                                  (nativeRegionPathFirst (dpoPaths c u system q p)))
                                (lambda unrestricted index : Nat .
                                  (nativeRegionPathArgument
                                    (naturalAdd
                                      (nativeRegionInputShardBase (dpoPaths c u system q p))
                                      index)))
                                nativePeerListenerSlot
                                nativePeerScratchSlot
                                extent
                                (dpoDestinations c u system q p layout role)
                                (dpResumed
                                  (dpoGeneric layout)
                                  (nativeRegionCaptureAt (dpoCheckpoint star))
                                  tail))))
                          resume))))))))))))

def dpoPublication =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
                (lambda unrestricted resume : Nat .
                  (lambda unrestricted edges : (family StdList (family NativePeerEdge)) .
                    (constructor
                      NativePeerRegionPublication
                      NativePeerRegionPublicationOf
                      (constructor
                        NativeRegionCheckpoint
                        NativeRegionCheckpointOf
                        (nativeRegionManifest (dpoCheckpoint star))
                        (nativeRegionShards (dpoCheckpoint star))
                        (naturalAdd (nativeRegionEpoch (dpoCheckpoint star)) resume)
                        (nativeRegionEnd (dpoCheckpoint star)))
                      (naturalPredecessor (dpOwners c u system))
                      (lambda unrestricted owner : Nat .
                        (niFrameAt
                          (dpPublicationIdentity c u system (dpoDigest layout) 0 owner 0)
                          (dpoStateBytes c u system q p)
                          (dpoCapture star)))
                      (lambda unrestricted owner : Nat .
                        (niFrameAt
                          (dpPublicationIdentity c u system (dpoDigest layout) 1 0 owner)
                          dpWord
                          (dpoCommitted star)))
                      (nativePeerEdgeConnectionTo edges)
                      (dpoStateAt layout)
                      (dpoStateBytes c u system q p)
                      (dpoAgreedBytes c u system q p)
                      (dpoShardOwner c u system q p)
                      (dpoShardOffset c u system q p)
                      (dpoPaths c u system q p)))))))))))

def dpoOutput =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
              (niOutput (dpoStateAt layout) (dpoStateBytes c u system q p) dpHalt)))))))

def dpoRunAdmitted =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted q : Nat .
          (lambda unrestricted p : Nat .
            (lambda unrestricted g : Nat .
              (lambda unrestricted role : Nat .
                (lambda unrestricted addresses : (family StdList Bytes) .
                  (lambda unrestricted resume : Nat .
                    (lambda unrestricted epoch : Nat .
                      (lambda unrestricted steps : Nat .
                        (naturalAnd
                          (dpRunAdmitted c u system role addresses resume epoch steps)
                          (naturalAnd
                            (naturalAnd (naturalLessOrEqual p 1) (naturalLessOrEqual g 1))
                            (dataParallelSeparable c u system q))))))))))))))

-- The read replica is resident only with partitioned parameters.
def dpoReplicaResidents =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted p : Nat .
          (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
            (lambda unrestricted rest : (family ArenaResidents) .
              (nat-eliminate
                (lambda unrestricted current : Nat . (family ArenaResidents))
                rest
                (lambda unrestricted previous : Nat .
                  (lambda unrestricted ignored : (family ArenaResidents) .
                    (constructor
                      ArenaResidents
                      ArenaResidentsNext
                      (constructor
                        ArenaResident
                        ArenaResidentValue
                        b"data-parallel-read-replica"
                        (dpoReplicaAt c u system layout)
                        (dpoReplicaBytes c u system p)
                        dpWord)
                      rest)))
                p)))))))

-- Every reduction's frames and local slots, and its leaf value.
def dpoGradientResidents =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
          (lambda unrestricted rest : (family ArenaResidents) .
            (stdListFold
              (family NativeCombinePlacement)
              (family ArenaResidents)
              (lambda unrestricted placement : (family NativeCombinePlacement) .
                (lambda unrestricted more : (family ArenaResidents) .
                  (nativeCombineResidents
                    placement
                    (constructor
                      ArenaResidents
                      ArenaResidentsNext
                      (constructor
                        ArenaResident
                        ArenaResidentValue
                        b"data-parallel-gradient-leaf"
                        (nativeCombineRegionEnd placement)
                        (dpoGradientBytes c u system)
                        dpWord)
                      more))))
              rest
              (dpoGradients layout)))))))

-- Compile the contribution program and this owner's part of the update
-- program, project the round protocol under `topology` and emit one owner's
-- executable for `target`. `start extent tail` begins the owner (fresh or
-- restored); `publish extent success` publishes after the last update and
-- continues with `success`; `residents` are the topology's publication
-- storage.
def dpoEmitWith =
  (lambda unrestricted target : (family NativeHostTarget) .
    (lambda unrestricted resources : (family NativeResourceContext) .
      (lambda unrestricted c : Nat .
        (lambda unrestricted u : Nat .
          (lambda unrestricted system : (family DataParallelSystem c u) .
            (lambda unrestricted q : Nat .
              (lambda unrestricted p : Nat .
                (lambda unrestricted topology : (family DataParallelTopology) .
                  (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
                    (lambda unrestricted role : Nat .
                      (lambda unrestricted edges : (family StdList (family NativePeerEdge)) .
                        (lambda unrestricted steps : Nat .
                          (lambda unrestricted available : Nat .
                            (lambda unrestricted start : (pi unrestricted extent : Nat . (pi unrestricted tail : (family NativePhysicalCommands) . (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)))) .
                              (lambda unrestricted publish : (pi unrestricted extent : Nat . (pi unrestricted success : (family NativePhysicalCommands) . (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)))) .
                                (lambda unrestricted residents : (family ArenaResidents) .
                                  (let unrestricted generic =
                                    (dpoGeneric layout)
                                    in
                                    (let unrestricted extent =
                                      (nativePeerEdgesEnd
                                        (dpEndpointAt c u system generic role)
                                        edges)
                                      in
                                      (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult Bytes (family NativePhysicalCommands)) .
                                        Bytes)
                                        (planNativePointwiseProgram
                                        c
                                        (dataParallelContributionProgram c u system)
                                        (dpContributionAt c u system generic)
                                        (dpContributionEnd c u system generic)
                                        nativePeerScratchSlot
                                        dpEnd)
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        contribution
                                        .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult Bytes (family NativePhysicalCommands)) .
                                        Bytes)
                                        (planNativePointwiseProgram
                                        u
                                        (dataParallelOwnedProgram
                                        c
                                        u
                                        system
                                        q
                                        (dpoOwns (dpOwners c u system) role))
                                        (dpUpdateAt c u system generic)
                                        (dpUpdateEnd c u system generic)
                                        nativePeerScratchSlot
                                        dpEnd)
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        update
                                        .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family OrderedCombineFailure) (family ParticipantProtocol (family DataParallelOwnedOperation) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                                        Bytes)
                                        (dpoProjected c u system p topology layout steps)
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        projected
                                        .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)) .
                                        Bytes)
                                        (publish
                                        extent
                                        (nativePeerEdgesFold
                                        edges
                                        (lambda unrestricted index : Nat .
                                        (lambda unrestricted edge : (family NativePeerEdge) .
                                        (nativeSocketCloseChecked
                                        (nativeLaunchRecipeSlotValue
                                        (nativePeerEdgeConnectionSlot index))
                                        nativePeerScratchSlot)))
                                        (dpoOutput c u system q p layout)))
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        published
                                        .
                                        (nat-eliminate
                                        (lambda unrestricted valid : Nat . Bytes)
                                        b""
                                        (lambda unrestricted previous : Nat .
                                        (lambda unrestricted ignored : Bytes .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)) .
                                        Bytes)
                                        (start
                                        extent
                                        (niFinite
                                        (naturalSelect
                                        p
                                        (dpoOwnedAt c u system q p layout)
                                        (dpoParametersAt layout))
                                        (naturalSelect
                                        p
                                        (dpoPerOwner (dpK c u system) (dpOwners c u system))
                                        (dpK c u system))
                                        (nativePeerEdgeSetup
                                        role
                                        edges
                                        (dpEndpointAt c u system generic role)
                                        (nativeProtocolFrames
                                        (family DataParallelOwnedOperation)
                                        (family RegionTransferIdentity)
                                        (family ProtocolNoDecision)
                                        (lambda unrestricted direction : (family ProtocolDirection) .
                                        (lambda unrestricted identity : (family RegionTransferIdentity) .
                                        (lambda unrestricted value : (family RegionTransferIdentity) .
                                        (nativeFrameInitialize
                                        (nativeCombineFrame (dpoPlacementOf layout value) value)))))
                                        projected
                                        (nativeProtocolCommands
                                        (family DataParallelOwnedOperation)
                                        (family RegionTransferIdentity)
                                        (family ProtocolNoDecision)
                                        (dpoLocal
                                        c
                                        u
                                        system
                                        q
                                        p
                                        layout
                                        role
                                        (nativePhysicalProgramComposeCommandsAppend contribution)
                                        (nativePhysicalProgramComposeCommandsAppend update))
                                        (lambda unrestricted direction : (family ProtocolDirection) .
                                        (lambda unrestricted identity : (family RegionTransferIdentity) .
                                        (lambda unrestricted value : (family RegionTransferIdentity) .
                                        (nativeCombineTransfer
                                        (dpoPlacementOf layout value)
                                        (lambda unrestricted peer : Nat .
                                        (nativeLaunchRecipeSlotValue
                                        (nativePeerEdgeConnectionTo edges peer)))
                                        direction
                                        identity
                                        value))))
                                        (nativeCombineDecisions (dpoSum layout))
                                        b"data-parallel-owned"
                                        projected
                                        published)))))
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        started
                                        .
                                        (nativeHostTargetResourceArtifact
                                        target
                                        resources
                                        (niProgramFor
                                        (nativePeerEdgeSlots edges)
                                        extent
                                        (dpInitialize c u system generic role started)))))))
                                        (naturalAnd
                                        (naturalLessOrEqual extent available)
                                        (arenaCertificate
                                        extent
                                        (nativeCombineResidents
                                        (dpoSum layout)
                                        (constructor
                                        ArenaResidents
                                        ArenaResidentsNext
                                        (constructor
                                        ArenaResident
                                        ArenaResidentValue
                                        b"data-parallel-leaf"
                                        (dpLeafAt generic)
                                        (dpValueBytes c u system)
                                        dpWord)
                                        (nativeCombineResidents
                                        (dpoAssemblePlacement layout)
                                        (constructor
                                        ArenaResidents
                                        ArenaResidentsNext
                                        (constructor
                                        ArenaResident
                                        ArenaResidentValue
                                        b"data-parallel-offer"
                                        (nativeCombineRegionEnd (dpoAssemblePlacement layout))
                                        (dpoOfferBytes c u system)
                                        dpWord)
                                        (dpoReplicaResidents
                                        c
                                        u
                                        system
                                        p
                                        layout
                                        (dpoGradientResidents
                                        c
                                        u
                                        system
                                        layout
                                        (constructor
                                        ArenaResidents
                                        ArenaResidentsNext
                                        (constructor
                                        ArenaResident
                                        ArenaResidentValue
                                        b"data-parallel-owner-state"
                                        (dpoStateAt layout)
                                        (naturalSaturatingSubtract extent (dpoStateAt layout))
                                        dpWord)
                                        residents))))))))))))))))))))))))))))))))))))

def dpoEmit =
  (lambda unrestricted target : (family NativeHostTarget) .
    (lambda unrestricted resources : (family NativeResourceContext) .
      (lambda unrestricted c : Nat .
        (lambda unrestricted u : Nat .
          (lambda unrestricted system : (family DataParallelSystem c u) .
            (lambda unrestricted q : Nat .
              (lambda unrestricted p : Nat .
                (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
                  (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
                    (lambda unrestricted role : Nat .
                      (lambda unrestricted edges : (family StdList (family NativePeerEdge)) .
                        (lambda unrestricted resume : Nat .
                          (lambda unrestricted steps : Nat .
                            (lambda unrestricted available : Nat .
                              (dpoEmitWith
                                target
                                resources
                                c
                                u
                                system
                                q
                                p
                                (constructor DataParallelTopology DataParallelStar)
                                layout
                                role
                                edges
                                steps
                                available
                                (lambda unrestricted extent : Nat .
                                  (lambda unrestricted tail : (family NativePhysicalCommands) .
                                    (dpoStart c u system q p layout star role extent resume tail)))
                                (lambda unrestricted extent : Nat .
                                  (lambda unrestricted success : (family NativePhysicalCommands) .
                                    (planNativePeerRegionPublish
                                      (dpoPublication c u system q p layout star resume edges)
                                      extent
                                      role
                                      success)))
                                (constructor ArenaResidents ArenaResidentsEnd))))))))))))))))

def nativeDataParallelOwnedArtifact =
  (lambda unrestricted target : (family NativeHostTarget) .
    (lambda unrestricted resources : (family NativeResourceContext) .
      (lambda unrestricted c : Nat .
        (lambda unrestricted u : Nat .
          (lambda unrestricted system : (family DataParallelSystem c u) .
            (lambda unrestricted q : Nat .
              (lambda unrestricted p : Nat .
                (lambda unrestricted g : Nat .
                  (lambda unrestricted role : Nat .
                    (lambda unrestricted addresses : (family StdList Bytes) .
                      (lambda unrestricted resume : Nat .
                        (lambda unrestricted epoch : Nat .
                          (lambda unrestricted steps : Nat .
                            (lambda unrestricted available : Nat .
                              (let unrestricted edges =
                                (nativePeerStarEdges role addresses)
                                in
                                (nat-eliminate
                                  (lambda unrestricted valid : Nat . Bytes)
                                  b""
                                  (lambda unrestricted previous : Nat .
                                    (lambda unrestricted ignored : Bytes .
                                      (dpoWithStarLayout
                                        c
                                        u
                                        system
                                        q
                                        p
                                        g
                                        role
                                        epoch
                                        available
                                        (lambda unrestricted layout : (family NativeDataParallelOwnedLayout) .
                                        (lambda unrestricted star : (family NativeDataParallelOwnedStarLayout) .
                                        (dpoEmit
                                        target
                                        resources
                                        c
                                        u
                                        system
                                        q
                                        p
                                        layout
                                        star
                                        role
                                        edges
                                        resume
                                        steps
                                        available))))))
                                  (dpoRunAdmitted
                                    c
                                    u
                                    system
                                    q
                                    p
                                    g
                                    role
                                    addresses
                                    resume
                                    epoch
                                    steps)))))))))))))))))