module Runtime.NativeDataParallel

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.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

-- The native host realization of a data-parallel learner
-- (Learning.DataParallel) over owners 0..n-1 on a star centred on owner 0 or
-- on a bounded-fan-out hierarchy (Runtime.NativeDataParallelTree).
--
-- One update is one round of a global protocol:
--   every owner: sum its own samples' contributions in sample order into a
--     (numerator, mass) value, starting from +0;
--   ordered combine: on a star the left-deep sum (((v0 + v1) + v2) + ...)
--     at owner 0; on a hierarchy each node's own value followed by its
--     children's subtree values, left-deep at the node; replicated to every
--     owner (Runtime.NativeCombine), elementwise binary64 addition of
--     numerators and of masses;
--   every owner: mean = numerator / mass, then the update program on
--     (parameters, mean, rate); version and input cursor advance.
-- Every owner computes the same bits, so the parameters stay replicated
-- without a separate broadcast. The evaluation order is the combine tree;
-- its numerical contract and digest are part of the frame identities, so a
-- peer compiled for another order refuses before any result.
--
-- Persistence: the parameters are one logical region held (replicated) by
-- every owner; owner 0 publishes it as one shard and every owner's whole
-- replicated state must equal owner 0's before anything is published. Restore
-- reads that shard into every owner's parameters. Samples are immutable
-- program inputs and are not checkpointed. This module owns everything
-- except the publication itself; the star's publication is below and the
-- hierarchy's in Runtime.NativeDataParallelTree.
family DataParallelRoundOperation : Type 0
constructor DataParallelContribute
constructor DataParallelCombine
field unrestricted dataParallelCombineOperation : (family OrderedCombineOperation (family RegionTransferIdentity))
constructor DataParallelUpdate

end-family

-- How the owners exchange: a star centred on owner 0, or a bounded-fan-out
-- hierarchy (Representation.ParticipantTree). The topology fixes the
-- combine tree, hence the floating-point order, so it is part of the
-- numerical contract.
family DataParallelTopology : Type 0
constructor DataParallelStar
constructor DataParallelHierarchy
field unrestricted dataParallelFanout : Nat

end-family

-- Where one owner's combine storage and owner state live; the publication's
-- storage lies between them.
family NativeDataParallelLayout : Type 0
constructor NativeDataParallelLayoutOf
field unrestricted dataParallelPlacement : (family NativeCombinePlacement)
field unrestricted dataParallelStateAt : Nat
-- the learner's identity digest, computed once per artifact: every frame
-- and anchor identity of the owner names it
field unrestricted dataParallelDigest : Bytes

end-family

-- The star publication's frames and checkpoint plan.
family NativeDataParallelStarLayout : Type 0
constructor NativeDataParallelStarLayoutOf
field unrestricted dataParallelCaptureFrame : (family RegionFrameLayout)
field unrestricted dataParallelCommittedFrame : (family RegionFrameLayout)
field unrestricted dataParallelCheckpoint : (family NativeRegionCheckpoint)

end-family

def dpPlacement =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (eliminate
      NativeDataParallelLayout
      (lambda unrestricted current : (family NativeDataParallelLayout) .
        (family NativeCombinePlacement))
      layout
      (branch NativeDataParallelLayoutOf placement stateAt digest . placement)))

def dpStateAt =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (eliminate
      NativeDataParallelLayout
      (lambda unrestricted current : (family NativeDataParallelLayout) . Nat)
      layout
      (branch NativeDataParallelLayoutOf placement stateAt digest . stateAt)))

def dpDigest =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (eliminate
      NativeDataParallelLayout
      (lambda unrestricted current : (family NativeDataParallelLayout) . Bytes)
      layout
      (branch NativeDataParallelLayoutOf placement stateAt digest . digest)))

def dpCapture =
  (lambda unrestricted star : (family NativeDataParallelStarLayout) .
    (eliminate
      NativeDataParallelStarLayout
      (lambda unrestricted current : (family NativeDataParallelStarLayout) .
        (family RegionFrameLayout))
      star
      (branch NativeDataParallelStarLayoutOf capture committed checkpoint . capture)))

def dpCommitted =
  (lambda unrestricted star : (family NativeDataParallelStarLayout) .
    (eliminate
      NativeDataParallelStarLayout
      (lambda unrestricted current : (family NativeDataParallelStarLayout) .
        (family RegionFrameLayout))
      star
      (branch NativeDataParallelStarLayoutOf capture committed checkpoint . committed)))

def dpCheckpoint =
  (lambda unrestricted star : (family NativeDataParallelStarLayout) .
    (eliminate
      NativeDataParallelStarLayout
      (lambda unrestricted current : (family NativeDataParallelStarLayout) .
        (family NativeRegionCheckpoint))
      star
      (branch NativeDataParallelStarLayoutOf capture committed checkpoint . checkpoint)))

def dpWord : Nat =
  nativePointwiseWordBytes

def dpEnd =
  (constructor NativePhysicalCommands NativePhysicalCommandsEnd)

def dpHalt =
  (nativeLaunchRecipeCommand
    (constructor NativePhysicalOperation NativePhysicalHaltSuccess)
    b"data-parallel-done"
    dpEnd)

