mmt.go

package lit

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"slices"
	"strings"
	"time"

	"github.com/lightning-ai/sdk/go/internal/lightningcloud/openapi/generated/client/jobs_service"
	"github.com/lightning-ai/sdk/go/internal/lightningcloud/openapi/generated/models"
	"github.com/lightning-ai/sdk/go/internal/sdkclient"
)

// MMT represents a Lightning multi-machine job.
type MMT struct {
	id          string
	name        string
	teamspaceID string
	teamspace   string
	ownerName   string
	status      string
	numMachines int64
	machine     string
	command     string
	image       string
	studioID    string
	totalCost   float32
	startedAt   time.Time
	stoppedAt   time.Time

	maxRunAttempts    int64
	currentRunAttempt int64
	keepMinutes       int64
	tags              []string
}

// MachineDict is the JSON-friendly public representation of one MMT machine.
type MachineDict struct {
	// Name is the machine job name.
	Name string `json:"name"`
	// Status is the latest known machine job state.
	Status string `json:"status"`
	// Machine is the assigned machine type.
	Machine string `json:"machine"`
}

// MMTDict is the JSON-friendly public representation of an MMT.
type MMTDict struct {
	// Name is the multi-machine job display name.
	Name string `json:"name"`
	// Teamspace is the owning teamspace name.
	Teamspace string `json:"teamspace"`
	// Studio is the source studio name when the MMT was launched from a studio.
	Studio *string `json:"studio"`
	// Image is the container image used by the MMT.
	Image *string `json:"image"`
	// Command is the shell command run by each machine.
	Command string `json:"command"`
	// Status is the latest known MMT state.
	Status string `json:"status"`
	// Machine is the machine type assigned to each MMT worker.
	Machine string `json:"machine"`
	// Machines lists the worker jobs that belong to the MMT.
	Machines []MachineDict `json:"machines"`
	// TotalCost is the accumulated MMT cost reported by Lightning.
	TotalCost float32 `json:"total_cost"`
}

// MMTPathMapping maps a teamspace connection path into an MMT container.
type MMTPathMapping struct {
	// ContainerPath is the mount path inside each worker container.
	ContainerPath string
	// ConnectionName is the teamspace connection to mount.
	ConnectionName string
	// ConnectionPath is the path within the teamspace connection.
	ConnectionPath string
}

type mmtOptions struct {
	id               string
	teamspaceID      string
	teamspaceName    string
	ownerName        string
	studioID         string
	status           string
	numMachines      int64
	machine          string
	command          string
	image            string
	totalCost        float32
	env              map[string]string
	cloud            string
	interruptible    bool
	imageCredentials string
	cloudAccountAuth bool
	entrypoint       *string
	pathMappings     []MMTPathMapping
	artifactsSource  string
	artifactsDest    string
	maxRuntime       int
	maxRunAttempts   int64
	keepMinutes      int64
}

type mmtWaitOptions struct {
	interval      time.Duration
	timeout       time.Duration
	stopOnTimeout bool
}