-- ---- shapes ----
def dpK =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (dataParallelParameters c u system))))

def dpS =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (dataParallelWidth c u system))))

def dpOwners =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (dataParallelOwners c u system))))

-- A combined value: k numerator words followed by the mass word.
def dpValueBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalMultiply dpWord (succ (dpK c u system))))))

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

-- m, the optimizer state's words; they follow the parameters in the owner
-- state, so the replicated bytes are one interval: parameters, then state.
def dpM =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (dataParallelStateCount c u system))))

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

def dpReplicatedBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalAdd (dpParameterBytes c u system) (dpOptimizerBytes c u system)))))

def dpStateBytes =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalAdd nativeRegionCaptureBytes (dpReplicatedBytes c u system)))))

-- ---- numerical contract and identities ----
def dpIdentityDigest =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (sha256RawDigestOrEmpty (dataParallelIdentity c u system)))))

-- Elementwise binary64 numerator and mass addition in the combine tree's
-- order; the tree digest itself enters every transfer identity.
def dpRelation =
  (lambda unrestricted topology : (family DataParallelTopology) .
    (lambda unrestricted owners : Nat .
      (eliminate
        DataParallelTopology
        (lambda unrestricted current : (family DataParallelTopology) . Bytes)
        topology
        (branch
          DataParallelStar
          .
          (bytes-append b"data-parallel-left-chain-sum;mass-add;owners=" (regionFrameWord owners)))
        (branch
          DataParallelHierarchy
          fanout
          .
          (bytes-append
            b"data-parallel-hierarchy-sum;mass-add;owners="
            (bytes-append
              (regionFrameWord owners)
              (bytes-append b";fanout=" (regionFrameWord fanout))))))))

def dpContract =
  (lambda unrestricted topology : (family DataParallelTopology) .
    (lambda unrestricted owners : Nat .
      (constructor
        NumericalContract
        NumericalContractOf
        nativeFiniteFloat64Profile
        (constructor
          StdList
          StdListCons
          (family StdNumericOperation)
          stdOpAdd
          (constructor
            StdList
            StdListCons
            (family StdNumericOperation)
            stdOpAdd
            (constructor StdList StdListEmpty (family StdNumericOperation))))
        (constructor NumericalRelation NumericalExactOrder (dpRelation topology owners)))))

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

def dpMembers =
  (lambda unrestricted owners : Nat .
    (constructor CombineCoverage CombineOrderedMembers (stdListNaturalsBelow owners)))

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

def dpTree =
  (lambda unrestricted topology : (family DataParallelTopology) .
    (lambda unrestricted owners : Nat .
      (eliminate
        DataParallelTopology
        (lambda unrestricted current : (family DataParallelTopology) . (family OrderedCombineTree))
        topology
        (branch DataParallelStar . (nativeCombineChain 0 owners))
        (branch DataParallelHierarchy fanout . (nativeCombineHierarchy fanout owners)))))

-- Capture and release frames of the publication star, one per edge.
def dpPublicationIdentity =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted operation : Nat .
            (lambda unrestricted sender : Nat .
              (lambda unrestricted receiver : Nat .
                (constructor
                  RegionTransferIdentity
                  RegionTransferIdentityOf
                  digest
                  b"data-parallel-publication-v1"
                  b"parameters"
                  b"finite-f64-rne-separate"
                  (constructor LearningRound LearningRoundOf 1 0)
                  operation
                  sender
                  receiver
                  (constructor LogicalInterval LogicalIntervalOf 0 (dpReplicatedBytes c u system))))))))))

def dpCaptureFor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted layout : (family NativeDataParallelStarLayout) .
            (lambda unrestricted owner : Nat .
              (niFrameAt
                (dpPublicationIdentity c u system digest 0 owner 0)
                (dpStateBytes c u system)
                (dpCapture layout))))))))

def dpCommittedFor =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (lambda unrestricted layout : (family NativeDataParallelStarLayout) .
            (lambda unrestricted owner : Nat .
              (niFrameAt
                (dpPublicationIdentity c u system digest 1 0 owner)
                dpWord
                (dpCommitted layout))))))))

def dpContext =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted digest : Bytes .
          (constructor
            CheckpointEnvelopeContract
            CheckpointEnvelopeContractValue
            (bytes-append
              b"data-parallel-replicated-f64-finite-rne-separate-v1;"
              (bytes-append
                (numericalExactContractDigest
                  (dpContract
                    (constructor DataParallelTopology DataParallelStar)
                    (dpOwners c u system)))
                digest))
            b"typed-sample-contributions;ordered-combine;typed-update"
            b"immutable-partitioned-samples;one-pass-per-update"
            b"declared-initial-parameters;no-randomness"
            1
            1
            1)))))

def dpRegions =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (constructor
          StdList
          StdListCons
          (family CheckpointLogicalRegion)
          (constructor
            CheckpointLogicalRegion
            CheckpointLogicalRegionOf
            b"parameters"
            b"f64-le"
            (dpParameterBytes c u system))
          (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"optimizer-state"
                    b"f64-le"
                    (dpOptimizerBytes c u system))
                  (constructor StdList StdListEmpty (family CheckpointLogicalRegion)))))
            (dpM c u system))))))