// MMTOptions configures MMT lookup and creation.
type MMTOptions struct {
	// ID resolves an existing MMT directly without looking it up by name.
	ID string
	// Teamspace scopes MMT lookup or creation to a specific teamspace.
	Teamspace *Teamspace
	// Studio launches the MMT from a specific studio environment.
	Studio *Studio
	// Status seeds the latest known state for ID-based handles.
	Status string
	// NumMachines is the number of worker machines to start.
	NumMachines int64
	// Machine selects the compute type used for each worker.
	Machine Machine
	// Command is the command executed by each worker.
	Command string
	// Image is the container image used when creating an MMT.
	Image string
	// TotalCost seeds the latest known cost for ID-based handles.
	TotalCost float32
	// Env sets environment variables for each worker.
	Env map[string]string
	// Cloud selects a cloud account or provider for the MMT.
	Cloud string
	// Interruptible overrides whether workers run on interruptible capacity.
	Interruptible *bool
	// ImageCredentials names credentials used to pull the container image.
	ImageCredentials string
	// CloudAccountAuth enables cloud account authentication for workers.
	CloudAccountAuth bool
	// Entrypoint overrides the container entrypoint.
	Entrypoint string
	// PathMappings mounts teamspace connection paths into each worker container.
	PathMappings []MMTPathMapping
	// ArtifactsSource is the local path copied from workers after completion.
	ArtifactsSource string
	// ArtifactsDestination is the remote teamspace connection destination.
	ArtifactsDestination string
	// MaxRuntime limits the MMT runtime in seconds.
	MaxRuntime int
	// MaxRunAttempts is the max number of run attempts for this multi-machine
	// job. 0 means unset (legacy, no retries); 1 means a single attempt (no
	// retries). When greater than 1 the RECREATE_ALL_NODES fault tolerance
	// strategy is set automatically on the multi-machine job body. Set at the
	// multi-machine job level, not on the per-machine JobSpec.
	MaxRunAttempts int64
	// KeepMinutes is minutes to keep the machines after the job stops.
	// 0 means the machines are released immediately.
	KeepMinutes int64
}

// MMTWaitOptions configures polling behavior while waiting for an MMT.
type MMTWaitOptions struct {
	// Interval controls how often MMT status is polled.
	Interval time.Duration
	// Timeout limits how long Wait polls before returning.
	Timeout time.Duration
	// StopOnTimeout stops the remote MMT when the timeout expires.
	StopOnTimeout bool
}

// ID returns the MMT ID.
func (m *MMT) ID() string {
	if m == nil {
		return ""
	}
	return m.id
}

// TeamspaceID returns the owning teamspace ID.
func (m *MMT) TeamspaceID() string {
	if m == nil {
		return ""
	}
	return m.teamspaceID
}

// StudioID returns the source studio ID when the MMT uses a studio.
func (m *MMT) StudioID() string {
	if m == nil {
		return ""
	}
	return m.studioID
}

// Name returns the MMT name.
func (m *MMT) Name() string {
	if m == nil {
		return ""
	}
	return m.name
}

// Teamspace returns the owning teamspace name.
func (m *MMT) Teamspace() string {
	if m == nil {
		return ""
	}
	return m.teamspace
}

// OwnerName returns the owning user or organization name.
func (m *MMT) OwnerName() string {
	if m == nil {
		return ""
	}
	return m.ownerName
}

// Status returns the latest known MMT state.
func (m *MMT) Status() string {
	if m == nil {
		return ""
	}
	return m.status
}

// NumMachines returns the requested number of machines.
func (m *MMT) NumMachines() int64 {
	if m == nil {
		return 0
	}
	return m.numMachines
}

// Machine returns the configured machine.
func (m *MMT) Machine() string {
	if m == nil {
		return ""
	}
	return m.machine
}

// Command returns the command run by the MMT.
func (m *MMT) Command() string {
	if m == nil {
		return ""
	}
	return m.command
}

// Image returns the image used by the MMT.
func (m *MMT) Image() string {
	if m == nil {
		return ""
	}
	return m.image
}

// TotalCost returns the latest known total cost.
func (m *MMT) TotalCost() float32 {
	if m == nil {
		return 0
	}
	return m.totalCost
}

// StartedAt returns when the MMT's machines started running, or the zero time
// if they haven't started yet.
func (m *MMT) StartedAt() time.Time {
	if m == nil {
		return time.Time{}
	}
	return m.startedAt
}

// StoppedAt returns when the MMT stopped running, or the zero time if it
// hasn't stopped yet.
func (m *MMT) StoppedAt() time.Time {
	if m == nil {
		return time.Time{}
	}
	return m.stoppedAt
}

// MaxRunAttempts returns the max number of run attempts for this multi-machine
// job, read from the multi-machine job body. 0 means unset.
func (m *MMT) MaxRunAttempts() int64 {
	if m == nil {
		return 0
	}
	return m.maxRunAttempts
}

// CurrentRunAttempt returns the current run attempt for this multi-machine job,
// read from the multi-machine job body. 0 means unset.
func (m *MMT) CurrentRunAttempt() int64 {
	if m == nil {
		return 0
	}
	return m.currentRunAttempt
}

// KeepMinutes returns minutes the machines are kept after the job stops.
// 0 means unset (the machines are released immediately).
func (m *MMT) KeepMinutes() int64 {
	if m == nil {
		return 0
	}
	return m.keepMinutes
}

// Tags returns the teamspace tags applied to this multi-machine job, in the
// order the platform returns them.
func (m *MMT) Tags() []string {
	return slices.Clone(m.tags)
}

// GetMMT returns an existing MMT by name or ID.
func GetMMT(name string, opts ...MMTOptions) (*MMT, error) {
	resolved := applyMMTOptions(opts...)
	if name == "" && resolved.id == "" {
		return nil, errors.New("mmt requires name or ID")
	}
	if resolved.id != "" {
		return &MMT{
			id:          resolved.id,
			name:        name,
			teamspaceID: resolved.teamspaceID,
			teamspace:   resolved.teamspaceName,
			ownerName:   resolved.ownerName,
			status:      resolved.status,
			numMachines: resolved.numMachines,
			machine:     resolved.machine,
			command:     resolved.command,
			image:       resolved.image,
			studioID:    resolved.studioID,
			totalCost:   resolved.totalCost,
		}, nil
	}
	api, err := sdkclient.New()
	if err != nil {
		return nil, err
	}
	if resolved.teamspaceID == "" {
		return nil, errors.New("mmt requires teamspace")
	}
	resp, err := api.JobsService.JobsServiceGetMultiMachineJobByName(
		jobs_service.NewJobsServiceGetMultiMachineJobByNameParamsWithContext(context.Background()).
			WithName(name).
			WithProjectID(resolved.teamspaceID),
	)
	if err != nil {
		return nil, err
	}
	return mmtFromModel(resp.Payload, resolved), nil
}

// RunMMT creates and starts a new MMT.
func RunMMT(name string, numMachines int64, machine Machine, command string, opts ...MMTOptions) (*MMT, error) {
	resolved := applyMMTOptions(opts...)
	if name == "" {
		return nil, errors.New("mmt run requires name")
	}
	if numMachines <= 1 {
		return nil, errors.New("mmt run cannot use less than 2 machines")
	}
	if err := validateMMTComputeEnvironment(command, resolved); err != nil {
		return nil, err
	}
	if err := validateKeepMinutes(resolved.keepMinutes); err != nil {
		return nil, err
	}
	api, err := sdkclient.New()
	if err != nil {
		return nil, err
	}
	if resolved.teamspaceID == "" {
		return nil, errors.New("mmt run requires teamspace or studio")
	}
	// Fault tolerance is auto-derived: when the job is multi-machine and allows
	// retries (max_run_attempts > 1) the only currently supported strategy
	// (RECREATE_ALL_NODES) is set automatically; otherwise the field is left
	// unset.
	var faultTolerance *models.V1MultiMachineJobFaultTolerance
	if numMachines > 1 && resolved.maxRunAttempts > 1 {
		s := models.V1MultiMachineJobFaultToleranceStrategyMULTIMACHINEJOBFAULTTOLERANCESTRATEGYRECREATEALLNODES
		faultTolerance = &models.V1MultiMachineJobFaultTolerance{Strategy: &s}
	}
	body := &models.JobsServiceCreateMultiMachineJobBody{
		ClusterID:      resolved.cloud,
		Machines:       numMachines,
		Name:           name,
		Spec:           mmtJobSpec(numMachines, string(machine), command, resolved),
		MaxRunAttempts: resolved.maxRunAttempts,
		FaultTolerance: faultTolerance,
	}
	resp, err := api.JobsService.JobsServiceCreateMultiMachineJob(
		jobs_service.NewJobsServiceCreateMultiMachineJobParamsWithContext(context.Background()).WithProjectID(resolved.teamspaceID).WithBody(body),
	)
	if err != nil {
		return nil, err
	}
	return mmtFromModel(resp.Payload, resolved), nil
}