def dpPartitions =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (constructor
          StdList
          StdListCons
          (family CheckpointRegionPartition)
          (constructor
            CheckpointRegionPartition
            CheckpointRegionPartitionOf
            0
            (constructor
              StdList
              StdListCons
              (family LogicalInterval)
              (constructor LogicalInterval LogicalIntervalOf 0 (dpParameterBytes c u system))
              (constructor StdList StdListEmpty (family LogicalInterval))))
          (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
                    1
                    (constructor
                      StdList
                      StdListCons
                      (family LogicalInterval)
                      (constructor
                        LogicalInterval
                        LogicalIntervalOf
                        0
                        (dpOptimizerBytes c u system))
                      (constructor StdList StdListEmpty (family LogicalInterval))))
                  (constructor StdList StdListEmpty (family CheckpointRegionPartition)))))
            (dpM c u system))))))

-- Where a shard's bytes lie in an owner's state: past the capture prefix,
-- parameters (region 0) then optimizer state (region 1).
def dpShardOffset =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted shard : (family NativeRegionShard) .
          (naturalAdd
            nativeRegionCaptureBytes
            (naturalAdd
              (naturalMultiply (nativeRegionOrdinal shard) (dpParameterBytes c u system))
              (intervalBegin (nativeRegionInterval shard))))))))

-- The replicated regions' shard count: parameters, and optimizer state when
-- the learner has any.
def dpShardCount =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (naturalAdd 1 (naturalNonzero (dpM c u system))))))

-- argv with h shards: 1 previous manifest, 2..h+1 previous shards (parameters,
-- then optimizer state), then each shard's candidate and committed file, the
-- candidate and committed manifest and the directory (with one shard: 3/4
-- shard, 5/6 manifest, 7 directory).
def dpPaths =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (constructor NativeRegionPaths NativeRegionPathsOf 1 (dpShardCount c u system)))))

-- ---- layout ----
def dpLeafAt =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (nativeCombineRegionEnd (dpPlacement layout)))

def dpParametersAt =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (naturalAdd (dpStateAt layout) nativeRegionCaptureBytes))

def dpContributionAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (naturalAdd (dpParametersAt layout) (dpReplicatedBytes c u system))))))

def dpContributionEnd =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (nativePointwiseExtent
            c
            (dataParallelContributionProgram c u system)
            (dpContributionAt c u system layout))))))

def dpUpdateAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (dpContributionEnd c u system layout)))))

def dpUpdateEnd =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (nativePointwiseExtent
            u
            (dataParallelUpdateProgram c u system)
            (dpUpdateAt c u system layout))))))

-- The binary64 constant 1, the mass of one sample.
def dpOneAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (dpUpdateEnd c u system layout)))))

def dpSamplesAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (naturalAdd (dpOneAt c u system layout) dpWord)))))

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

def dpEndpointAt =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted owner : Nat .
            (naturalAdd
              (dpSamplesAt c u system layout)
              (naturalMultiply
                (dpSampleBytes c u system)
                (stdListLength (family StdList StdU64) (dataParallelSamplesOf c u system owner)))))))))

-- This role's combine placement under `topology`, from address 0.
def dpPlacementFor =
  (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 .
              (constructor
                NativeCombinePlacement
                NativeCombinePlacementOf
                (nativeCombineContextFor
                  (dpAnchor c u system digest topology)
                  role
                  (dpTree topology (dpOwners c u system)))
                0
                (dpValueBytes c u system))))))))

-- Where a publication's storage may begin: after the combine region and this
-- owner's leaf value.
def dpPublicationBase =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted placement : (family NativeCombinePlacement) .
          (frameAlign (naturalAdd (nativeCombineRegionEnd placement) (dpValueBytes c u system)))))))

-- The star: capture and release frames and the region checkpoint after the
-- combine storage, then the owner state.
def dpWithStarLayout =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted role : Nat .
          (lambda unrestricted epoch : Nat .
            (lambda unrestricted available : Nat .
              (lambda unrestricted emit : (pi unrestricted layout : (family NativeDataParallelLayout) . (pi unrestricted star : (family NativeDataParallelStarLayout) . Bytes)) .
                (let unrestricted digest =
                  (dpIdentityDigest c u system)
                  in
                  (let unrestricted placement =
                    (dpPlacementFor
                      c
                      u
                      system
                      digest
                      (constructor DataParallelTopology DataParallelStar)
                      role)
                    in
                    (let unrestricted frames =
                      (dpPublicationBase c u system placement)
                      in
                      (eliminate
                        StdResult
                        (lambda unrestricted result : (family StdResult (family RegionFrameLayoutFailure) (family RegionFrameLayout)) .
                          Bytes)
                        (planRegionFrameLayout
                          (dpPublicationIdentity c u system digest 0 1 0)
                          (dpStateBytes c u system)
                          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
                                  (dpContext c u system digest)
                                  (dpRegions c u system)
                                  (dpPartitions c u system)
                                  epoch
                                  (frameAlign (frameEnd committed))
                                  available
                                  (dpShardCount c u system))
                                (branch StdFailure error . b"")
                                (branch
                                  StdSuccess
                                  checkpoint
                                  .
                                  (emit
                                    (constructor
                                      NativeDataParallelLayout
                                      NativeDataParallelLayoutOf
                                      placement
                                      (nativeRegionEnd checkpoint)
                                      digest)
                                    (constructor
                                      NativeDataParallelStarLayout
                                      NativeDataParallelStarLayoutOf
                                      capture
                                      committed
                                      checkpoint))))))))))))))))))

-- ---- local work ----
def dpAdd =
  (nativeFiniteFloat64Binary
    b"data-parallel-add"
    nativePeerScratchSlot
    (constructor NativePhysicalFloat64Operation NativePhysicalFloat64Add))

def dpDivide =
  (nativeFiniteFloat64Binary
    b"data-parallel-mean"
    nativePeerScratchSlot
    (constructor NativePhysicalFloat64Operation NativePhysicalFloat64Divide))

def dpCopy =
  (lambda unrestricted destination : Nat .
    (lambda unrestricted source : Nat .
      (lambda unrestricted bytes : Nat .
        (nativeMemoryCopyOperands
          (nativeMemoryState destination)
          (nativeMemoryState source)
          bytes
          nativePeerScratchSlot
          b"data-parallel-copy"))))

-- `step index` for index in 0..count-1, in order.
def dpEach =
  (lambda unrestricted count : Nat .
    (lambda unrestricted step : (pi unrestricted index : Nat . (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands))) .
      (lambda unrestricted tail : (family NativePhysicalCommands) .
        (stdListFold
          Nat
          (family NativePhysicalCommands)
          (lambda unrestricted index : Nat .
            (lambda unrestricted rest : (family NativePhysicalCommands) . (step index rest)))
          tail
          (stdListNaturalsBelow count)))))

-- Combine arithmetic: out_i = left_i + right_i for every numerator word and
-- the mass, each separately rounded.
def dpApply =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted out : Nat .
          (lambda unrestricted left : Nat .
            (lambda unrestricted right : Nat .
              (dpEach
                (succ (dpK c u system))
                (lambda unrestricted index : Nat .
                  (dpAdd
                    (nativePointwiseAt out index)
                    (nativePointwiseAt left index)
                    (nativePointwiseAt right index))))))))))

-- This owner's samples into its (numerator, mass) leaf value, in order,
-- reading the parameters at `parameters`.
def dpContribute =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted parameters : Nat .
            (lambda unrestricted owner : Nat .
              (lambda unrestricted contribution : (pi unrestricted tail : (family NativePhysicalCommands) . (family NativePhysicalCommands)) .
                (lambda unrestricted tail : (family NativePhysicalCommands) .
                  (let unrestricted k =
                    (dpK c u system)
                    in
                    (let unrestricted s =
                      (dpS c u system)
                      in
                      (let unrestricted leaf =
                        (dpLeafAt layout)
                        in
                        (let unrestricted at =
                          (dpContributionAt c u system layout)
                          in
                          (dpEach
                            (succ k)
                            (lambda unrestricted index : Nat .
                              (nativeMemoryStoreWord64
                                b"data-parallel-zero"
                                (nativePointwiseAt leaf index)
                                (nativeLaunchRecipeImmediate 0)))
                            (dpEach
                              (stdListLength
                                (family StdList StdU64)
                                (dataParallelSamplesOf c u system owner))
                              (lambda unrestricted sample : Nat .
                                (lambda unrestricted rest : (family NativePhysicalCommands) .
                                  (dpCopy
                                    at
                                    parameters
                                    (dpParameterBytes c u system)
                                    (dpCopy
                                      (nativePointwiseAt at k)
                                      (naturalAdd
                                        (dpSamplesAt c u system layout)
                                        (naturalMultiply sample (dpSampleBytes c u system)))
                                      (dpSampleBytes c u system)
                                      (contribution
                                        (dpEach
                                        k
                                        (lambda unrestricted index : Nat .
                                        (dpAdd
                                        (nativePointwiseAt leaf index)
                                        (nativePointwiseAt leaf index)
                                        (nativePointwiseAt at (naturalAdd (naturalAdd k s) index))))
                                        (dpAdd
                                        (nativePointwiseAt leaf k)
                                        (nativePointwiseAt leaf k)
                                        (dpOneAt c u system layout)
                                        rest)))))))
                              tail))))))))))))))

-- mean = numerator / mass, then the update program on (parameters, state,
-- mean, rate), then the logical position advances. Every owner runs this on
-- the replicated root value.
def dpUpdate =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (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 total =
                    (nativeCombineResultAt (dpPlacement layout))
                    in
                    (let unrestricted at =
                      (dpUpdateAt c u system layout)
                      in
                      (dpEach
                        k
                        (lambda unrestricted index : Nat .
                          (dpDivide
                            (nativePointwiseAt at (naturalAdd (naturalAdd k m) index))
                            (nativePointwiseAt total index)
                            (nativePointwiseAt total k)))
                        (dpCopy
                          at
                          (dpParametersAt layout)
                          (dpReplicatedBytes c u system)
                          (nativeMemoryStoreWord64
                            b"data-parallel-rate"
                            (nativePointwiseAt at (naturalAdd (naturalMultiply 2 k) m))
                            (nativeLaunchRecipeImmediate
                              (modelWord64Natural (dataParallelRateWord c u system)))
                            (update
                              (dpCopy
                                (dpParametersAt layout)
                                at
                                (dpReplicatedBytes c u system)
                                (niAdvanceAt (dpStateAt layout) tail))))))))))))))))