// Stop stops the MMT.
func (m *MMT) Stop() error {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return errors.New("mmt stop requires teamspace ID and MMT ID")
	}
	api, err := sdkclient.New()
	if err != nil {
		return err
	}
	desiredState := models.V1MultiMachineJobStateMultiMachineJobSTATESTOP
	resp, err := api.JobsService.JobsServiceUpdateMultiMachineJob(
		jobs_service.NewJobsServiceUpdateMultiMachineJobParamsWithContext(context.Background()).
			WithProjectID(m.teamspaceID).
			WithID(m.id).
			WithBody(&models.JobsServiceUpdateMultiMachineJobBody{DesiredState: &desiredState}),
	)
	if err != nil {
		return err
	}
	updated := mmtFromModel(resp.Payload, mmtOptions{teamspaceName: m.teamspace, ownerName: m.ownerName})
	if updated != nil {
		*m = *updated
	}
	return nil
}

// SetTags replaces the MMT's tag set with the given tag names. Tags that don't
// exist in the teamspace yet are created. Pass zero names to remove all tags.
func (m *MMT) SetTags(tagNames ...string) error {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return errors.New("mmt set tags requires teamspace ID and MMT ID")
	}
	return setWorkloadTags(m.teamspaceID, "multi_machine_job", m.id, tagNames)
}

// Delete deletes the MMT.
func (m *MMT) Delete() error {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return errors.New("mmt delete requires teamspace ID and MMT ID")
	}
	api, err := sdkclient.New()
	if err != nil {
		return err
	}
	_, err = api.JobsService.JobsServiceDeleteMultiMachineJob(
		jobs_service.NewJobsServiceDeleteMultiMachineJobParamsWithContext(context.Background()).
			WithProjectID(m.teamspaceID).
			WithID(m.id),
	)
	return err
}

// Wait polls until the MMT reaches a terminal state.
func (m *MMT) Wait(opts ...MMTWaitOptions) error {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return errors.New("mmt wait requires teamspace ID and MMT ID")
	}
	resolved := applyMMTWaitOptions(opts...)
	api, err := sdkclient.New()
	if err != nil {
		return err
	}
	start := time.Now()
	for {
		resp, err := api.JobsService.JobsServiceGetMultiMachineJob(
			jobs_service.NewJobsServiceGetMultiMachineJobParamsWithContext(context.Background()).
				WithProjectID(m.teamspaceID).
				WithID(m.id),
		)
		if err != nil {
			return err
		}
		updated := mmtFromModel(resp.Payload, mmtOptions{teamspaceName: m.teamspace, ownerName: m.ownerName})
		if updated == nil {
			return fmt.Errorf("get mmt returned empty payload")
		}
		*m = *updated
		if isMMTTerminalStatus(m.status) {
			return nil
		}
		if resolved.timeout > 0 && time.Since(start) > resolved.timeout {
			if resolved.stopOnTimeout {
				_ = m.Stop()
			}
			return fmt.Errorf("mmt didn't finish within the provided timeout")
		}
		time.Sleep(resolved.interval)
	}
}

// Machines lists the jobs backing the MMT.
func (m *MMT) Machines() ([]*Job, error) {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return nil, errors.New("mmt machines requires teamspace ID and MMT ID")
	}
	api, err := sdkclient.New()
	if err != nil {
		return nil, err
	}
	var result []*Job
	var pageToken string
	for {
		params := jobs_service.NewJobsServiceListJobsParamsWithContext(context.Background()).
			WithProjectID(m.teamspaceID).
			WithMultiMachineJobID(&m.id)
		if pageToken != "" {
			params = params.WithPageToken(&pageToken)
		}
		resp, err := api.JobsService.JobsServiceListJobs(params)
		if err != nil {
			return nil, err
		}
		if resp.Payload == nil {
			return nil, fmt.Errorf("list mmt machines returned empty payload")
		}
		for _, model := range resp.Payload.Jobs {
			if model == nil {
				continue
			}
			result = append(result, m.jobFromModel(model))
		}
		if resp.Payload.NextPageToken == "" {
			return result, nil
		}
		pageToken = resp.Payload.NextPageToken
	}
}