def dpLocal =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (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 DataParallelRoundOperation) .
                  (eliminate
                    DataParallelRoundOperation
                    (lambda unrestricted current : (family DataParallelRoundOperation) .
                      (pi unrestricted tail : (family NativePhysicalCommands) .
                        (family NativePhysicalCommands)))
                    operation
                    (branch
                      DataParallelContribute
                      .
                      (dpContribute c u system layout (dpParametersAt layout) owner contribution))
                    (branch
                      DataParallelCombine
                      value
                      .
                      (nativeCombineLocal
                        (dpPlacement layout)
                        (dpLeafAt layout)
                        (dpApply c u system)
                        value))
                    (branch DataParallelUpdate . (dpUpdate c u system layout update)))))))))))

-- ---- the round protocol ----
def dpEachRole =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted operation : (family DataParallelRoundOperation) .
      (lambda unrestricted tail : (family GlobalProtocol (family DataParallelRoundOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
        (stdListFold
          Nat
          (family
            GlobalProtocol
            (family DataParallelRoundOperation)
            (family RegionTransferIdentity)
            (family ProtocolNoDecision))
          (lambda unrestricted role : Nat .
            (lambda unrestricted rest : (family GlobalProtocol (family DataParallelRoundOperation) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
              (protocolLocal
                (family DataParallelRoundOperation)
                (family RegionTransferIdentity)
                (family ProtocolNoDecision)
                role
                operation
                rest)))
          tail
          (stdListNaturalsBelow owners)))))

def dpRoundEnd =
  (constructor
    GlobalProtocol
    GlobalProtocolEnd
    (family DataParallelRoundOperation)
    (family RegionTransferIdentity)
    (family ProtocolNoDecision))

def dpRound =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted combine : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
      (dpEachRole
        owners
        (constructor DataParallelRoundOperation DataParallelContribute)
        (appendGlobalProtocol
          (family DataParallelRoundOperation)
          (family RegionTransferIdentity)
          (family ProtocolNoDecision)
          (mapGlobalProtocolOperations
            (family OrderedCombineOperation (family RegionTransferIdentity))
            (family DataParallelRoundOperation)
            (family RegionTransferIdentity)
            (family ProtocolNoDecision)
            (lambda unrestricted value : (family OrderedCombineOperation (family RegionTransferIdentity)) .
              (constructor DataParallelRoundOperation DataParallelCombine value))
            combine)
          (dpEachRole owners (constructor DataParallelRoundOperation DataParallelUpdate) dpRoundEnd)))))

def dpLocalAdmitted =
  (lambda unrestricted owners : Nat .
    (lambda unrestricted placement : (family NativeCombinePlacement) .
      (lambda unrestricted role : Nat .
        (lambda unrestricted operation : (family DataParallelRoundOperation) .
          (eliminate
            DataParallelRoundOperation
            (lambda unrestricted current : (family DataParallelRoundOperation) . Nat)
            operation
            (branch DataParallelContribute . (naturalLess role owners))
            (branch DataParallelCombine value . (nativeCombineLocalAdmitted placement role value))
            (branch DataParallelUpdate . (naturalLess role owners)))))))

-- The projected round program, repeated `steps` times, for this role.
def dpProjected =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted topology : (family DataParallelTopology) .
          (lambda unrestricted layout : (family NativeDataParallelLayout) .
            (lambda unrestricted steps : Nat .
              (let unrestricted owners =
                (dpOwners c u system)
                in
                (let unrestricted placement =
                  (dpPlacement layout)
                  in
                  (stdResultBind
                    (family OrderedCombineFailure)
                    (family
                      GlobalProtocol
                      (family OrderedCombineOperation (family RegionTransferIdentity))
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))
                    (family
                      ParticipantProtocol
                      (family DataParallelRoundOperation)
                      (family RegionTransferIdentity)
                      (family ProtocolNoDecision))
                    (planNativeCombineGlobal
                      placement
                      (dpContract topology owners)
                      owners
                      (dpMembers owners)
                      (dpBaseAnchor c u system (dpDigest layout)))
                    (lambda unrestricted combine : (family GlobalProtocol (family OrderedCombineOperation (family RegionTransferIdentity)) (family RegionTransferIdentity) (family ProtocolNoDecision)) .
                      (eliminate
                        StdResult
                        (lambda unrestricted result : (family StdResult (family ProtocolFailure) (family ParticipantProtocol (family DataParallelRoundOperation) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                          (family
                            StdResult
                            (family OrderedCombineFailure)
                            (family
                              ParticipantProtocol
                              (family DataParallelRoundOperation)
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision))))
                        (planParticipantProtocol
                          (family DataParallelRoundOperation)
                          (family RegionTransferIdentity)
                          (family ProtocolNoDecision)
                          owners
                          (nativeCombineRole (nativeCombinePlacementContext placement))
                          (dpLocalAdmitted owners placement)
                          (nativeCombineTransferAdmitted placement)
                          (protocolNoDecision (family RegionTransferIdentity))
                          (protocolRepeat
                            (family DataParallelRoundOperation)
                            (family RegionTransferIdentity)
                            (family ProtocolNoDecision)
                            steps
                            (dpRound owners combine)
                            dpRoundEnd))
                        (branch
                          StdFailure
                          failure
                          .
                          (constructor
                            StdResult
                            StdFailure
                            (family OrderedCombineFailure)
                            (family
                              ParticipantProtocol
                              (family DataParallelRoundOperation)
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision))
                            (constructor OrderedCombineFailure CombineProtocolRefused failure)))
                        (branch
                          StdSuccess
                          projected
                          .
                          (constructor
                            StdResult
                            StdSuccess
                            (family OrderedCombineFailure)
                            (family
                              ParticipantProtocol
                              (family DataParallelRoundOperation)
                              (family RegionTransferIdentity)
                              (family ProtocolNoDecision))
                            projected)))))))))))))

-- ---- start, restore, publication ----
def dpStoreWords =
  (lambda unrestricted at : Nat .
    (lambda unrestricted words : (family StdList StdU64) .
      (lambda unrestricted tail : (family NativePhysicalCommands) .
        (stdListFold
          StdU64
          (pi unrestricted index : Nat . (family NativePhysicalCommands))
          (lambda unrestricted word : StdU64 .
            (lambda unrestricted next : (pi unrestricted index : Nat . (family NativePhysicalCommands)) .
              (lambda unrestricted index : Nat .
                (nativeMemoryStoreWord64
                  b"data-parallel-constant"
                  (nativePointwiseAt at index)
                  (nativeLaunchRecipeImmediate (modelWord64Natural word))
                  (next (succ index))))))
          (lambda unrestricted index : Nat . tail)
          words
          0))))

-- The owner's samples, in order, and the constant 1.0 (sample mass).
def dpInitialize =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted owner : Nat .
            (lambda unrestricted tail : (family NativePhysicalCommands) .
              (nativeFloat64Natural
                b"data-parallel-one"
                (dpOneAt c u system layout)
                1
                (dpStoreWords
                  (dpSamplesAt c u system layout)
                  (stdListFold
                    (family StdList StdU64)
                    (family StdList StdU64)
                    (lambda unrestricted sample : (family StdList StdU64) .
                      (lambda unrestricted rest : (family StdList StdU64) .
                        (stdListAppend StdU64 sample rest)))
                    (constructor StdList StdListEmpty StdU64)
                    (dataParallelSamplesOf c u system owner))
                  tail))))))))

-- Fresh: version and cursor 0, the plan's epoch and the declared initial
-- parameters and optimizer state.
def dpFresh =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted epoch : Nat .
            (lambda unrestricted tail : (family NativePhysicalCommands) .
              (nativeMemoryStoreWord64
                b"data-parallel-version"
                (dpStateAt layout)
                (nativeLaunchRecipeImmediate 0)
                (nativeMemoryStoreWord64
                  b"data-parallel-cursor"
                  (nativePointwiseAt (dpStateAt layout) 1)
                  (nativeLaunchRecipeImmediate 0)
                  (nativeMemoryStoreWord64
                    b"data-parallel-epoch"
                    (nativePointwiseAt (dpStateAt layout) 2)
                    (nativeLaunchRecipeImmediate epoch)
                    (dpStoreWords
                      (dpParametersAt layout)
                      (dataParallelInitialParameters c u system)
                      (dpStoreWords
                        (naturalAdd (dpParametersAt layout) (dpParameterBytes c u system))
                        (dataParallelInitialStateWords c u system)
                        tail)))))))))))

-- After a verified restore of the parameters: the restored (version, cursor,
-- epoch) from `capture`, then epoch + 1 (a new membership epoch).
def dpResumed =
  (lambda unrestricted layout : (family NativeDataParallelLayout) .
    (lambda unrestricted capture : Nat .
      (lambda unrestricted tail : (family NativePhysicalCommands) .
        (nativeMemoryCopyOperands
          (nativeMemoryState (dpStateAt layout))
          (nativeMemoryState capture)
          nativeRegionCaptureBytes
          nativePeerScratchSlot
          b"data-parallel-restore-capture"
          (nativeLaunchRecipeAssertEqual
            (nativeMemoryLoad (dpStateAt layout))
            (nativeMemoryLoad (nativePointwiseAt (dpStateAt layout) 1))
            b"data-parallel-input-position"
            (nativeCheckedWordAdd
              (nativeMemoryState (nativePointwiseAt (dpStateAt layout) 2))
              (nativeMemoryLoad (nativePointwiseAt (dpStateAt layout) 2))
              (nativeLaunchRecipeImmediate 1)
              nativePeerScratchSlot
              b"data-parallel-epoch-overflow"
              tail))))))