// Logs returns the merged logs of every machine in the MMT, each line prefixed
// with the bracketed name of the machine it came from.
func (m *MMT) Logs() (string, error) {
	if m == nil || m.teamspaceID == "" || m.id == "" {
		return "", errors.New("mmt logs requires teamspace ID and MMT ID")
	}
	machines, err := m.Machines()
	if err != nil {
		return "", err
	}
	names := make(map[string]string, len(machines))
	for _, machine := range machines {
		if machine != nil {
			names[machine.ID()] = machine.Name()
		}
	}
	lines, err := readLogs(m.teamspaceID, func(params *jobs_service.JobsServiceGetLogsParams) {
		params.SetMmtID(&m.id)
	}, names)
	if err != nil {
		return "", err
	}
	if len(lines) == 0 && isJobTerminalStatus(m.status) {
		return "", errLogsMigrating
	}
	return strings.Join(lines, "\n"), nil
}

// Dict returns a JSON-friendly public representation of the MMT.
func (m *MMT) Dict() (MMTDict, error) {
	if m == nil {
		return MMTDict{}, nil
	}
	machines, err := m.Machines()
	if err != nil {
		return MMTDict{}, err
	}
	machineDicts := make([]MachineDict, 0, len(machines))
	for _, machine := range machines {
		if machine == nil {
			continue
		}
		machineDicts = append(machineDicts, MachineDict{
			Name:    machine.Name(),
			Status:  machine.Status(),
			Machine: machine.Machine(),
		})
	}
	var studioName *string
	if m.studioID != "" {
		studioName = &m.studioID
	}
	var image *string
	if m.image != "" {
		image = &m.image
	}
	return MMTDict{
		Name:      m.name,
		Teamspace: ownerTeamspace(m.ownerName, m.teamspace),
		Studio:    studioName,
		Image:     image,
		Command:   m.command,
		Status:    m.status,
		Machine:   m.machine,
		Machines:  machineDicts,
		TotalCost: m.totalCost,
	}, nil
}

// JSON returns an indented JSON representation of the MMT.
func (m *MMT) JSON() (string, error) {
	dict, err := m.Dict()
	if err != nil {
		return "", err
	}
	body, err := json.MarshalIndent(dict, "", "    ")
	if err != nil {
		return "", err
	}
	return string(body), nil
}

// ArtifactsURI is always empty for an MMT: each machine writes its own
// artifacts, so there is no single address for the run. Use Machines and each
// machine's ArtifactsURI.
func (m *MMT) ArtifactsURI() string {
	return ""
}

// ListArtifacts lists what every machine of this MMT wrote to the teamspace
// drive. Each path is prefixed with the name of the machine that wrote it, so
// the listing matches the layout DownloadArtifacts writes.
func (m *MMT) ListArtifacts(path string, recursive bool) ([]FileEntry, error) {
	machines, err := m.Machines()
	if err != nil {
		return nil, err
	}
	var entries []FileEntry
	for _, machine := range machines {
		machineEntries, err := machine.ListArtifacts(path, recursive)
		if err != nil {
			return nil, err
		}
		for _, entry := range machineEntries {
			entry.Path = machine.Name() + "/" + entry.Path
			entries = append(entries, entry)
		}
	}
	return entries, nil
}

// DownloadArtifacts downloads what every machine of this MMT wrote, each into
// its own folder named after the machine under targetDir.
func (m *MMT) DownloadArtifacts(targetDir, path string) error {
	machines, err := m.Machines()
	if err != nil {
		return err
	}
	for _, machine := range machines {
		if err := machine.DownloadArtifacts(filepath.Join(targetDir, machine.Name()), path); err != nil {
			return err
		}
	}
	return nil
}