-- The replicated regions' addresses in this owner: parameters, then (when
-- the learner has any) optimizer state.
def dpDestinations =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (constructor
            StdList
            StdListCons
            (family NativeRegionDestination)
            (constructor
              NativeRegionDestination
              NativeRegionDestinationOf
              0
              (constructor LogicalInterval LogicalIntervalOf 0 (dpParameterBytes c u system))
              (dpParametersAt layout))
            (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
                      1
                      (constructor
                        LogicalInterval
                        LogicalIntervalOf
                        0
                        (dpOptimizerBytes c u system))
                      (naturalAdd (dpParametersAt layout) (dpParameterBytes c u system)))
                    (constructor StdList StdListEmpty (family NativeRegionDestination)))))
              (dpM c u system)))))))

def dpRestore =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted star : (family NativeDataParallelStarLayout) .
            (lambda unrestricted extent : Nat .
              (lambda unrestricted tail : (family NativePhysicalCommands) .
                (planNativeRegionRestore
                  (dpCheckpoint star)
                  (nativeRegionPathArgument (nativeRegionPathFirst (dpPaths c u system)))
                  (lambda unrestricted index : Nat .
                    (nativeRegionPathArgument
                      (naturalAdd (nativeRegionInputShardBase (dpPaths c u system)) index)))
                  nativePeerListenerSlot
                  nativePeerScratchSlot
                  extent
                  (dpDestinations c u system layout)
                  (dpResumed layout (nativeRegionCaptureAt (dpCheckpoint star)) tail)))))))))

def dpStart =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted star : (family NativeDataParallelStarLayout) .
            (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)
                      (dpFresh c u system layout (nativeRegionEpoch (dpCheckpoint star)) tail))
                    (lambda unrestricted previous : Nat .
                      (lambda unrestricted ignored : (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)) .
                        (dpRestore c u system layout star extent tail)))
                    resume)))))))))

def dpPublication =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (lambda unrestricted star : (family NativeDataParallelStarLayout) .
            (lambda unrestricted resume : Nat .
              (lambda unrestricted edges : (family StdList (family NativePeerEdge)) .
                (constructor
                  NativePeerRegionPublication
                  NativePeerRegionPublicationOf
                  (constructor
                    NativeRegionCheckpoint
                    NativeRegionCheckpointOf
                    (nativeRegionManifest (dpCheckpoint star))
                    (nativeRegionShards (dpCheckpoint star))
                    (naturalAdd (nativeRegionEpoch (dpCheckpoint star)) resume)
                    (nativeRegionEnd (dpCheckpoint star)))
                  (naturalPredecessor (dpOwners c u system))
                  (dpCaptureFor c u system (dpDigest layout) star)
                  (dpCommittedFor c u system (dpDigest layout) star)
                  (nativePeerEdgeConnectionTo edges)
                  (dpStateAt layout)
                  (dpStateBytes c u system)
                  (dpStateBytes c u system)
                  (lambda unrestricted shard : (family NativeRegionShard) . 0)
                  (dpShardOffset c u system)
                  (dpPaths c u system)))))))))

def dpOutput =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted layout : (family NativeDataParallelLayout) .
          (niOutput (dpStateAt layout) (dpStateBytes c u system) dpHalt)))))

def dpRunAdmitted =
  (lambda unrestricted c : Nat .
    (lambda unrestricted u : Nat .
      (lambda unrestricted system : (family DataParallelSystem c u) .
        (lambda unrestricted role : Nat .
          (lambda unrestricted addresses : (family StdList Bytes) .
            (lambda unrestricted resume : Nat .
              (lambda unrestricted epoch : Nat .
                (lambda unrestricted steps : Nat .
                  (naturalAnd
                    (dataParallelAdmitted c u system)
                    (naturalAnd
                      (nativePeerStarAdmitted (dpOwners c u system) role addresses)
                      (naturalAnd
                        (naturalLessOrEqual resume 1)
                        (naturalAnd
                          (naturalNonzero epoch)
                          (naturalAnd
                            (naturalBelowTwoPower 64 (naturalAdd epoch resume))
                            (naturalAnd (naturalNonzero steps) (naturalBelowTwoPower 64 steps)))))))))))))))

-- Compile both pointwise programs, 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`.
def dpEmitWith =
  (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 topology : (family DataParallelTopology) .
              (lambda unrestricted layout : (family NativeDataParallelLayout) .
                (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 extent =
                                (nativePeerEdgesEnd (dpEndpointAt c u system layout role) edges)
                                in
                                (let unrestricted placement =
                                  (dpPlacement layout)
                                  in
                                  (eliminate
                                    StdResult
                                    (lambda unrestricted result : (family StdResult Bytes (family NativePhysicalCommands)) .
                                      Bytes)
                                    (planNativePointwiseProgram
                                      c
                                      (dataParallelContributionProgram c u system)
                                      (dpContributionAt c u system layout)
                                      (dpContributionEnd c u system layout)
                                      nativePeerScratchSlot
                                      dpEnd)
                                    (branch StdFailure error . b"")
                                    (branch
                                      StdSuccess
                                      contribution
                                      .
                                      (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult Bytes (family NativePhysicalCommands)) .
                                        Bytes)
                                        (planNativePointwiseProgram
                                        u
                                        (dataParallelUpdateProgram c u system)
                                        (dpUpdateAt c u system layout)
                                        (dpUpdateEnd c u system layout)
                                        nativePeerScratchSlot
                                        dpEnd)
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        update
                                        .
                                        (eliminate
                                        StdResult
                                        (lambda unrestricted result : (family StdResult (family OrderedCombineFailure) (family ParticipantProtocol (family DataParallelRoundOperation) (family RegionTransferIdentity) (family ProtocolNoDecision))) .
                                        Bytes)
                                        (dpProjected c u system 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)))
                                        (dpOutput c u system 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
                                        (dpParametersAt layout)
                                        (dpK c u system)
                                        (nativePeerEdgeSetup
                                        role
                                        edges
                                        (dpEndpointAt c u system layout role)
                                        (nativeProtocolFrames
                                        (family DataParallelRoundOperation)
                                        (family RegionTransferIdentity)
                                        (family ProtocolNoDecision)
                                        (lambda unrestricted direction : (family ProtocolDirection) .
                                        (lambda unrestricted identity : (family RegionTransferIdentity) .
                                        (lambda unrestricted value : (family RegionTransferIdentity) .
                                        (nativeFrameInitialize (nativeCombineFrame placement value)))))
                                        projected
                                        (nativeProtocolCommands
                                        (family DataParallelRoundOperation)
                                        (family RegionTransferIdentity)
                                        (family ProtocolNoDecision)
                                        (dpLocal
                                        c
                                        u
                                        system
                                        layout
                                        role
                                        (nativePhysicalProgramComposeCommandsAppend contribution)
                                        (nativePhysicalProgramComposeCommandsAppend update))
                                        (nativeCombineTransfer
                                        placement
                                        (lambda unrestricted peer : Nat .
                                        (nativeLaunchRecipeSlotValue
                                        (nativePeerEdgeConnectionTo edges peer))))
                                        (nativeCombineDecisions placement)
                                        b"data-parallel"
                                        projected
                                        published)))))
                                        (branch StdFailure error . b"")
                                        (branch
                                        StdSuccess
                                        started
                                        .
                                        (nativeHostTargetResourceArtifact
                                        target
                                        resources
                                        (niProgramFor
                                        (nativePeerEdgeSlots edges)
                                        extent
                                        (dpInitialize c u system layout role started)))))))
                                        (naturalAnd
                                        (naturalLessOrEqual extent available)
                                        (arenaCertificate
                                        extent
                                        (nativeCombineResidents
                                        placement
                                        (constructor
                                        ArenaResidents
                                        ArenaResidentsNext
                                        (constructor
                                        ArenaResident
                                        ArenaResidentValue
                                        b"data-parallel-leaf"
                                        (dpLeafAt layout)
                                        (dpValueBytes c u system)
                                        dpWord)
                                        (constructor
                                        ArenaResidents
                                        ArenaResidentsNext
                                        (constructor
                                        ArenaResident
                                        ArenaResidentValue
                                        b"data-parallel-owner-state"
                                        (dpStateAt layout)
                                        (naturalSaturatingSubtract extent (dpStateAt layout))
                                        dpWord)
                                        residents))))))))))))))))))))))))))))))

def dpEmit =
  (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 layout : (family NativeDataParallelLayout) .
              (lambda unrestricted star : (family NativeDataParallelStarLayout) .
                (lambda unrestricted role : Nat .
                  (lambda unrestricted edges : (family StdList (family NativePeerEdge)) .
                    (lambda unrestricted resume : Nat .
                      (lambda unrestricted steps : Nat .
                        (lambda unrestricted available : Nat .
                          (dpEmitWith
                            target
                            resources
                            c
                            u
                            system
                            (constructor DataParallelTopology DataParallelStar)
                            layout
                            role
                            edges
                            steps
                            available
                            (lambda unrestricted extent : Nat .
                              (lambda unrestricted tail : (family NativePhysicalCommands) .
                                (dpStart c u system layout star extent resume tail)))
                            (lambda unrestricted extent : Nat .
                              (lambda unrestricted success : (family NativePhysicalCommands) .
                                (planNativePeerRegionPublish
                                  (dpPublication c u system layout star resume edges)
                                  extent
                                  role
                                  success)))
                            (constructor ArenaResidents ArenaResidentsEnd))))))))))))))

def nativeDataParallelArtifact =
  (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 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 .
                                (dpWithStarLayout
                                  c
                                  u
                                  system
                                  role
                                  epoch
                                  available
                                  (lambda unrestricted layout : (family NativeDataParallelLayout) .
                                    (lambda unrestricted star : (family NativeDataParallelStarLayout) .
                                      (dpEmit
                                        target
                                        resources
                                        c
                                        u
                                        system
                                        layout
                                        star
                                        role
                                        edges
                                        resume
                                        steps
                                        available))))))
                            (dpRunAdmitted c u system role addresses resume epoch steps))))))))))))))