// Link returns the Lightning web URL for the MMT.
func (m *MMT) Link() string {
	if m == nil || m.ownerName == "" || m.teamspace == "" || m.name == "" {
		return ""
	}
	return fmt.Sprintf("https://platform.lightning.ai/%s/%s/jobs/%s?app_id=mmt", m.ownerName, m.teamspace, m.name)
}

func applyMMTOptions(opts ...MMTOptions) mmtOptions {
	var resolved mmtOptions
	if len(opts) > 0 {
		resolved.id = opts[0].ID
		if opts[0].Teamspace != nil {
			resolved.teamspaceID = opts[0].Teamspace.ID()
			resolved.teamspaceName = opts[0].Teamspace.Name()
			if resolved.cloud == "" {
				resolved.cloud = opts[0].Teamspace.DefaultCloudAccount()
			}
			if owner := opts[0].Teamspace.Owner(); owner != nil {
				resolved.ownerName = owner.Name()
			}
		}
		if opts[0].Studio != nil {
			resolved.studioID = opts[0].Studio.ID()
			resolved.cloud = firstNonEmpty(opts[0].Cloud, opts[0].Studio.Cloud(), resolved.cloud)
			if resolved.teamspaceID == "" {
				resolved.teamspaceID = opts[0].Studio.TeamspaceID()
				resolved.teamspaceName = opts[0].Studio.Teamspace()
				resolved.ownerName = opts[0].Studio.OwnerName()
			}
		}
		resolved.status = opts[0].Status
		resolved.numMachines = opts[0].NumMachines
		resolved.machine = string(opts[0].Machine)
		resolved.command = opts[0].Command
		resolved.image = opts[0].Image
		resolved.totalCost = opts[0].TotalCost
		resolved.env = opts[0].Env
		resolved.cloud = firstNonEmpty(opts[0].Cloud, resolved.cloud)
		if opts[0].Interruptible != nil {
			resolved.interruptible = *opts[0].Interruptible
		}
		resolved.imageCredentials = opts[0].ImageCredentials
		resolved.cloudAccountAuth = opts[0].CloudAccountAuth
		if opts[0].Entrypoint != "" {
			resolved.entrypoint = &opts[0].Entrypoint
		}
		resolved.pathMappings = opts[0].PathMappings
		resolved.artifactsSource = opts[0].ArtifactsSource
		resolved.artifactsDest = opts[0].ArtifactsDestination
		resolved.maxRuntime = opts[0].MaxRuntime
		resolved.maxRunAttempts = opts[0].MaxRunAttempts
		resolved.keepMinutes = opts[0].KeepMinutes
	}
	return resolved
}

func applyMMTWaitOptions(opts ...MMTWaitOptions) mmtWaitOptions {
	resolved := mmtWaitOptions{interval: 5 * time.Second}
	if len(opts) > 0 {
		resolved.interval = opts[0].Interval
		resolved.timeout = opts[0].Timeout
		resolved.stopOnTimeout = opts[0].StopOnTimeout
		if resolved.interval == 0 {
			resolved.interval = 5 * time.Second
		}
	}
	if resolved.interval < 0 {
		resolved.interval = 0
	}
	return resolved
}

func validateMMTComputeEnvironment(command string, opts mmtOptions) error {
	if opts.studioID != "" {
		if opts.image != "" {
			return errors.New("image and studio are mutually exclusive as both define the environment to run the job in")
		}
		if command == "" {
			return errors.New("command is required when using a studio")
		}
		if opts.artifactsSource != "" || opts.artifactsDest != "" {
			return errors.New("artifacts are only supported when using a custom image")
		}
		return nil
	}
	if opts.image == "" {
		return errors.New("either image or studio must be provided")
	}
	if err := validateArtifacts(opts.artifactsSource, opts.artifactsDest); err != nil {
		return err
	}
	return nil
}

func mmtFromModel(model *models.V1MultiMachineJob, opts mmtOptions) *MMT {
	if model == nil {
		return nil
	}
	result := &MMT{
		id:          model.ID,
		name:        model.Name,
		teamspaceID: model.ProjectID,
		teamspace:   opts.teamspaceName,
		ownerName:   opts.ownerName,
		numMachines: model.Machines,
		studioID:    model.CloudspaceID,
		totalCost:   model.TotalCost,

		maxRunAttempts:    model.MaxRunAttempts,
		currentRunAttempt: model.CurrentRunAttempt,
		tags:              tagNames(model.Tags),
	}
	if model.State != nil {
		result.status = string(*model.State)
	}
	if model.Status != nil {
		result.startedAt = time.Time(model.Status.StartedAt)
		result.stoppedAt = time.Time(model.Status.StoppedAt)
	}
	if model.Spec != nil {
		result.machine = model.Spec.InstanceName
		result.command = model.Spec.Command
		result.image = model.Spec.Image
		if result.studioID == "" {
			result.studioID = model.Spec.CloudspaceID
		}
		result.keepMinutes = model.Spec.ReserveMachinesTimeoutMinutes
	}
	return result
}

// isMMTTerminalStatus reports whether an observed MMT state is terminal. The
// API reports either the bare state or the MultiMachineJob_STATE_* enum name,
// so both spellings are matched. Mirrors the backend's terminal set:
// completed, stopped, failed, deleted.
func isMMTTerminalStatus(status string) bool {
	switch strings.ToLower(status) {
	case "completed", "stopped", "failed", "deleted",
		"multimachinejob_state_completed", "multimachinejob_state_stopped", "multimachinejob_state_failed", "multimachinejob_state_deleted":
		return true
	default:
		return false
	}
}

func (m *MMT) jobFromModel(model *models.V1Job) *Job {
	if model == nil {
		return nil
	}
	opts := jobOptions{
		id:            model.ID,
		teamspaceID:   model.ProjectID,
		teamspaceName: m.teamspace,
		ownerName:     m.ownerName,
		mmtID:         model.MultiMachineJobID,
		status:        model.State,
		publicIP:      model.PublicIPAddress,
		totalCost:     model.TotalCost,
	}
	if model.Spec != nil {
		opts.studioID = model.Spec.CloudspaceID
		opts.machine = model.Spec.InstanceName
		opts.command = model.Spec.Command
		opts.image = model.Spec.Image
	}
	return jobFromModel(model, opts)
}

func mmtJobSpec(numMachines int64, machine, command string, opts mmtOptions) *models.V1JobSpec {
	return &models.V1JobSpec{
		CloudspaceID:                  opts.studioID,
		ClusterID:                     opts.cloud,
		Command:                       command,
		Entrypoint:                    resolveEntrypoint(command, opts.entrypoint, opts.image),
		Env:                           envVars(opts.env),
		Image:                         opts.image,
		ImageClusterCredentials:       opts.cloudAccountAuth,
		ImageSecretRef:                opts.imageCredentials,
		InstanceName:                  machine,
		PathMappings:                  mmtPathMappings(opts),
		Quantity:                      numMachines,
		RequestedRunDurationSeconds:   maxRuntime(opts.maxRuntime),
		KeepMachineAfterStop:          opts.keepMinutes > 0,
		ReserveMachinesTimeoutMinutes: opts.keepMinutes,
		Spot:                          opts.interruptible,
	}
}

func mmtPathMappings(opts mmtOptions) []*models.V1PathMapping {
	if len(opts.pathMappings) == 0 && opts.artifactsSource == "" {
		return nil
	}
	result := make([]*models.V1PathMapping, 0, len(opts.pathMappings)+1)
	for _, mapping := range opts.pathMappings {
		result = append(result, &models.V1PathMapping{
			ConnectionName: mapping.ConnectionName,
			ConnectionPath: mapping.ConnectionPath,
			ContainerPath:  mapping.ContainerPath,
		})
	}
	if opts.artifactsSource != "" {
		connectionName, connectionPath := artifactConnection(opts.artifactsDest)
		result = append(result, &models.V1PathMapping{
			ConnectionName: connectionName,
			ConnectionPath: connectionPath,
			ContainerPath:  opts.artifactsSource,
		})
	}
	return result
}