Repository: bacalhau-project/bacalhau (language: python)

Issue:
model.Execution and models.Execution have different definitions of terminal state
The model Execution considers the following states to be terminal:
- `ExecutionStateAskForBidRejected`
- `ExecutionStateBidRejected`
- `ExecutionStateCancelled`
- `ExecutionStateFailed`
- `ExecutionStateCompleted`

https://github.com/bacalhau-project/bacalhau/blob/17dbd8dac2fc47485b2a752f40d0d414078e6fe3/pkg/model/execution_state.go#L77-L79
 https://github.com/bacalhau-project/bacalhau/blob/17dbd8dac2fc47485b2a752f40d0d414078e6fe3/pkg/model/execution_state.go#L60-L63

The model Execution considers the following states to be terminal:
- `ExecutionStateBidRejected`
- `ExecutionStateCompleted`
- `ExecutionStateFailed`
- `ExecutionStateCancelled`

https://github.com/bacalhau-project/bacalhau/blob/17dbd8dac2fc47485b2a752f40d0d414078e6fe3/pkg/models/execution.go#L45-L50

The different being `ExecutionStateAskForBidRejected`. @wdbaruni is this a bug?
Remove Requester API test suites



Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index f3ef02c1e6..53b6c035bb 100644
--- a/Makefile
+++ b/Makefile
@@ -339,7 +339,6 @@ integration-test:
 
 .PHONY: bash-test
 bash-test: 
-	export REQUESTER_ENDPOINT_USE_DEPRECATED_ENV=true
 	${BINARY_PATH}
 	cd test && bin/bashtub *.sh
 
diff --git a/cmd/cli/version/version.go b/cmd/cli/version/version.go
index 7c873575af..137a20fc2e 100644
--- a/cmd/cli/version/version.go
+++ b/cmd/cli/version/version.go
@@ -25,7 +25,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/version"
 
 	"github.com/bacalhau-project/bacalhau/cmd/util"
@@ -59,7 +59,7 @@ func NewCmd() *cobra.Command {
 				return fmt.Errorf("failed to setup repo: %w", err)
 			}
 			// create an api client
-			api, err := util.GetAPIClient(cfg)
+			api, err := util.GetAPIClientV2(cmd, cfg)
 			if err != nil {
 				return fmt.Errorf("failed to create api client: %w", err)
 			}
@@ -72,7 +72,7 @@ func NewCmd() *cobra.Command {
 	return versionCmd
 }
 
-func runVersion(cmd *cobra.Command, cfg types.BacalhauConfig, api *client.APIClient, oV *VersionOptions) error {
+func runVersion(cmd *cobra.Command, cfg types.BacalhauConfig, api clientv2.API, oV *VersionOptions) error {
 	ctx := cmd.Context()
 
 	err := oV.Run(ctx, cmd, cfg, api)
@@ -103,7 +103,7 @@ var updateMessageColumn = output.TableColumn[util.Versions]{
 	Value:        func(v util.Versions) string { return v.UpdateMessage },
 }
 
-func (oV *VersionOptions) Run(ctx context.Context, cmd *cobra.Command, cfg types.BacalhauConfig, api *client.APIClient) error {
+func (oV *VersionOptions) Run(ctx context.Context, cmd *cobra.Command, cfg types.BacalhauConfig, api clientv2.API) error {
 	var (
 		versions util.Versions
 		columns  []output.TableColumn[util.Versions]
diff --git a/cmd/util/api.go b/cmd/util/api.go
index 90eed608cf..39384e2de1 100644
--- a/cmd/util/api.go
+++ b/cmd/util/api.go
@@ -11,44 +11,10 @@ import (
 	"github.com/bacalhau-project/bacalhau/cmd/util/auth"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/version"
 )
 
-func GetAPIClient(cfg types.BacalhauConfig) (*client.APIClient, error) {
-	tlsCfg := cfg.Node.ClientAPI.ClientTLS
-	apiHost := cfg.Node.ClientAPI.Host
-	apiPort := cfg.Node.ClientAPI.Port
-	tokenPath := cfg.Auth.TokensPath
-
-	if tlsCfg.CACert != "" {
-		if _, err := os.Stat(tlsCfg.CACert); os.IsNotExist(err) {
-			return nil, fmt.Errorf("CA certificate file %q does not exists", tlsCfg.CACert)
-		} else if err != nil {
-			return nil, fmt.Errorf("CA certificate file %q cannot be read: %w", tlsCfg.CACert, err)
-		}
-	}
-
-	apiClient, err := client.NewAPIClient(client.LegacyTLSSupport(tlsCfg), cfg.User, apiHost, uint16(apiPort))
-	if err != nil {
-		return nil, err
-	}
-
-	apiSheme := "http"
-	if tlsCfg.UseTLS {
-		apiSheme = "https"
-	}
-
-	if token, err := ReadToken(tokenPath, fmt.Sprintf("%s://%s:%d", apiSheme, apiHost, apiPort)); err != nil {
-		log.Warn().Err(err).Msg("Failed to read access tokens – API calls will be without authorization")
-	} else if token != nil {
-		apiClient.DefaultHeaders["Authorization"] = token.String()
-	}
-
-	return apiClient, nil
-}
-
 func GetAPIClientV2(cmd *cobra.Command, cfg types.BacalhauConfig) (clientv2.API, error) {
 	tlsCfg := cfg.Node.ClientAPI.ClientTLS
 	apiHost := cfg.Node.ClientAPI.Host
diff --git a/cmd/util/execute.go b/cmd/util/execute.go
deleted file mode 100644
index eeca56b4d2..0000000000
--- a/cmd/util/execute.go
+++ /dev/null
@@ -1,22 +0,0 @@
-package util
-
-import (
-	"context"
-
-	"github.com/rs/zerolog/log"
-
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
-)
-
-//nolint:funlen,gocyclo // Refactor later
-func ExecuteJob(ctx context.Context, j *model.Job, api client.APIClient) (*model.Job, error) {
-	err := legacy_job.VerifyJob(ctx, j)
-	if err != nil {
-		log.Ctx(ctx).Err(err).Msg("Job failed to validate.")
-		return nil, err
-	}
-
-	return api.Submit(ctx, j)
-}
diff --git a/cmd/util/printer/print_legacy.go b/cmd/util/printer/print_legacy.go
deleted file mode 100644
index abc1faa791..0000000000
--- a/cmd/util/printer/print_legacy.go
+++ /dev/null
@@ -1,358 +0,0 @@
-package printer
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"os/signal"
-	"strings"
-	"time"
-
-	"github.com/pkg/errors"
-	"github.com/rs/zerolog/log"
-	"github.com/spf13/cobra"
-	"golang.org/x/exp/slices"
-
-	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	"github.com/bacalhau-project/bacalhau/pkg/lib/math"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
-	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
-
-	"github.com/bacalhau-project/bacalhau/cmd/util"
-	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
-	"github.com/bacalhau-project/bacalhau/pkg/bacerrors"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-)
-
-var eventsWorthPrintingLegacy = map[model.JobStateType]eventStruct{
-	model.JobStateNew:        {Message: "Creating job for submission", IsTerminal: false, IsError: false},
-	model.JobStateQueued:     {Message: "Job waiting to be scheduled", IsTerminal: false, IsError: false},
-	model.JobStateInProgress: {Message: "Job in progress", IsTerminal: false, IsError: false},
-	model.JobStateError:      {Message: "Error while executing the job", IsTerminal: true, IsError: true},
-	model.JobStateCancelled:  {Message: "Job canceled", IsTerminal: true, IsError: false},
-	model.JobStateCompleted:  {Message: "Job finished", IsTerminal: true, IsError: false},
-}
-
-// PrintJobExecutionLegacy displays information about the execution of a job
-// TODO gocyclo rates this as a 24, which is very high, we should refactor someday.
-//
-//nolint:gocyclo,funlen
-func PrintJobExecutionLegacy(
-	ctx context.Context,
-	j *model.Job,
-	cmd *cobra.Command,
-	downloadSettings *cliflags.DownloaderSettings,
-	runtimeSettings *cliflags.RunTimeSettingsWithDownload,
-	apiLegacy *client.APIClient,
-	apiV2 clientv2.API,
-	cfg types.BacalhauConfig,
-) error {
-	// if we are in --wait=false - print the id then exit
-	// because all code after this point is related to
-	// "wait for the job to finish" (via WaitForJobAndPrintResultsToUser)
-	if !runtimeSettings.WaitForJobToFinish {
-		cmd.Print(j.Metadata.ID + "\n")
-		return nil
-	}
-
-	// if we are in --id-only mode - print the id
-	if runtimeSettings.PrintJobIDOnly {
-		cmd.Print(j.Metadata.ID + "\n")
-	}
-
-	if runtimeSettings.Follow {
-		cmd.Printf("Job successfully submitted. Job ID: %s\n", j.Metadata.ID)
-		cmd.Printf("Waiting for logs... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
-
-		// Wait until the job has actually been accepted and started, otherwise this will fail waiting for
-		// the execution to appear.
-		for i := 0; i < 10; i++ {
-			jobState, stateErr := apiLegacy.GetJobState(ctx, j.ID())
-			if stateErr != nil {
-				return fmt.Errorf("failed waiting for execution to start: %w", stateErr)
-			}
-
-			executionID := ""
-			for _, execution := range jobState.Executions {
-				if execution.State.IsActive() {
-					executionID = execution.ComputeReference
-				}
-			}
-
-			if executionID != "" {
-				break
-			}
-			time.Sleep(time.Duration(1) * time.Second)
-		}
-
-		return util.Logs(cmd, apiV2, util.LogOptions{
-			JobID:  j.ID(),
-			Follow: true,
-		})
-	}
-
-	// if we are only printing the id, set the rest of the output to "quiet",
-	// i.e. don't print
-	quiet := runtimeSettings.PrintJobIDOnly
-
-	jobErr := WaitForJobAndPrintResultsToUser(ctx, cmd, apiLegacy, j, quiet)
-	if jobErr != nil {
-		if jobErr.Error() == PrintoutCanceledButRunningNormally {
-			return nil
-		} else {
-			cmd.PrintErrf("\nError submitting job: %s", jobErr)
-		}
-	}
-
-	jobReturn, found, err := apiLegacy.Get(ctx, j.Metadata.ID)
-	if err != nil {
-		return fmt.Errorf("error getting job: %w", err)
-	}
-	if !found {
-		return fmt.Errorf("weird. Just ran the job, but we couldn't find it. Should be impossible. ID: %s", j.Metadata.ID)
-	}
-
-	js, err := apiLegacy.GetJobState(ctx, jobReturn.Job.Metadata.ID)
-	if err != nil {
-		return fmt.Errorf("error getting job state: %w", err)
-	}
-
-	if runtimeSettings.PrintNodeDetails || jobErr != nil {
-		cmd.Println("\nJob Results By Node:")
-		for message, nodes := range summariseExecutionsLegacy(js) {
-			cmd.Printf("• Node %s: ", strings.Join(nodes, ", "))
-			if strings.ContainsRune(message, '\n') {
-				cmd.Printf("\n\t%s\n", strings.Join(strings.Split(message, "\n"), "\n\t"))
-			} else {
-				cmd.Println(message)
-			}
-		}
-	}
-
-	hasResults := slices.ContainsFunc(js.Executions, func(e model.ExecutionState) bool { return e.RunOutput != nil })
-	if !quiet && hasResults {
-		cmd.Printf("\nTo download the results, execute:\n\t"+os.Args[0]+" get %s\n", j.ID())
-	}
-
-	if !quiet {
-		cmd.Printf("\nTo get more details about the run, execute:\n\t"+os.Args[0]+" describe %s\n", j.ID())
-	}
-
-	if hasResults && runtimeSettings.AutoDownloadResults {
-		if err := util.DownloadResultsHandler(ctx, cmd, cfg, apiV2, j.Metadata.ID, downloadSettings); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// WaitForJobAndPrintResultsToUser uses events to decide what to output to the terminal
-// using a spinner to show long-running tasks. When the job is complete (or the user
-// triggers SIGINT) then the function will complete and stop outputting to the terminal.
-//
-//nolint:gocyclo,funlen
-func WaitForJobAndPrintResultsToUser(
-	ctx context.Context,
-	cmd *cobra.Command,
-	apiLegacy *client.APIClient,
-	j *model.Job,
-	quiet bool,
-) error {
-	if j == nil || j.Metadata.ID == "" {
-		return errors.New("No job returned from the server.")
-	}
-
-	getMoreInfoString := fmt.Sprintf(`
-To get more information at any time, run:
-   bacalhau describe %s`, j.Metadata.ID)
-
-	cancelString := fmt.Sprintf(`
-To cancel the job, run:
-   bacalhau cancel %s`, j.Metadata.ID)
-
-	if !quiet {
-		cmd.Printf("Job successfully submitted. Job ID: %s\n", j.Metadata.ID)
-		cmd.Printf("Checking job status... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
-	}
-
-	// Create a map of job state types -> boolean, this is a record of what has been printed
-	// so far.
-	printedEventsTracker := make(map[model.JobStateType]bool)
-	for _, jobEventType := range model.JobStateTypes() {
-		printedEventsTracker[jobEventType] = false
-	}
-
-	// Inject "Job Initiated Event" to start - should we do this on the server?
-	// TODO: #1068 Should jobs auto add a "start event" on the client at creation?
-	// Faking an initial time (sometimes it happens too fast to see)
-	// fullLineMessage.TimerString = spinnerFmtDuration(DefaultSpinnerFormatDuration)
-	startMessage := "Communicating with the network"
-
-	// Decide where the spinner should write it's output, by default we want stdout
-	// but we can also disable output here if we have been asked to be quiet.
-	writer := cmd.OutOrStdout()
-	if quiet {
-		writer = io.Discard
-	}
-
-	widestString := len(startMessage)
-	for _, v := range eventsWorthPrintingLegacy {
-		widestString = math.Max(widestString, len(v.Message))
-	}
-
-	spinner, err := NewSpinner(ctx, writer, widestString, false)
-	if err != nil {
-		return err
-	}
-	spinner.Run()
-	spinner.NextStep(startMessage)
-
-	// Capture Ctrl+C if the user wants to finish early the job
-	ctx, cancel := context.WithCancel(ctx)
-	signalChan := make(chan os.Signal, 2)
-	signal.Notify(signalChan, util.ShutdownSignals...)
-	defer func() {
-		signal.Stop(signalChan)
-		cancel()
-	}()
-
-	cmdShuttingDown := false
-	var returnError error = nil
-
-	// goroutine for handling SIGINT from the signal channel, or context
-	// completion messages.
-	go func() {
-		for {
-			select {
-			case s := <-signalChan: // first signal, cancel context
-				log.Ctx(ctx).Debug().Msgf("Captured %v. Exiting...", s)
-				if s == os.Interrupt {
-					cmdShuttingDown = true
-					spinner.Done(StopCancel)
-
-					if !quiet {
-						cmd.Println("\n\n\rPrintout canceled (the job is still running).")
-						cmd.Println(getMoreInfoString)
-						cmd.Println(cancelString)
-					}
-					returnError = fmt.Errorf(PrintoutCanceledButRunningNormally)
-				} else {
-					cmd.Println("Unexpected signal received. Exiting.")
-				}
-				cancel()
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
-
-	var lastSeenTimestamp int64 = 0
-	var lastEventState model.JobStateType
-	for !cmdShuttingDown {
-		// Get the job level history events that happened since the last one we saw
-		jobEvents, err := apiLegacy.GetEvents(ctx, j.Metadata.ID, legacymodels.EventFilterOptions{
-			Since:                 lastSeenTimestamp,
-			ExcludeExecutionLevel: true,
-		})
-		if err != nil {
-			if _, ok := err.(*bacerrors.ContextCanceledError); ok {
-				// We're done, the user canceled the job
-				cmdShuttingDown = true
-				continue
-			} else {
-				return errors.Wrap(err, "Error getting job events")
-			}
-		}
-
-		// Iterate through the events, looking for ones we have not yet processed
-		for _, event := range jobEvents {
-			if event.Time.Unix() > lastSeenTimestamp {
-				lastSeenTimestamp = event.Time.Unix()
-			}
-
-			if event.Type != model.JobHistoryTypeJobLevel {
-				continue
-			}
-
-			if !quiet {
-				jet := event.JobState.New // Get the type of the new state
-				wasPrinted := printedEventsTracker[jet]
-
-				// If it hasn't been printed yet, we'll print this event.
-				// We'll also skip lines where there's no message to print.
-				if !wasPrinted && eventsWorthPrintingLegacy[jet].Message != "" {
-					printedEventsTracker[jet] = true
-
-					// We shouldn't do anything with execution errors because there could
-					// be retries following, so for now we will
-					if !eventsWorthPrintingLegacy[jet].IsError && !eventsWorthPrintingLegacy[jet].IsTerminal {
-						spinner.NextStep(eventsWorthPrintingLegacy[jet].Message)
-					}
-
-					if event.JobState.New == model.JobStateQueued {
-						spinner.msgMutex.Lock()
-						spinner.msg.Waiting = true
-						spinner.msg.Detail = event.Comment
-						spinner.msgMutex.Unlock()
-					}
-				} else if wasPrinted && lastEventState == event.JobState.New {
-					// Printing the same again but with new information, so we
-					// can just update the existing message.
-					spinner.msgMutex.Lock()
-					spinner.msg.Detail = event.Comment
-					spinner.msgMutex.Unlock()
-				}
-			}
-
-			lastEventState = event.JobState.New
-
-			if event.JobState.New.IsTerminal() {
-				if event.JobState.New != model.JobStateCompleted {
-					returnError = errors.New(event.Comment)
-					spinner.Done(StopFailed)
-				} else {
-					spinner.Done(StopSuccess)
-				}
-				cmdShuttingDown = true
-				break
-			}
-		}
-
-		// Have we been cancel(l)ed?
-		if condition := ctx.Err(); condition != nil {
-			break
-		}
-
-		time.Sleep(time.Duration(500) * time.Millisecond) //nolint:gomnd // 500ms sleep
-	}
-
-	return returnError
-}
-
-// Groups the executions in the job state, returning a map of printable messages
-// to node(s) that generated that message.
-func summariseExecutionsLegacy(state model.JobState) map[string][]string {
-	results := make(map[string][]string, len(state.Executions))
-	for _, execution := range state.Executions {
-		var message string
-		if execution.RunOutput != nil {
-			if execution.RunOutput.ErrorMsg != "" {
-				message = execution.RunOutput.ErrorMsg
-			} else if execution.RunOutput.ExitCode > 0 {
-				message = execution.RunOutput.STDERR
-			} else {
-				message = execution.RunOutput.STDOUT
-			}
-		} else if execution.State.IsDiscarded() {
-			message = execution.Status
-		}
-
-		if message != "" {
-			results[message] = append(results[message], idgen.ShortNodeID(execution.NodeID))
-		}
-	}
-	return results
-}
diff --git a/cmd/util/version.go b/cmd/util/version.go
index 9964551f16..9ab431aded 100644
--- a/cmd/util/version.go
+++ b/cmd/util/version.go
@@ -11,7 +11,7 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/config"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/version"
 )
 
@@ -22,14 +22,23 @@ type Versions struct {
 	UpdateMessage string                   `json:"updateMessage,omitempty"`
 }
 
-func GetAllVersions(ctx context.Context, cfg types.BacalhauConfig, api *client.APIClient) (Versions, error) {
+func GetAllVersions(ctx context.Context, cfg types.BacalhauConfig, api clientv2.API) (Versions, error) {
 	var err error
 	versions := Versions{ClientVersion: version.Get()}
 
-	versions.ServerVersion, err = api.Version(ctx)
+	resp, err := api.Agent().Version(ctx)
 	if err != nil {
 		return versions, errors.Wrap(err, "error running version command")
 	}
+	versions.ServerVersion = &models.BuildVersionInfo{
+		Major:      resp.Major,
+		Minor:      resp.Minor,
+		GitVersion: resp.GitVersion,
+		GitCommit:  resp.GitCommit,
+		BuildDate:  resp.BuildDate,
+		GOOS:       resp.GOOS,
+		GOARCH:     resp.GOARCH,
+	}
 
 	clientID, err := config.GetClientID(cfg.User.KeyPath)
 	if err != nil {
diff --git a/pkg/models/execution.go b/pkg/models/execution.go
index 9efcba29a8..e796da8b57 100644
--- a/pkg/models/execution.go
+++ b/pkg/models/execution.go
@@ -46,7 +46,8 @@ func (s ExecutionStateType) IsTermainl() bool {
 	return s == ExecutionStateBidRejected ||
 		s == ExecutionStateCompleted ||
 		s == ExecutionStateFailed ||
-		s == ExecutionStateCancelled
+		s == ExecutionStateCancelled ||
+		s == ExecutionStateAskForBidRejected
 }
 
 type ExecutionDesiredStateType int
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index 38fd14c0dc..626727b031 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -46,14 +46,12 @@ import (
 
 type Requester struct {
 	// Visible for testing
-	Endpoint   requester.Endpoint
-	EndpointV2 *orchestrator.BaseEndpoint
-	JobStore   jobstore.Store
+	Endpoint *orchestrator.BaseEndpoint
+	JobStore jobstore.Store
 	// We need a reference to the node info store until libp2p is removed
 	NodeInfoStore      routing.NodeInfoStore
 	NodeDiscoverer     orchestrator.NodeDiscoverer
 	nodeManager        *manager.NodeManager
-	localCallback      compute.Callback
 	cleanupFunc        func(ctx context.Context)
 	debugInfoProviders []model.DebugInfoProvider
 }
@@ -208,15 +206,6 @@ func NewRequesterNode(
 		resultTransformers = append(resultTransformers, resultSigner)
 	}
 
-	endpoint := requester.NewBaseEndpoint(&requester.BaseEndpointParams{
-		ID:                nodeID,
-		EventEmitter:      eventEmitter,
-		ComputeEndpoint:   computeProxy,
-		Store:             jobStore,
-		DefaultJobTimeout: requesterConfig.JobDefaults.TotalTimeout,
-		DefaultPublisher:  requesterConfig.DefaultPublisher,
-	})
-
 	var translationProvider translation.TranslatorProvider
 	if requesterConfig.TranslationEnabled {
 		translationProvider = translation.NewStandardTranslatorsProvider()
@@ -264,14 +253,8 @@ func NewRequesterNode(
 		discovery.NewDebugInfoProvider(nodeManager),
 	}
 
-	// register requester public http apis
-	requesterAPIServer := requester_endpoint.NewEndpoint(requester_endpoint.EndpointParams{
-		Router:             apiServer.Router,
-		Requester:          endpoint,
-		DebugInfoProviders: debugInfoProviders,
-		JobStore:           jobStore,
-		NodeDiscoverer:     nodeManager,
-	})
+	// TODO: delete this when we are ready to stop serving a deprecation notice.
+	requester_endpoint.NewEndpoint(apiServer.Router)
 
 	orchestrator_endpoint.NewEndpoint(orchestrator_endpoint.EndpointParams{
 		Router:       apiServer.Router,
@@ -305,7 +288,6 @@ func NewRequesterNode(
 		// record the event in a log
 		eventTracer,
 		// dispatches events to listening websockets
-		requesterAPIServer,
 	)
 
 	// A single Cleanup function to make sure the order of closing dependencies is correct
@@ -333,14 +315,20 @@ func NewRequesterNode(
 		}
 	}
 
+	// This endpoint implements the protocol formerly known as `bprotocol`.
+	// It provides the compute call back endpoints for interacting with compute nodes.
+	// e.g. bidding, job completions, cancellations, and failures
+	endpoint := requester.NewBaseEndpoint(&requester.BaseEndpointParams{
+		ID:           nodeID,
+		EventEmitter: eventEmitter,
+		Store:        jobStore,
+	})
 	if err = transportLayer.RegisterComputeCallback(endpoint); err != nil {
 		return nil, err
 	}
 
 	return &Requester{
-		Endpoint:           endpoint,
-		localCallback:      endpoint,
-		EndpointV2:         endpointV2,
+		Endpoint:           endpointV2,
 		NodeDiscoverer:     nodeManager,
 		NodeInfoStore:      nodeManager,
 		JobStore:           jobStore,
diff --git a/pkg/publicapi/apimodels/legacymodels/legacy.go b/pkg/publicapi/apimodels/legacymodels/legacy.go
deleted file mode 100644
index d87e23170e..0000000000
--- a/pkg/publicapi/apimodels/legacymodels/legacy.go
+++ /dev/null
@@ -1,75 +0,0 @@
-package legacymodels
-
-import (
-	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-)
-
-type VersionRequest struct {
-	ClientID string `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
-}
-
-type VersionResponse struct {
-	VersionInfo *models.BuildVersionInfo `json:"build_version_info"`
-}
-
-type CancelRequest = signatures.SignedRequest[model.JobCancelPayload]
-
-type CancelResponse struct {
-	State *model.JobState `json:"state"`
-}
-
-type EventsRequest struct {
-	ClientID string             `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
-	JobID    string             `json:"job_id" example:"9304c616-291f-41ad-b862-54e133c0149e"`
-	Options  EventFilterOptions `json:"filters"` // Records the number of seconds since the unix epoch (UTC)
-}
-
-type EventsResponse struct {
-	Events []model.JobHistory `json:"events"`
-}
-
-type EventFilterOptions = jobstore.JobHistoryFilterOptions
-
-type ListRequest struct {
-	JobID       string   `json:"id" example:"9304c616-291f-41ad-b862-54e133c0149e"`
-	ClientID    string   `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
-	IncludeTags []string `json:"include_tags" example:"['any-tag']"`
-	ExcludeTags []string `json:"exclude_tags" example:"['any-tag']"`
-	MaxJobs     uint32   `json:"max_jobs" example:"10"`
-	ReturnAll   bool     `json:"return_all" `
-	SortBy      string   `json:"sort_by" example:"created_at"`
-	SortReverse bool     `json:"sort_reverse"`
-}
-
-type ListResponse struct {
-	Jobs []*model.JobWithInfo `json:"jobs"`
-}
-
-type ResultsRequest struct {
-	ClientID string `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
-	JobID    string `json:"job_id" example:"9304c616-291f-41ad-b862-54e133c0149e"`
-}
-
-type ResultsResponse struct {
-	Results []model.PublishedResult `json:"results"`
-}
-
-type StateRequest struct {
-	ClientID string `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
-	JobID    string `json:"job_id" example:"9304c616-291f-41ad-b862-54e133c0149e"`
-}
-
-type StateResponse struct {
-	State model.JobState `json:"state"`
-}
-
-type SubmitRequest = signatures.SignedRequest[model.JobCreatePayload]
-
-type SubmitResponse struct {
-	Job *model.Job `json:"job"`
-}
-
-type LogRequest = signatures.SignedRequest[model.LogsPayload]
diff --git a/pkg/publicapi/client/client.go b/pkg/publicapi/client/client.go
deleted file mode 100644
index ed99c17f06..0000000000
--- a/pkg/publicapi/client/client.go
+++ /dev/null
@@ -1,228 +0,0 @@
-package client
-
-import (
-	"bytes"
-	"context"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net/http"
-	"net/url"
-	"os"
-	"reflect"
-	"time"
-
-	"github.com/pkg/errors"
-	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-
-	"github.com/bacalhau-project/bacalhau/pkg/bacerrors"
-	"github.com/bacalhau-project/bacalhau/pkg/config"
-	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	"github.com/bacalhau-project/bacalhau/pkg/lib/marshaller"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-)
-
-var NoTLS = LegacyTLSSupport{UseTLS: false}
-
-// APIClient is a utility for interacting with a node's API server against v1 APIs.
-type APIClient struct {
-	BaseURI        *url.URL
-	DefaultHeaders map[string]string
-	Client         *http.Client
-	Signer         system.Signer
-	ClientID       string
-}
-
-// LegacyTLSSupport contains information on how to use TLS (or not) to communicate
-// with the v1 APIs
-type LegacyTLSSupport struct {
-	UseTLS   bool
-	CACert   string
-	Insecure bool
-}
-
-// NewAPIClient returns a new client for a node's API server against v1 APIs
-// the client will use /api/v1 path by default is no custom path is defined
-// Deprecated: use util.GetAPIClient
-// TODOD(forrest) [WTF]: just more spaghetti. This needs to be a private method called from util.GetAPIClient
-func NewAPIClient(
-	tlsinfo LegacyTLSSupport,
-	cfg types.UserConfig,
-	host string,
-	port uint16,
-	path ...string,
-) (*APIClient, error) {
-	scheme := "http"
-	if tlsinfo.UseTLS {
-		scheme = "https"
-	}
-
-	sk, err := config.GetClientPrivateKey(cfg.KeyPath)
-	if err != nil {
-		return nil, err
-	}
-
-	baseURI := system.MustParseURL(fmt.Sprintf("%s://%s:%d", scheme, host, port)).JoinPath(path...)
-
-	tr := getTLSTransport(tlsinfo)
-
-	clientID, err := config.GetClientID(cfg.KeyPath)
-	if err != nil {
-		return nil, err
-	}
-
-	return &APIClient{
-		BaseURI:        baseURI,
-		DefaultHeaders: map[string]string{},
-		Signer:         system.NewMessageSigner(sk),
-		ClientID:       clientID,
-
-		Client: &http.Client{
-			Timeout: 300 * time.Second,
-			Transport: otelhttp.NewTransport(tr,
-				otelhttp.WithSpanOptions(
-					trace.WithAttributes(
-						attribute.String("clientID", clientID),
-					),
-				),
-			),
-		},
-	}, nil
-}
-
-// getTLSTransport builds a http.Transport from the TLS options
-func getTLSTransport(config LegacyTLSSupport) *http.Transport {
-	tr := &http.Transport{}
-
-	if !config.UseTLS {
-		return tr
-	}
-
-	if config.CACert != "" {
-		caCert, err := os.ReadFile(config.CACert)
-		if err != nil {
-			// unreachable: we already checked that the file exists at CLI startup
-			// if it has gone missing in the meantime then something is very wrong
-			newErr := errors.Wrap(err, fmt.Sprintf("Error: unable to read CA certificate: %s", config.CACert))
-			panic(newErr.Error())
-		}
-
-		caCertPool := x509.NewCertPool()
-		caCertPool.AppendCertsFromPEM(caCert)
-		tr.TLSClientConfig = &tls.Config{
-			RootCAs:    caCertPool,
-			MinVersion: tls.VersionTLS12,
-		}
-	} else if config.Insecure {
-		tr.TLSClientConfig = &tls.Config{
-			InsecureSkipVerify: true, //nolint:gosec
-			MinVersion:         tls.VersionTLS12,
-		}
-	}
-	return tr
-}
-
-func (apiClient *APIClient) doGet(ctx context.Context, api string, resData any) error {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/publicapi.Client.Get")
-	defer span.End()
-
-	addr := apiClient.BaseURI.JoinPath(api).String()
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, addr, nil)
-	if err != nil {
-		return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: error creating Get request: %v", err))
-	}
-	return apiClient.do(ctx, req, resData)
-}
-
-func (apiClient *APIClient) doPostSigned(ctx context.Context, api string, reqData, resData interface{}) error {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/publicapi.Client.DoPostSigned")
-	defer span.End()
-
-	req, err := signatures.SignRequest(apiClient.Signer, reqData)
-	if err != nil {
-		return err
-	}
-
-	return apiClient.DoPost(ctx, api, req, resData)
-}
-
-func (apiClient *APIClient) DoPost(ctx context.Context, api string, reqData, resData interface{}) error {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/publicapi.Client.DoPost")
-	defer span.End()
-
-	var body bytes.Buffer
-	var err error
-	if err = json.NewEncoder(&body).Encode(reqData); err != nil {
-		return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: error encoding request body: %v", err))
-	}
-
-	addr := apiClient.BaseURI.JoinPath(api).String()
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, addr, &body)
-	if err != nil {
-		return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: error creating Post request: %v", err))
-	}
-	req.Header.Set("Content-type", "application/json")
-	return apiClient.do(ctx, req, resData)
-}
-
-func (apiClient *APIClient) do(ctx context.Context, req *http.Request, resData any) error {
-	for header, value := range apiClient.DefaultHeaders {
-		req.Header.Set(header, value)
-	}
-	req.Close = true // don't keep connections lying around
-
-	var res *http.Response
-	res, err := apiClient.Client.Do(req)
-	if err != nil {
-		errString := err.Error()
-		if errorResponse, ok := err.(*bacerrors.ErrorResponse); ok {
-			return errorResponse
-		} else if errString == "context canceled" {
-			return bacerrors.NewContextCanceledError(err.Error())
-		} else {
-			return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: after posting request: %v", err))
-		}
-	}
-
-	defer func() {
-		if err = res.Body.Close(); err != nil {
-			err = fmt.Errorf("error closing response body: %v", err)
-		}
-	}()
-
-	if res.StatusCode != http.StatusOK {
-		var responseBody []byte
-		responseBody, err = io.ReadAll(res.Body)
-		if err != nil {
-			return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: error reading response body: %v", err))
-		}
-
-		var serverError *bacerrors.ErrorResponse
-		if err = marshaller.JSONUnmarshalWithMax(responseBody, &serverError); err != nil {
-			return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: after posting request: %v",
-				string(responseBody)))
-		}
-
-		if !reflect.DeepEqual(serverError, &bacerrors.ErrorResponse{}) {
-			return serverError
-		}
-
-		return errors.New(string(responseBody))
-	}
-
-	err = json.NewDecoder(res.Body).Decode(resData)
-	if err != nil {
-		if err == io.EOF {
-			return nil // No error, just no data
-		} else {
-			return bacerrors.NewResponseUnknownError(fmt.Errorf("publicapi: error decoding response body: %v", err))
-		}
-	}
-
-	return nil
-}
diff --git a/pkg/publicapi/client/core_apis.go b/pkg/publicapi/client/core_apis.go
deleted file mode 100644
index cb55422677..0000000000
--- a/pkg/publicapi/client/core_apis.go
+++ /dev/null
@@ -1,47 +0,0 @@
-package client
-
-import (
-	"context"
-	"io"
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-	"github.com/bacalhau-project/bacalhau/pkg/util/closer"
-)
-
-// Alive calls the node's API server health check.
-func (apiClient *APIClient) Alive(ctx context.Context) (bool, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/publicapi.Client.Alive")
-	defer span.End()
-
-	var body io.Reader
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiClient.BaseURI.JoinPath("/api/v1/livez").String(), body)
-	if err != nil {
-		return false, nil
-	}
-	res, err := apiClient.Client.Do(req) //nolint:bodyclose // golangcilint is dumb - this is closed
-	if err != nil {
-		return false, nil
-	}
-	defer closer.DrainAndCloseWithLogOnError(ctx, "apiClient response", res.Body)
-
-	return res.StatusCode == http.StatusOK, nil
-}
-
-func (apiClient *APIClient) Version(ctx context.Context) (*models.BuildVersionInfo, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/publicapi.Client.Version")
-	defer span.End()
-
-	req := legacymodels.VersionRequest{
-		ClientID: apiClient.ClientID,
-	}
-
-	var res legacymodels.VersionResponse
-	if err := apiClient.DoPost(ctx, "/api/v1/version", req, &res); err != nil {
-		return nil, err
-	}
-
-	return res.VersionInfo, nil
-}
diff --git a/pkg/publicapi/client/legacy_apis.go b/pkg/publicapi/client/legacy_apis.go
deleted file mode 100644
index 28386fe673..0000000000
--- a/pkg/publicapi/client/legacy_apis.go
+++ /dev/null
@@ -1,339 +0,0 @@
-package client
-
-import (
-	"context"
-	"fmt"
-	"net/url"
-	"strings"
-	"time"
-
-	"github.com/gorilla/websocket"
-	"github.com/rs/zerolog/log"
-
-	"github.com/bacalhau-project/bacalhau/pkg/bacerrors"
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-)
-
-// APIRetryCount - for some queries (like read events and read state)
-// we want to fail early (10 seconds should be ample time)
-// but retry a number of times - this is to avoid network
-// flakes failing the canary
-const APIRetryCount = 5
-const APIShortTimeoutSeconds = 10
-
-// List returns the list of jobs in the node's transport.
-func (apiClient *APIClient) List(
-	ctx context.Context,
-	idFilter string,
-	includeTags []string,
-	excludeTags []string,
-	maxJobs int,
-	returnAll bool,
-	sortBy string,
-	sortReverse bool,
-) (
-	[]*model.JobWithInfo, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.List")
-	defer span.End()
-
-	req := legacymodels.ListRequest{
-		ClientID:    apiClient.ClientID,
-		MaxJobs:     uint32(maxJobs),
-		JobID:       idFilter,
-		IncludeTags: includeTags,
-		ExcludeTags: excludeTags,
-		ReturnAll:   returnAll,
-		SortBy:      sortBy,
-		SortReverse: sortReverse,
-	}
-
-	var res legacymodels.ListResponse
-	if err := apiClient.DoPost(ctx, "/api/v1/requester/list", req, &res); err != nil {
-		return nil, err
-	}
-
-	return res.Jobs, nil
-}
-
-func (apiClient *APIClient) Nodes(ctx context.Context) ([]models.NodeInfo, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Nodes")
-	defer span.End()
-
-	var nodes []models.NodeInfo
-	err := apiClient.doGet(ctx, "/api/v1/requester/nodes", &nodes)
-	if err != nil {
-		return nil, err
-	}
-
-	return nodes, nil
-}
-
-// Cancel will request that the job with the specified ID is stopped. The JobInfo will be returned if the cancel
-// was submitted. If no match is found, Cancel returns false with a nil error.
-func (apiClient *APIClient) Cancel(ctx context.Context, jobID string, reason string) (*model.JobState, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Cancel")
-	defer span.End()
-
-	if jobID == "" {
-		return &model.JobState{}, fmt.Errorf("jobID must be non-empty in a Cancel call")
-	}
-
-	// Check the existence of a job with the provided ID, whether it is a short or long ID.
-	jobInfo, found, err := apiClient.Get(ctx, jobID)
-	if err != nil {
-		return &model.JobState{}, err
-	}
-	if !found {
-		return &model.JobState{}, bacerrors.NewJobNotFound(jobID)
-	}
-
-	// We potentially used the short jobID which `Get` supports and so let's switch
-	// to use the longer version.
-	jobID = jobInfo.State.JobID
-
-	// Create a payload before signing it with our local key (for verification on the
-	// server).
-	req := model.JobCancelPayload{
-		ClientID: apiClient.ClientID,
-		JobID:    jobID,
-		Reason:   reason,
-	}
-
-	var res legacymodels.CancelResponse
-	if err := apiClient.doPostSigned(ctx, "/api/v1/requester/cancel", req, &res); err != nil {
-		return &model.JobState{}, err
-	}
-
-	return res.State, nil
-}
-
-// Get returns job data for a particular job ID. If no match is found, Get returns false with a nil error.
-func (apiClient *APIClient) Get(ctx context.Context, jobID string) (*model.JobWithInfo, bool, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Get")
-	defer span.End()
-
-	if jobID == "" {
-		return &model.JobWithInfo{}, false, fmt.Errorf("jobID must be non-empty in a Get call")
-	}
-
-	jobsList, err := apiClient.List(ctx, jobID, model.IncludeAny, model.ExcludeNone, 1, false, "created_at", true)
-	if err != nil {
-		return &model.JobWithInfo{}, false, err
-	}
-
-	if len(jobsList) > 0 {
-		return jobsList[0], true, nil
-	} else {
-		return &model.JobWithInfo{}, false, bacerrors.NewJobNotFound(jobID)
-	}
-}
-
-func (apiClient *APIClient) GetJobState(ctx context.Context, jobID string) (model.JobState, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.GetJobState")
-	defer span.End()
-
-	if jobID == "" {
-		return model.JobState{}, fmt.Errorf("jobID must be non-empty in a GetJobStates call")
-	}
-
-	req := legacymodels.StateRequest{
-		ClientID: apiClient.ClientID,
-		JobID:    jobID,
-	}
-
-	var res legacymodels.StateResponse
-	var outerErr error
-
-	for i := 0; i < APIRetryCount; i++ {
-		shortTimeoutCtx, cancelFn := context.WithTimeout(ctx, time.Second*APIShortTimeoutSeconds)
-		defer cancelFn()
-		err := apiClient.DoPost(shortTimeoutCtx, "/api/v1/requester/states", req, &res)
-		if err == nil {
-			return res.State, nil
-		} else {
-			log.Ctx(ctx).Debug().Err(err).Msg("apiclient read state error")
-			outerErr = err
-		}
-	}
-	return model.JobState{}, outerErr
-}
-
-func (apiClient *APIClient) GetJobStateResolver() *legacy_job.StateResolver {
-	jobLoader := func(ctx context.Context, jobID string) (model.Job, error) {
-		j, _, err := apiClient.Get(ctx, jobID)
-		if err != nil {
-			return model.Job{}, fmt.Errorf("failed to load job %s: %w", jobID, err)
-		}
-		return j.Job, err
-	}
-	stateLoader := func(ctx context.Context, jobID string) (model.JobState, error) {
-		return apiClient.GetJobState(ctx, jobID)
-	}
-	return legacy_job.NewStateResolver(jobLoader, stateLoader)
-}
-
-func (apiClient *APIClient) GetEvents(
-	ctx context.Context,
-	jobID string,
-	options legacymodels.EventFilterOptions) (events []model.JobHistory, err error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.GetEvents")
-	defer span.End()
-
-	if jobID == "" {
-		return nil, fmt.Errorf("jobID must be non-empty in a GetEvents call")
-	}
-
-	req := legacymodels.EventsRequest{
-		ClientID: apiClient.ClientID,
-		JobID:    jobID,
-		Options:  options,
-	}
-
-	// Test if the context has been canceled before making the request.
-	var res legacymodels.EventsResponse
-	var outerErr error
-
-	for i := 0; i < APIRetryCount; i++ {
-		shortTimeoutCtx, cancelFn := context.WithTimeout(ctx, time.Second*APIShortTimeoutSeconds)
-		defer cancelFn()
-		err = apiClient.DoPost(shortTimeoutCtx, "/api/v1/requester/events", req, &res)
-		if err == nil {
-			return res.Events, nil
-		} else {
-			log.Ctx(ctx).Debug().Err(err).Msg("apiclient read events error")
-			outerErr = err
-			if strings.Contains(err.Error(), "context canceled") {
-				outerErr = bacerrors.NewContextCanceledError(ctx.Err().Error())
-			}
-		}
-	}
-	return nil, outerErr
-}
-
-func (apiClient *APIClient) GetResults(ctx context.Context, jobID string) (results []model.PublishedResult, err error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.GetResults")
-	defer span.End()
-
-	if jobID == "" {
-		return nil, fmt.Errorf("jobID must be non-empty in a GetResults call")
-	}
-
-	req := legacymodels.ResultsRequest{
-		ClientID: apiClient.ClientID,
-		JobID:    jobID,
-	}
-
-	var res legacymodels.ResultsResponse
-	if err := apiClient.DoPost(ctx, "/api/v1/requester/results", req, &res); err != nil {
-		return nil, err
-	}
-
-	return res.Results, nil
-}
-
-// Submit submits a new job to the node's transport.
-func (apiClient *APIClient) Submit(
-	ctx context.Context,
-	j *model.Job,
-) (*model.Job, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Submit")
-	defer span.End()
-
-	data := model.JobCreatePayload{
-		ClientID:   apiClient.ClientID,
-		APIVersion: j.APIVersion,
-		Spec:       &j.Spec,
-	}
-
-	var res legacymodels.SubmitResponse
-	err := apiClient.doPostSigned(ctx, "/api/v1/requester/submit", data, &res)
-	if err != nil {
-		return &model.Job{}, err
-	}
-
-	return res.Job, nil
-}
-
-// Logs will retrieve the address of an endpoint where a client connection can be
-// made to stream the results of an execution back to a TTY
-func (apiClient *APIClient) Logs(
-	ctx context.Context,
-	jobID string,
-	executionID string,
-	withHistory bool,
-	follow bool) (*websocket.Conn, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Logs")
-	defer span.End()
-
-	if jobID == "" {
-		return nil, fmt.Errorf("jobID must be non-empty in a logs call")
-	}
-
-	if executionID == "" {
-		return nil, fmt.Errorf("executionID must be non-empty in a logs call")
-	}
-
-	// Check the existence of a job with the provided ID, whether it is a short or long ID.
-	jobInfo, found, err := apiClient.Get(ctx, jobID)
-	if err != nil {
-		return nil, err
-	}
-	if !found {
-		return nil, bacerrors.NewJobNotFound(jobID)
-	}
-
-	// We potentially used the short jobID which `Get` supports and so let's switch
-	// to use the longer version.
-	jobID = jobInfo.State.JobID
-
-	// Create a payload before signing it with our local key (for verification on the
-	// server).
-	payload := model.LogsPayload{
-		ClientID:    apiClient.ClientID,
-		JobID:       jobID,
-		ExecutionID: executionID,
-		Tail:        withHistory,
-		Follow:      follow,
-	}
-
-	req, err := signatures.SignRequest(apiClient.Signer, payload)
-	if err != nil {
-		return nil, err
-	}
-
-	u, _ := url.Parse(apiClient.BaseURI.String())
-	u.Scheme = "ws"
-	u.Path = "/api/v1/requester/logs"
-
-	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil) //nolint:bodyclose
-	if err != nil {
-		log.Ctx(ctx).Error().Msgf("Failed to dial to : %s", u.String())
-		return nil, err
-	}
-
-	err = c.WriteJSON(req)
-	if err != nil {
-		log.Ctx(ctx).Error().Msg("Failed to write the JSON for the start of the logs request")
-		return nil, err
-	}
-
-	return c, nil
-}
-
-func (apiClient *APIClient) Debug(ctx context.Context) (map[string]model.DebugInfo, error) {
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester/publicapi.APIClient.Debug")
-	defer span.End()
-
-	req := struct{}{}
-	var res map[string]model.DebugInfo
-	if err := apiClient.DoPost(ctx, "/api/v1/requester/debug", req, &res); err != nil {
-		return res, err
-	}
-
-	return res, nil
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoint.go b/pkg/publicapi/endpoint/requester/endpoint.go
index dab5cec08c..b006437a50 100644
--- a/pkg/publicapi/endpoint/requester/endpoint.go
+++ b/pkg/publicapi/endpoint/requester/endpoint.go
@@ -3,68 +3,23 @@ package requester
 import (
 	"fmt"
 	"net/http"
-	"os"
-	"sync"
 
-	"github.com/gorilla/websocket"
 	"github.com/labstack/echo/v4"
 
-	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/middleware"
-	"github.com/bacalhau-project/bacalhau/pkg/requester"
 )
 
-type EndpointParams struct {
-	Router             *echo.Echo
-	Requester          requester.Endpoint
-	DebugInfoProviders []model.DebugInfoProvider
-	JobStore           jobstore.Store
-	NodeDiscoverer     orchestrator.NodeDiscoverer
-}
-
 type Endpoint struct {
-	router             *echo.Echo
-	requester          requester.Endpoint
-	debugInfoProviders []model.DebugInfoProvider
-	jobStore           jobstore.Store
-	nodeDiscoverer     orchestrator.NodeDiscoverer
-	// jobId or "" (for all events) -> connections for that subscription
-	websockets      map[string][]*websocket.Conn
-	websocketsMutex sync.RWMutex
+	router *echo.Echo
 }
 
-// NB: to whomever removes this, grep for the env var when doing so, we can't
-// use this variable everywhere due to circular deps
-// TODO: https://github.com/bacalhau-project/bacalhau/issues/4119
-const UseDeprecatedEndpointsForTesting = "REQUESTER_ENDPOINT_USE_DEPRECATED_ENV"
-
-func NewEndpoint(params EndpointParams) *Endpoint {
+func NewEndpoint(router *echo.Echo) *Endpoint {
 	e := &Endpoint{
-		router:             params.Router,
-		requester:          params.Requester,
-		debugInfoProviders: params.DebugInfoProviders,
-		jobStore:           params.JobStore,
-		nodeDiscoverer:     params.NodeDiscoverer,
-		websockets:         make(map[string][]*websocket.Conn),
+		router: router,
 	}
 
 	g := e.router.Group("/api/v1/requester")
 	g.Use(middleware.SetContentType(echo.MIMEApplicationJSON))
-	if key := os.Getenv(UseDeprecatedEndpointsForTesting); key != "" {
-		g.POST("/list", e.list)
-		g.GET("/nodes", e.nodes)
-		g.POST("/states", e.states)
-		g.POST("/results", e.results)
-		g.POST("/events", e.events)
-		g.POST("/submit", e.submit)
-		g.POST("/cancel", e.cancel)
-		g.POST("/debug", e.debug)
-		g.GET("/websocket/events", e.websocketJobEvents)
-		return e
-	}
-
 	registerDeprecatedLegacyMethods(g)
 	return e
 }
diff --git a/pkg/publicapi/endpoint/requester/endpoint_debug.go b/pkg/publicapi/endpoint/requester/endpoint_debug.go
deleted file mode 100644
index 2c0928f4fa..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoint_debug.go
+++ /dev/null
@@ -1,30 +0,0 @@
-package requester
-
-import (
-	"net/http"
-
-	"github.com/labstack/echo/v4"
-	"github.com/rs/zerolog/log"
-)
-
-// debug godoc
-//
-//	@ID			pkg/requester/publicapi/debug
-//	@Summary	Returns debug information on what the current node is doing.
-//	@Tags		Health
-//	@Produce	json
-//	@Success	200	{object}	string
-//	@Failure	500	{object}	string
-//	@Router		/api/v1/requester/debug [get]
-func (s *Endpoint) debug(c echo.Context) error {
-	debugInfoMap := make(map[string]interface{})
-	for _, provider := range s.debugInfoProviders {
-		debugInfo, err := provider.GetDebugInfo(c.Request().Context())
-		if err != nil {
-			log.Ctx(c.Request().Context()).Error().Msgf("could not get debug info from some providers: %s", err)
-			continue
-		}
-		debugInfoMap[debugInfo.Component] = debugInfo.Info
-	}
-	return c.JSON(http.StatusOK, debugInfoMap)
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_cancel.go b/pkg/publicapi/endpoint/requester/endpoints_cancel.go
deleted file mode 100644
index 0bf467954f..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_cancel.go
+++ /dev/null
@@ -1,82 +0,0 @@
-package requester
-
-import (
-	"fmt"
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-	"github.com/bacalhau-project/bacalhau/pkg/requester"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-	"github.com/labstack/echo/v4"
-	"github.com/pkg/errors"
-)
-
-// @ID				pkg/requester/publicapi/cancel
-// @Summary		Cancels the job with the job-id specified in the body payload.
-// @Description	Cancels a job specified by `id` as long as that job belongs to `client_id`.
-// @Description	Returns the current jobstate after the cancel request has been processed.
-// @Tags			Job
-// @Accept			json
-// @Produce		json
-// @Param			CancelRequest	body		legacymodels.CancelRequest	true	" "
-// @Success		200				{object}	legacymodels.CancelResponse
-// @Failure		400				{object}	string
-// @Failure		401				{object}	string
-// @Failure		403				{object}	string
-// @Failure		500				{object}	string
-// @Router			/api/v1/requester/cancel [post]
-func (s *Endpoint) cancel(c echo.Context) error {
-	ctx := c.Request().Context()
-	jobCancelPayload, err := signatures.UnmarshalSigned[model.JobCancelPayload](ctx, c.Request().Body)
-	if err != nil {
-		publicapi.HTTPError(c, err, http.StatusBadRequest)
-		return nil
-	}
-
-	c.Response().Header().Set(apimodels.HTTPHeaderClientID, jobCancelPayload.ClientID)
-	ctx = system.AddJobIDToBaggage(ctx, jobCancelPayload.ClientID)
-
-	// Get the job, check it exists and check it belongs to the same client
-	job, err := s.jobStore.GetJob(ctx, jobCancelPayload.JobID)
-	if err != nil {
-		publicapi.HTTPError(c, errors.Wrap(err, "missing job"), http.StatusNotFound)
-		return nil
-	}
-
-	// We can compare the payload's client ID against the existing job's metadata
-	// as we have confirmed the public key that the request was signed with matches
-	// the client ID the request claims.
-	if job.Namespace != jobCancelPayload.ClientID {
-		err = fmt.Errorf("mismatched ClientIDs for cancel, existing job: %s and cancel request: %s",
-			job.Namespace, jobCancelPayload.ClientID)
-		publicapi.HTTPError(c, err, http.StatusUnauthorized)
-		return nil
-	}
-
-	_, err = s.requester.CancelJob(ctx, requester.CancelJobRequest{
-		JobID:         jobCancelPayload.JobID,
-		Reason:        jobCancelPayload.Reason,
-		UserTriggered: true,
-	})
-	if err != nil {
-		publicapi.HTTPError(c, err, http.StatusInternalServerError)
-		return nil
-	}
-
-	jobState, err := legacy.GetJobState(ctx, s.jobStore, jobCancelPayload.JobID)
-	if err != nil {
-		publicapi.HTTPError(c, err, http.StatusInternalServerError)
-		return nil
-	}
-
-	c.Response().Header().Set(apimodels.HTTPHeaderClientID, jobCancelPayload.ClientID)
-	c.Response().Header().Set(apimodels.HTTPHeaderJobID, jobCancelPayload.JobID)
-	return c.JSON(http.StatusOK, legacymodels.CancelResponse{
-		State: &jobState,
-	})
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_events.go b/pkg/publicapi/endpoint/requester/endpoints_events.go
deleted file mode 100644
index 00c92d945c..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_events.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package requester
-
-import (
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/labstack/echo/v4"
-)
-
-// @ID				pkg/requester/publicapi/events
-// @Summary		Returns the events related to the job-id passed in the body payload. Useful for troubleshooting.
-// @Description	Events (e.g. Created, Bid, BidAccepted, ..., ResultsAccepted, ResultsPublished) are useful to track the progress of a job.
-// @Tags			Job
-// @Accept			json
-// @Produce		json
-// @Param			EventsRequest	body		legacymodels.EventsRequest	true	"Request must specify a `client_id`. To retrieve your `client_id`, you can do the following: (1) submit a dummy job to Bacalhau (or use one you created before), (2) run `bacalhau describe <job-id>` and fetch the `ClientID` field."
-// @Success		200				{object}	legacymodels.EventsResponse
-// @Failure		400				{object}	string
-// @Failure		500				{object}	string
-// @Router			/api/v1/requester/events [post]
-//
-//nolint:lll
-//nolint:dupl
-func (s *Endpoint) events(c echo.Context) error {
-	var eventsReq legacymodels.EventsRequest
-
-	if err := c.Bind(&eventsReq); err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
-	}
-
-	c.Response().Header().Set(apimodels.HTTPHeaderClientID, eventsReq.ClientID)
-	c.Response().Header().Set(apimodels.HTTPHeaderJobID, eventsReq.JobID)
-
-	ctx := c.Request().Context()
-	events, err := s.jobStore.GetJobHistory(ctx, eventsReq.JobID, eventsReq.Options)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
-	}
-
-	legacyEvents := make([]model.JobHistory, len(events))
-	for i := range events {
-		legacyEvents[i] = *legacy.ToLegacyJobHistory(&events[i])
-	}
-
-	return c.JSON(http.StatusOK, legacymodels.EventsResponse{
-		Events: legacyEvents,
-	})
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_list.go b/pkg/publicapi/endpoint/requester/endpoints_list.go
deleted file mode 100644
index 8b58f613a0..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_list.go
+++ /dev/null
@@ -1,109 +0,0 @@
-package requester
-
-import (
-	"context"
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/bacerrors"
-	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/labstack/echo/v4"
-	"github.com/rs/zerolog/log"
-)
-
-// @ID				pkg/requester/publicapi/list
-// @Summary		Simply lists jobs.
-// @Description	Returns the first (sorted) #`max_jobs` jobs that belong to the `client_id` passed in the body payload (by default).
-// @Description	If `return_all` is set to true, it returns all jobs on the Bacalhau network.
-// @Description	If `id` is set, it returns only the job with that ID.
-// @Tags			Job
-// @Accept			json
-// @Produce		json
-// @Param			ListRequest	body		legacymodels.ListRequest	true	"Set `return_all` to `true` to return all jobs on the network (may degrade performance, use with care!)."
-// @Success		200			{object}	legacymodels.ListResponse
-// @Failure		400			{object}	string
-// @Failure		500			{object}	string
-// @Router			/api/v1/requester/list [post]
-//
-//nolint:lll
-func (s *Endpoint) list(c echo.Context) error {
-	var listReq legacymodels.ListRequest
-	var err error
-	var jobList []model.Job
-
-	if err := c.Bind(&listReq); err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
-	}
-
-	if listReq.JobID != "" {
-		jobList, err = s.getSingleJob(c.Request().Context(), listReq.JobID)
-	} else {
-		jobList, err = s.getJobsList(c.Request().Context(), listReq)
-	}
-
-	if err != nil {
-		_, ok := err.(*bacerrors.JobNotFound)
-		if ok {
-			http.Error(c.Response(), bacerrors.ErrorToErrorResponse(err), http.StatusBadRequest)
-			return nil
-		}
-		return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
-	}
-
-	jobWithInfos := make([]*model.JobWithInfo, len(jobList))
-	for i, job := range jobList {
-		jobState, innerErr := legacy.GetJobState(c.Request().Context(), s.jobStore, job.ID())
-		if innerErr != nil {
-			log.Ctx(c.Request().Context()).Error().Err(innerErr).Msg("error getting job states")
-			return echo.NewHTTPError(http.StatusInternalServerError, innerErr.Error())
-		}
-		jobWithInfos[i] = &model.JobWithInfo{
-			Job:   job,
-			State: jobState,
-		}
-	}
-
-	return c.JSON(http.StatusOK, legacymodels.ListResponse{
-		Jobs: jobWithInfos,
-	})
-}
-
-func (s *Endpoint) getSingleJob(ctx context.Context, jobID string) ([]model.Job, error) {
-	job, err := s.jobStore.GetJob(ctx, jobID)
-	if err != nil {
-		return nil, err
-	}
-
-	convertedJob, err := legacy.ToLegacyJob(&job)
-	if err != nil {
-		return nil, err
-	}
-
-	return []model.Job{*convertedJob}, nil
-}
-
-func (s *Endpoint) getJobsList(ctx context.Context, listReq legacymodels.ListRequest) ([]model.Job, error) {
-	response, err := s.jobStore.GetJobs(ctx, jobstore.JobQuery{
-		Namespace:   listReq.ClientID,
-		Limit:       listReq.MaxJobs,
-		IncludeTags: listReq.IncludeTags,
-		ExcludeTags: listReq.ExcludeTags,
-		ReturnAll:   listReq.ReturnAll,
-		SortBy:      listReq.SortBy,
-		SortReverse: listReq.SortReverse,
-	})
-	if err != nil {
-		return nil, err
-	}
-	res := make([]model.Job, len(response.Jobs))
-	for i := range response.Jobs {
-		legacyJob, err := legacy.ToLegacyJob(&response.Jobs[i])
-		if err != nil {
-			return nil, err
-		}
-		res[i] = *legacyJob
-	}
-	return res, nil
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_nodes.go b/pkg/publicapi/endpoint/requester/endpoints_nodes.go
deleted file mode 100644
index 1b31ca034b..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_nodes.go
+++ /dev/null
@@ -1,26 +0,0 @@
-package requester
-
-import (
-	"net/http"
-
-	"github.com/labstack/echo/v4"
-)
-
-// nodes godoc
-//
-//	@ID			pkg/requester/publicapi/nodes
-//	@Summary	Displays the nodes that this requester knows about
-//	@Accept		json
-//	@Produce	json
-//	@Success	200	{object}	[]models.NodeInfo
-//	@Failure	500	{object}	string
-//	@Router		/api/v1/requester/nodes [get]
-func (s *Endpoint) nodes(c echo.Context) error {
-	ctx := c.Request().Context()
-	nodes, err := s.nodeDiscoverer.List(ctx)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
-	}
-
-	return c.JSON(http.StatusOK, nodes)
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_result.go b/pkg/publicapi/endpoint/requester/endpoints_result.go
deleted file mode 100644
index 1a18225879..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_result.go
+++ /dev/null
@@ -1,61 +0,0 @@
-package requester
-
-import (
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-	"github.com/labstack/echo/v4"
-	oteltrace "go.opentelemetry.io/otel/trace"
-)
-
-// @ID			pkg/requester/publicapi/results
-// @Summary	Returns the results of the job-id specified in the body payload.
-// @Tags		Job
-// @Accept		json
-// @Produce	json
-// @Param		StateRequest	body		legacymodels.StateRequest	true	" "
-// @Success	200				{object}	legacymodels.ResultsResponse
-// @Failure	400				{object}	string
-// @Failure	500				{object}	string
-// @Router		/api/v1/requester/results [post]
-func (s *Endpoint) results(c echo.Context) error {
-	var stateReq legacymodels.StateRequest
-	if err := c.Bind(&stateReq); err != nil {
-		return c.String(http.StatusBadRequest, err.Error())
-	}
-	c.Response().Header().Set(apimodels.HTTPHeaderClientID, stateReq.ClientID)
-	c.Response().Header().Set(apimodels.HTTPHeaderJobID, stateReq.JobID)
-
-	ctx := c.Request().Context()
-	ctx = system.AddJobIDToBaggage(ctx, stateReq.JobID)
-	system.AddJobIDFromBaggageToSpan(ctx, oteltrace.SpanFromContext(ctx))
-
-	executions, err := s.jobStore.GetExecutions(ctx, jobstore.GetExecutionsOptions{
-		JobID: stateReq.JobID,
-	})
-	if err != nil {
-		return c.JSON(http.StatusInternalServerError, err.Error())
-	}
-
-	results := make([]model.PublishedResult, 0)
-	for _, execution := range executions {
-		if execution.ComputeState.StateType == models.ExecutionStateCompleted {
-			storageConfig, err := legacy.ToLegacyStorageSpec(execution.PublishedResult)
-			if err != nil {
-				return c.JSON(http.StatusInternalServerError, err.Error())
-			}
-			results = append(results, model.PublishedResult{
-				NodeID: execution.NodeID,
-				Data:   storageConfig,
-			})
-		}
-	}
-
-	return c.JSON(http.StatusOK, legacymodels.ResultsResponse{Results: results})
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_states.go b/pkg/publicapi/endpoint/requester/endpoints_states.go
deleted file mode 100644
index f76decb44b..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_states.go
+++ /dev/null
@@ -1,42 +0,0 @@
-package requester
-
-import (
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-	"github.com/labstack/echo/v4"
-)
-
-// states godoc
-//
-//	@ID			pkg/requester/publicapi/states
-//	@Summary	Returns the state of the job-id specified in the body payload.
-//	@Tags		Job
-//	@Accept		json
-//	@Produce	json
-//	@Param		StateRequest	body		legacymodels.StateRequest	true	" "
-//	@Success	200				{object}	legacymodels.StateResponse
-//	@Failure	400				{object}	string
-//	@Failure	500				{object}	string
-//	@Router		/api/v1/requester/states [post]
-func (s *Endpoint) states(c echo.Context) error {
-	var stateReq legacymodels.StateRequest
-	if err := c.Bind(&stateReq); err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
-	}
-	c.Response().Header().Set(apimodels.HTTPHeaderClientID, stateReq.ClientID)
-	c.Response().Header().Set(apimodels.HTTPHeaderJobID, stateReq.JobID)
-
-	ctx := c.Request().Context()
-	ctx = system.AddJobIDToBaggage(ctx, stateReq.JobID)
-
-	js, err := legacy.GetJobState(ctx, s.jobStore, stateReq.JobID)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
-	}
-
-	return c.JSON(http.StatusOK, legacymodels.StateResponse{State: js})
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_submit.go b/pkg/publicapi/endpoint/requester/endpoints_submit.go
deleted file mode 100644
index 72f6e2f17d..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_submit.go
+++ /dev/null
@@ -1,59 +0,0 @@
-package requester
-
-import (
-	"fmt"
-	"net/http"
-
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
-	"github.com/labstack/echo/v4"
-	oteltrace "go.opentelemetry.io/otel/trace"
-)
-
-// @ID			pkg/requester/publicapi/submit
-// @Summary	Submits a new job to the network.
-// @Tags		Job
-// @Accept		json
-// @Produce	json
-// @Param		SubmitRequest	body		legacymodels.SubmitRequest	true	" "
-// @Success	200				{object}	legacymodels.SubmitResponse
-// @Failure	400				{object}	string
-// @Failure	500				{object}	string
-// @Router		/api/v1/requester/submit [post]
-func (s *Endpoint) submit(c echo.Context) error {
-	ctx := c.Request().Context()
-
-	if otherJobID := c.Request().Header.Get("X-Bacalhau-Job-ID"); otherJobID != "" {
-		err := fmt.Errorf("rejecting job because HTTP header X-Bacalhau-Job-ID was set")
-		publicapi.HTTPError(c, err, http.StatusBadRequest)
-		return nil
-	}
-
-	jobCreatePayload, err := signatures.UnmarshalSigned[model.JobCreatePayload](ctx, c.Request().Body)
-	if err != nil {
-		publicapi.HTTPError(c, err, http.StatusBadRequest)
-		return nil
-	}
-
-	if err := legacy_job.VerifyJobCreatePayload(ctx, &jobCreatePayload); err != nil {
-		publicapi.HTTPError(c, err, http.StatusBadRequest)
-		return nil
-	}
-
-	j, err := s.requester.SubmitJob(ctx, jobCreatePayload)
-	if err != nil {
-		publicapi.HTTPError(c, err, http.StatusInternalServerError)
-		return nil
-	}
-
-	c.Response().Header().Set(apimodels.HTTPHeaderJobID, j.Metadata.ID)
-	ctx = system.AddJobIDToBaggage(ctx, j.Metadata.ID)
-	system.AddJobIDFromBaggageToSpan(ctx, oteltrace.SpanFromContext(ctx))
-
-	return c.JSON(http.StatusOK, legacymodels.SubmitResponse{Job: j})
-}
diff --git a/pkg/publicapi/endpoint/requester/endpoints_websockets_events.go b/pkg/publicapi/endpoint/requester/endpoints_websockets_events.go
deleted file mode 100644
index 4d935c0827..0000000000
--- a/pkg/publicapi/endpoint/requester/endpoints_websockets_events.go
+++ /dev/null
@@ -1,114 +0,0 @@
-package requester
-
-import (
-	"context"
-	"net/http"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/gorilla/websocket"
-	"github.com/labstack/echo/v4"
-	"github.com/rs/zerolog/log"
-)
-
-var upgrader = websocket.Upgrader{
-	ReadBufferSize:  1024,
-	WriteBufferSize: 1024,
-}
-
-// TODO: Godoc
-
-func (s *Endpoint) websocketJobEvents(c echo.Context) error {
-	ctx := c.Request().Context()
-	conn, err := upgrader.Upgrade(c.Response(), c.Request(), nil)
-	if err != nil {
-		return c.String(http.StatusInternalServerError, err.Error())
-	}
-	log.Ctx(ctx).Debug().Msgf("New websocketJobEvents connection.")
-	defer conn.Close()
-
-	// NB: jobId == "" is the case for subscriptions to "all events"
-
-	// get job_id from query string
-	jobID := c.QueryParam("job_id")
-
-	func() {
-		s.websocketsMutex.Lock()
-		defer s.websocketsMutex.Unlock()
-
-		sockets, ok := s.websockets[jobID]
-		if !ok {
-			sockets = []*websocket.Conn{}
-			s.websockets[jobID] = sockets
-		}
-		s.websockets[jobID] = append(sockets, conn)
-	}()
-
-	if jobID != "" {
-		// list events for job out of localDB and send them to the client
-		events, err := s.jobStore.GetJobHistory(context.Background(), jobID, legacymodels.EventFilterOptions{})
-		if err != nil {
-			log.Ctx(ctx).Error().Msgf("error listing job events: %s\n", err.Error())
-			return err
-		}
-		for _, event := range events {
-			err := conn.WriteJSON(event)
-			if err != nil {
-				log.Ctx(ctx).Error().Msgf("error writing event JSON: %s\n", err.Error())
-			}
-		}
-	}
-
-	for {
-		// read and throw away any incoming messages, exit when client
-		// disconnects (which is a sort of error)
-		_, _, err := conn.ReadMessage()
-		if err != nil {
-			break
-		}
-	}
-
-	return nil
-}
-
-func (s *Endpoint) HandleJobEvent(ctx context.Context, event model.JobEvent) error {
-	s.websocketsMutex.Lock()
-	defer s.websocketsMutex.Unlock()
-
-	dispatchAndCleanup := func(jobId string) {
-		connections, ok := s.websockets[jobId]
-		if !ok {
-			return
-		}
-		errIdxs := []int{}
-		for idx, connection := range connections {
-			// TODO: dispatch to subscribers in parallel, to avoid one slow
-			// reader slowing all the others down.
-			err := connection.WriteJSON(event)
-			if err != nil {
-				log.Ctx(ctx).Error().Msgf(
-					"error writing event to subscriber '%s'/%d: %s, closing ws\n",
-					jobId, idx, err.Error(),
-				)
-				errIdxs = append(errIdxs, idx)
-				// close the connection, if possible, to allow the other side to
-				// retry. Ignore errors from closing, since we are going to
-				// delete this connection anyway.
-				connection.Close()
-			}
-		}
-		// reverse errIdxs (so we don't mess up the indexes for cleanup) and
-		// clean up dud connections
-		for i := len(errIdxs)/2 - 1; i >= 0; i-- {
-			opp := len(errIdxs) - 1 - i
-			errIdxs[i], errIdxs[opp] = errIdxs[opp], errIdxs[i]
-		}
-		for _, idx := range errIdxs {
-			connections = append(connections[:idx], connections[idx+1:]...)
-		}
-		s.websockets[jobId] = connections
-	}
-	dispatchAndCleanup("")
-	dispatchAndCleanup(event.JobID)
-	return nil
-}
diff --git a/pkg/publicapi/endpoint/shared/endpoint.go b/pkg/publicapi/endpoint/shared/endpoint.go
index 6b2fe1f2c6..000bd88675 100644
--- a/pkg/publicapi/endpoint/shared/endpoint.go
+++ b/pkg/publicapi/endpoint/shared/endpoint.go
@@ -6,7 +6,6 @@ import (
 	"github.com/labstack/echo/v4"
 
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/middleware"
 	"github.com/bacalhau-project/bacalhau/pkg/version"
 )
@@ -78,28 +77,37 @@ func (e *Endpoint) nodeInfo(c echo.Context) error {
 	return c.JSON(http.StatusOK, e.nodeStateProvider.GetNodeState(c.Request().Context()))
 }
 
+type VersionRequest struct {
+	ClientID string `json:"client_id" example:"ac13188e93c97a9c2e7cf8e86c7313156a73436036f30da1ececc2ce79f9ea51"`
+}
+
+type VersionResponse struct {
+	VersionInfo *models.BuildVersionInfo `json:"build_version_info"`
+}
+
 // version godoc
 //
-//	@ID				apiServer/version
-//	@Summary		Returns the build version running on the server.
-//	@Description	See https://github.com/bacalhau-project/bacalhau/releases for a complete list of `gitversion` tags.
-//	@Tags			Misc
-//	@Accept			json
-//	@Produce		json
-//	@Param			VersionRequest	body		legacymodels.VersionRequest	true	"Request must specify a `client_id`. To retrieve your `client_id`, you can do the following: (1) submit a dummy job to Bacalhau (or use one you created before), (2) run `bacalhau describe <job-id>` and fetch the `ClientID` field."
-//	@Success		200				{object}	legacymodels.VersionResponse
-//	@Failure		400				{object}	string
-//	@Failure		500				{object}	string
-//	@Router			/api/v1/version [post]
+//		@ID				apiServer/version
+//		@Summary		Returns the build version running on the server.
+//		@Description	See https://github.com/bacalhau-project/bacalhau/releases for a complete list of `gitversion` tags.
+//		@Tags			Misc
+//		@Accept			json
+//		@Produce		json
+//	 @Param          VersionRequest  body 		VersionRequest
+//		@Param			VersionRequest	body		legacymodels.VersionRequest	true	"Request must specify a `client_id`. To retrieve your `client_id`, you can do the following: (1) submit a dummy job to Bacalhau (or use one you created before), (2) run `bacalhau describe <job-id>` and fetch the `ClientID` field."
+//		@Success		200				{object}	legacymodels.VersionResponse
+//		@Failure		400				{object}	string
+//		@Failure		500				{object}	string
+//		@Router			/api/v1/version [post]
 //
 //nolint:lll
 func (e *Endpoint) version(c echo.Context) error {
-	var versionReq legacymodels.VersionRequest
+	var versionReq VersionRequest
 	if err := c.Bind(&versionReq); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
 	}
 
-	return c.JSON(http.StatusOK, legacymodels.VersionResponse{
+	return c.JSON(http.StatusOK, VersionResponse{
 		VersionInfo: version.Get(),
 	})
 }
diff --git a/pkg/requester/endpoint.go b/pkg/requester/endpoint.go
index 623804f886..5ae41a954b 100644
--- a/pkg/requester/endpoint.go
+++ b/pkg/requester/endpoint.go
@@ -2,211 +2,36 @@ package requester
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/rs/zerolog/log"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
 	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
-	"github.com/bacalhau-project/bacalhau/pkg/requester/jobtransform"
-	"github.com/bacalhau-project/bacalhau/pkg/system"
 )
 
 type BaseEndpointParams struct {
-	ID                     string
-	Store                  jobstore.Store
-	EventEmitter           orchestrator.EventEmitter
-	ComputeEndpoint        compute.Endpoint
-	MinJobExecutionTimeout time.Duration
-	DefaultJobTimeout      time.Duration
-	DefaultPublisher       string
+	ID           string
+	Store        jobstore.Store
+	EventEmitter orchestrator.EventEmitter
 }
 
 // BaseEndpoint base implementation of requester Endpoint
 type BaseEndpoint struct {
-	id             string
-	store          jobstore.Store
-	eventEmitter   orchestrator.EventEmitter
-	computesvc     compute.Endpoint
-	transforms     []jobtransform.Transformer
-	postTransforms []jobtransform.PostTransformer
+	id           string
+	store        jobstore.Store
+	eventEmitter orchestrator.EventEmitter
 }
 
 func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
-	transforms := []jobtransform.Transformer{
-		jobtransform.NewTimeoutApplier(params.MinJobExecutionTimeout, params.DefaultJobTimeout),
-		jobtransform.NewRequesterInfo(params.ID),
-		jobtransform.NewPublisherMigrator(params.DefaultPublisher),
-		jobtransform.NewEngineMigrator(),
-		// jobtransform.DockerImageDigest(),
-	}
-
-	postTransforms := []jobtransform.PostTransformer{
-		jobtransform.NewWasmStorageSpecConverter(),
-	}
-
 	return &BaseEndpoint{
-		id:             params.ID,
-		computesvc:     params.ComputeEndpoint,
-		store:          params.Store,
-		transforms:     transforms,
-		postTransforms: postTransforms,
-		eventEmitter:   params.EventEmitter,
-	}
-}
-
-func (e *BaseEndpoint) SubmitJob(ctx context.Context, data model.JobCreatePayload) (*model.Job, error) {
-	jobUUID, err := uuid.NewRandom()
-	if err != nil {
-		return &model.Job{}, fmt.Errorf("error creating job id: %w", err)
-	}
-	jobID := jobUUID.String()
-
-	// Creates a new root context to track a job's lifecycle for tracing. This
-	// should be fine as only one node will call SubmitJob(...) - the other
-	// nodes will hear about the job via events on the transport.
-	ctx, span := system.NewSpan(ctx, system.GetTracer(), "pkg/requester.BaseEndpoint.SubmitJob",
-		// job lifecycle spans go in their own, dedicated trace
-		trace.WithNewRoot(),
-		trace.WithLinks(trace.LinkFromContext(ctx)), // link to any api traces
-		trace.WithSpanKind(trace.SpanKindInternal),
-		trace.WithAttributes(
-			attribute.String(model.TracerAttributeNameNodeID, e.id),
-			attribute.String(model.TracerAttributeNameJobID, jobID),
-		),
-	)
-	defer span.End()
-
-	// TODO: Should replace the span above, with the below, but I don't understand how/why we're tracing contexts in a variable.
-	// Specifically tracking them all in ctrl.jobContexts
-	// ctx, span := system.NewRootSpan(ctx, system.GetTracer(), "pkg/controller.SubmitJob")
-	// defer span.End()
-
-	now := time.Now().UTC()
-	legacyJob := &model.Job{
-		APIVersion: data.APIVersion,
-		Metadata: model.Metadata{
-			ID:        jobID,
-			ClientID:  data.ClientID,
-			CreatedAt: now,
-		},
-		Spec: *data.Spec,
-	}
-
-	for _, transform := range e.transforms {
-		_, err = transform(ctx, legacyJob)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	// convert to new job model
-	job, err := legacy.FromLegacyJob(legacyJob)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, transform := range e.postTransforms {
-		_, err = transform(ctx, job)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	JobsSubmitted.Inc(ctx, job.MetricAttributes()...)
-
-	err = e.store.CreateJob(ctx, *job, orchestrator.JobSubmittedEvent())
-	if err != nil {
-		return nil, err
-	}
-
-	eval := &models.Evaluation{
-		ID:          uuid.NewString(),
-		JobID:       job.ID,
-		TriggeredBy: models.EvalTriggerJobRegister,
-		Type:        job.Type,
-		Status:      models.EvalStatusPending,
-		CreateTime:  job.CreateTime,
-		ModifyTime:  job.CreateTime,
+		id:           params.ID,
+		store:        params.Store,
+		eventEmitter: params.EventEmitter,
 	}
-
-	// TODO(ross): How can we create this evaluation in the same transaction that the CreateJob
-	// call uses.
-	err = e.store.CreateEvaluation(ctx, *eval)
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msgf("failed to save evaluation for job %s", jobID)
-		return nil, err
-	}
-
-	e.eventEmitter.EmitJobCreated(ctx, *job)
-	return legacyJob, nil
-}
-
-func (e *BaseEndpoint) CancelJob(ctx context.Context, request CancelJobRequest) (CancelJobResult, error) {
-	job, err := e.store.GetJob(ctx, request.JobID)
-	if err != nil {
-		return CancelJobResult{}, err
-	}
-	switch job.State.StateType {
-	case models.JobStateTypeStopped:
-		// no need to cancel a job that is already stopped
-		return CancelJobResult{}, nil
-	case models.JobStateTypeCompleted:
-		return CancelJobResult{}, fmt.Errorf("cannot cancel job in state %s", job.State.StateType)
-	}
-
-	// update the job state, except if the job is already completed
-	// we allow marking a failed job as canceled
-	err = e.store.UpdateJobState(ctx, jobstore.UpdateJobStateRequest{
-		JobID: request.JobID,
-		Condition: jobstore.UpdateJobCondition{
-			UnexpectedStates: []models.JobStateType{
-				models.JobStateTypeCompleted,
-			},
-		},
-		NewState: models.JobStateTypeStopped,
-		Event:    orchestrator.JobStoppedEvent("job canceled by user"),
-	})
-	if err != nil {
-		return CancelJobResult{}, err
-	}
-
-	// enqueue evaluation to allow the scheduler to cancel existing executions
-	// if the job is not terminal already, such as failed
-	if !job.IsTerminal() {
-		now := time.Now().UTC().UnixNano()
-		eval := &models.Evaluation{
-			ID:          uuid.NewString(),
-			JobID:       request.JobID,
-			TriggeredBy: models.EvalTriggerJobCancel,
-			Type:        job.Type,
-			Status:      models.EvalStatusPending,
-			CreateTime:  now,
-			ModifyTime:  now,
-		}
-
-		// TODO(ross): How can we create this evaluation in the same transaction that we update the jobstate
-		err = e.store.CreateEvaluation(ctx, *eval)
-		if err != nil {
-			log.Ctx(ctx).Error().Err(err).Msgf("failed to save evaluation for cancel job %s", request.JobID)
-			return CancelJobResult{}, err
-		}
-	}
-	e.eventEmitter.EmitEventSilently(ctx, model.JobEvent{
-		JobID:     request.JobID,
-		EventName: model.JobEventCanceled,
-		Status:    request.Reason,
-		EventTime: time.Now(),
-	})
-	return CancelJobResult{}, nil
 }
 
 // /////////////////////////////
@@ -362,5 +187,4 @@ func (e *BaseEndpoint) enqueueEvaluation(ctx context.Context, jobID, operation s
 }
 
 // Compile-time interface check:
-var _ Endpoint = (*BaseEndpoint)(nil)
 var _ compute.Callback = (*BaseEndpoint)(nil)
diff --git a/pkg/requester/jobtransform/default_timeout.go b/pkg/requester/jobtransform/default_timeout.go
deleted file mode 100644
index 288e39b75c..0000000000
--- a/pkg/requester/jobtransform/default_timeout.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-	"time"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-)
-
-// Sets a default timeout value if one is not passed or below an acceptable value
-func NewTimeoutApplier(minTimeout, defaultTimeout time.Duration) Transformer {
-	return func(ctx context.Context, job *model.Job) (modified bool, err error) {
-		if job.Spec.GetTimeout() <= minTimeout {
-			job.Spec.Timeout = int64(defaultTimeout.Seconds())
-			return true, nil
-		}
-		return
-	}
-}
diff --git a/pkg/requester/jobtransform/migrate_engine.go b/pkg/requester/jobtransform/migrate_engine.go
deleted file mode 100644
index d4fd6c77ac..0000000000
--- a/pkg/requester/jobtransform/migrate_engine.go
+++ /dev/null
@@ -1,98 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-)
-
-// NewEngineMigrator maintains backward compatibility for jobs that were defined with Engine/Wasm/Docker and EngineSpec.
-func NewEngineMigrator() Transformer {
-	return func(ctx context.Context, job *model.Job) (bool, error) {
-		// if we received an "old" job that doesn't use the new model.EngineSpec field
-		// migrate it to populate the model.EngineSpec fields.
-		//nolint:staticcheck
-		if model.IsValidEngine(job.Spec.Engine) {
-			return migrateForward(job)
-		}
-		// else we received a "new" job that uses the model.EngineSpec field, populate deprecated
-		// fields.
-		return migrateBack(job)
-	}
-}
-
-// migrateForward populates the model.EngineSpec field iff the engine is a known type.
-//
-//nolint:staticcheck
-func migrateForward(job *model.Job) (bool, error) {
-	switch job.Spec.Engine {
-	case model.EngineNoop:
-		job.Spec.EngineSpec = model.NewEngineBuilder().
-			WithType(model.EngineNoop.String()).
-			Build()
-		return true, nil
-	case model.EngineDocker:
-		job.Spec.EngineSpec = model.NewDockerEngineBuilder(job.Spec.Docker.Image).
-			WithEntrypoint(job.Spec.Docker.Entrypoint...).
-			WithParameters(job.Spec.Docker.Parameters...).
-			WithWorkingDirectory(job.Spec.Docker.WorkingDirectory).
-			WithEnvironmentVariables(job.Spec.Docker.EnvironmentVariables...).
-			Build()
-		return true, nil
-	case model.EngineWasm:
-		job.Spec.EngineSpec = model.NewWasmEngineBuilder(job.Spec.Wasm.EntryModule).
-			WithEntrypoint(job.Spec.Wasm.EntryPoint).
-			WithParameters(job.Spec.Wasm.Parameters...).
-			WithImportModules(job.Spec.Wasm.ImportModules...).
-			WithEnvironmentVariables(job.Spec.Wasm.EnvironmentVariables).
-			Build()
-		return true, nil
-	default:
-		return false, fmt.Errorf("unhandled valid engine: %s", job.Spec.Engine)
-	}
-}
-
-// migrateBack populates deprecated model.Spec engine fields iff the engine is a known type, else
-// no change is made.
-//
-//nolint:staticcheck
-func migrateBack(job *model.Job) (bool, error) {
-	// check if it's a know engine type and populate the deprecated fields.
-	switch job.Spec.EngineSpec.Engine() {
-	case model.EngineNoop:
-		job.Spec.Engine = model.EngineNoop
-		return true, nil
-	case model.EngineDocker:
-		dockerSpec, err := model.DecodeEngineSpec[model.DockerEngineSpec](job.Spec.EngineSpec)
-		if err != nil {
-			return false, fmt.Errorf("decoding docker engine spec: %w", err)
-		}
-		job.Spec.Engine = model.EngineDocker
-		job.Spec.Docker = model.JobSpecDocker{
-			Image:                dockerSpec.Image,
-			Entrypoint:           dockerSpec.Entrypoint,
-			Parameters:           dockerSpec.Parameters,
-			EnvironmentVariables: dockerSpec.EnvironmentVariables,
-			WorkingDirectory:     dockerSpec.WorkingDirectory,
-		}
-		return true, nil
-	case model.EngineWasm:
-		wasmSpec, err := model.DecodeEngineSpec[model.WasmEngineSpec](job.Spec.EngineSpec)
-		if err != nil {
-			return false, fmt.Errorf("decoding wasm engine spec: %w", err)
-		}
-		job.Spec.Engine = model.EngineWasm
-		job.Spec.Wasm = model.JobSpecWasm{
-			EntryModule:          wasmSpec.EntryModule,
-			EntryPoint:           wasmSpec.Entrypoint,
-			Parameters:           wasmSpec.Parameters,
-			EnvironmentVariables: wasmSpec.EnvironmentVariables,
-			ImportModules:        wasmSpec.ImportModules,
-		}
-		return true, nil
-	default:
-		// received a non-standard engine type, we cannot migrate this, exciting!
-		return false, nil
-	}
-}
diff --git a/pkg/requester/jobtransform/migrate_publisher.go b/pkg/requester/jobtransform/migrate_publisher.go
deleted file mode 100644
index 3d32d0e9cf..0000000000
--- a/pkg/requester/jobtransform/migrate_publisher.go
+++ /dev/null
@@ -1,42 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/rs/zerolog/log"
-)
-
-// Maintains backward compatibility for jobs that were defined with Publisher and PublisherSpec
-func NewPublisherMigrator(defaultPublisher string) Transformer {
-	var err error
-	var pubSpec model.PublisherSpec
-
-	if defaultPublisher != "" {
-		pubSpec, err = legacy_job.RawPublisherStringToPublisherSpec(defaultPublisher)
-		if err != nil {
-			log.Error().
-				Err(err).
-				Str("Publisher", defaultPublisher).
-				Msg("Failed to parse default publisher: using noop publisher instead")
-		}
-	}
-
-	return func(ctx context.Context, job *model.Job) (modified bool, err error) {
-		// Only set the default publisher if we have one defined and we are presented
-		// with a noop publisher (which is used when the user does not want to publish)
-		if job.Spec.PublisherSpec.Type == model.PublisherNoop && pubSpec.Type != model.PublisherNoop {
-			job.Spec.PublisherSpec = pubSpec
-		}
-
-		if model.IsValidPublisher(job.Spec.PublisherSpec.Type) {
-			//nolint:staticcheck
-			job.Spec.Publisher = job.Spec.PublisherSpec.Type
-		} else {
-			//nolint:staticcheck
-			job.Spec.PublisherSpec.Type = job.Spec.Publisher
-		}
-		return true, nil
-	}
-}
diff --git a/pkg/requester/jobtransform/requester_info.go b/pkg/requester/jobtransform/requester_info.go
deleted file mode 100644
index 7da5b0aae0..0000000000
--- a/pkg/requester/jobtransform/requester_info.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-)
-
-func NewRequesterInfo(requesterNodeID string) Transformer {
-	return func(ctx context.Context, j *model.Job) (modified bool, err error) {
-		j.Metadata.Requester = model.JobRequester{
-			RequesterNodeID: requesterNodeID,
-		}
-		return true, nil
-	}
-}
diff --git a/pkg/requester/jobtransform/types.go b/pkg/requester/jobtransform/types.go
deleted file mode 100644
index b9035f02e0..0000000000
--- a/pkg/requester/jobtransform/types.go
+++ /dev/null
@@ -1,14 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-)
-
-// Transformer is a function that can be used to modify a job before it is migrated to a new version.
-type Transformer func(context.Context, *model.Job) (modified bool, err error)
-
-// PostTransformer is a function that can be used to modify a job after it has been migrated to a new version.
-type PostTransformer func(context.Context, *models.Job) (modified bool, err error)
diff --git a/pkg/requester/jobtransform/wasm_storage_spec.go b/pkg/requester/jobtransform/wasm_storage_spec.go
deleted file mode 100644
index 92883d5504..0000000000
--- a/pkg/requester/jobtransform/wasm_storage_spec.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package jobtransform
-
-import (
-	"context"
-	"fmt"
-
-	wasm "github.com/bacalhau-project/bacalhau/pkg/executor/wasm/models"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-)
-
-// NewWasmStorageSpecConverter returns a job transformer that converts the
-// entry and import modules of wasm engine spec from the legacy storage spec to
-// input source definitions.
-func NewWasmStorageSpecConverter() PostTransformer {
-	return func(ctx context.Context, j *models.Job) (modified bool, err error) {
-		engineSpec := j.Task().Engine
-		if engineSpec.IsType(models.EngineWasm) {
-			wasmEngineSpec, err := wasm.DecodeLegacySpec(engineSpec)
-			if err != nil {
-				return false, fmt.Errorf("failed to decode wasm engine spec. %w", err)
-			}
-			engineSpec.Params = wasmEngineSpec.ToMap()
-			return true, nil
-		}
-		return false, nil
-	}
-}
diff --git a/pkg/setup/setup.go b/pkg/setup/setup.go
index 57e469d4ca..d1633426c5 100644
--- a/pkg/setup/setup.go
+++ b/pkg/setup/setup.go
@@ -57,9 +57,6 @@ func SetupBacalhauRepo(repoDir string, c config.ReadWriter) (*repo.FsRepo, error
 }
 
 func SetupBacalhauRepoForTesting(t testing.TB) (*repo.FsRepo, types.BacalhauConfig) {
-	if err := os.Setenv("REQUESTER_ENDPOINT_USE_DEPRECATED_ENV", "true"); err != nil {
-		t.Fatal(err)
-	}
 	// reset the global viper instance used by cmds pkg.
 	viper.Reset()
 	// create a specific viper instance for testing


Test patch:
diff --git a/cmd/cli/serve/serve_test.go b/cmd/cli/serve/serve_test.go
index 98bb9befb5..61f67dbff9 100644
--- a/cmd/cli/serve/serve_test.go
+++ b/cmd/cli/serve/serve_test.go
@@ -15,7 +15,8 @@ import (
 	"github.com/stretchr/testify/suite"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	apitest "github.com/bacalhau-project/bacalhau/pkg/publicapi/test"
 
@@ -26,7 +27,6 @@ import (
 	cmd2 "github.com/bacalhau-project/bacalhau/cmd/cli"
 	cfgtypes "github.com/bacalhau-project/bacalhau/pkg/config/types"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
 	"github.com/bacalhau-project/bacalhau/pkg/setup"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
 	"github.com/bacalhau-project/bacalhau/pkg/types"
@@ -200,16 +200,28 @@ func (s *ServeSuite) TestCanSubmitJob() {
 	docker.MustHaveDocker(s.T())
 	port, err := s.serve("--node-type", "requester,compute")
 	s.Require().NoError(err)
-	client, err := client.NewAPIClient(client.NoTLS, s.config.User, "localhost", port)
-	s.Require().NoError(err)
-
-	clientV2 := clientv2.New(fmt.Sprintf("http://127.0.0.1:%d", port))
-	s.Require().NoError(apitest.WaitForAlive(s.ctx, clientV2))
-
-	job, err := model.NewJobWithSaneProductionDefaults()
-	s.Require().NoError(err)
 
-	_, err = client.Submit(s.ctx, job)
+	client := clientv2.New(fmt.Sprintf("http://127.0.0.1:%d", port))
+	s.Require().NoError(apitest.WaitForAlive(s.ctx, client))
+
+	job := &models.Job{
+		Name:  s.T().Name(),
+		Type:  models.JobTypeBatch,
+		Count: 1,
+		Tasks: []*models.Task{
+			{
+				Name: s.T().Name(),
+				Engine: &models.SpecConfig{
+					Type:   models.EngineNoop,
+					Params: make(map[string]interface{}),
+				},
+			},
+		},
+	}
+	job.Normalize()
+	_, err = client.Jobs().Put(s.ctx, &apimodels.PutJobRequest{
+		Job: job,
+	})
 	s.NoError(err)
 }
 
diff --git a/cmd/cli/serve/timeout_test.go b/cmd/cli/serve/timeout_test.go
index 31ca19ede5..3546204f9a 100644
--- a/cmd/cli/serve/timeout_test.go
+++ b/cmd/cli/serve/timeout_test.go
@@ -7,9 +7,9 @@ import (
 	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/docker"
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
 	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	apitest "github.com/bacalhau-project/bacalhau/pkg/publicapi/test"
 )
@@ -53,26 +53,44 @@ func (s *ServeSuite) TestNoTimeoutSetOrApplied() {
 
 			port, err := s.serve(args...)
 			s.Require().NoError(err)
-			client, err := client.NewAPIClient(client.NoTLS, s.config.User, "localhost", port)
-			s.Require().NoError(err)
+
 			clientV2 := clientv2.New(fmt.Sprintf("http://127.0.0.1:%d", port))
 			s.Require().NoError(apitest.WaitForAlive(s.ctx, clientV2))
 
-			testJob := model.NewJob()
-			specOpts := []legacy_job.SpecOpt{}
+			testJob := &models.Job{
+				Name:  s.T().Name(),
+				Type:  models.JobTypeBatch,
+				Count: 1,
+				Tasks: []*models.Task{
+					{
+						Name: s.T().Name(),
+						Engine: &models.SpecConfig{
+							Type:   models.EngineNoop,
+							Params: make(map[string]interface{}),
+						},
+						Publisher: &models.SpecConfig{
+							Type:   models.PublisherNoop,
+							Params: make(map[string]interface{}),
+						},
+					},
+				},
+			}
+
 			if tc.timeoutSpecified != nil {
-				specOpts = append(specOpts, legacy_job.WithTimeout(int64(tc.timeoutSpecified.Seconds())))
+				testJob.Task().Timeouts = &models.TimeoutConfig{
+					TotalTimeout: int64(tc.timeoutSpecified.Seconds()),
+				}
 			}
-			testJob.Spec, err = legacy_job.MakeSpec(specOpts...)
-			s.Require().NoError(err)
 
-			returnedJob, err := client.Submit(s.ctx, testJob)
+			putResp, err := clientV2.Jobs().Put(s.ctx, &apimodels.PutJobRequest{
+				Job: testJob,
+			})
 			s.Require().NoError(err)
 
 			s.Eventually(func() bool {
-				jobState, err := client.GetJobState(s.ctx, returnedJob.ID())
+				getResp, err := clientV2.Jobs().Get(s.ctx, &apimodels.GetJobRequest{JobID: putResp.JobID})
 				s.Require().NoError(err)
-				s.Require().Equal(model.JobStateError.String(), jobState.State.String())
+				s.Require().Equal(models.JobStateTypeFailed, getResp.Job.State.StateType)
 				return true
 			}, 5*time.Second, 50*time.Millisecond)
 		})
diff --git a/cmd/testing/base.go b/cmd/testing/base.go
index 0b3d3e4935..982ecff3a9 100644
--- a/cmd/testing/base.go
+++ b/cmd/testing/base.go
@@ -22,7 +22,6 @@ import (
 	noop_executor "github.com/bacalhau-project/bacalhau/pkg/executor/noop"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
 	"github.com/bacalhau-project/bacalhau/pkg/node"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/setup"
 	"github.com/bacalhau-project/bacalhau/pkg/test/teststack"
@@ -31,7 +30,6 @@ import (
 type BaseSuite struct {
 	suite.Suite
 	Node            *node.Node
-	Client          *client.APIClient
 	ClientV2        clientv2.API
 	Config          types.BacalhauConfig
 	Host            string
@@ -78,8 +76,6 @@ func (s *BaseSuite) SetupTest() {
 	s.Node = stack.Nodes[0]
 	s.Host = s.Node.APIServer.Address
 	s.Port = s.Node.APIServer.Port
-	s.Client, err = client.NewAPIClient(client.NoTLS, cfg.User, s.Host, s.Port)
-	s.Require().NoError(err)
 	s.ClientV2 = clientv2.New(fmt.Sprintf("http://%s:%d", s.Host, s.Port))
 }
 
diff --git a/cmd/testing/basetls.go b/cmd/testing/basetls.go
index 7c36c2bb39..d2918237d3 100644
--- a/cmd/testing/basetls.go
+++ b/cmd/testing/basetls.go
@@ -11,7 +11,6 @@ import (
 	noop_executor "github.com/bacalhau-project/bacalhau/pkg/executor/noop"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
 	"github.com/bacalhau-project/bacalhau/pkg/node"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/setup"
 	"github.com/bacalhau-project/bacalhau/pkg/test/teststack"
@@ -69,8 +68,6 @@ func (s *BaseTLSSuite) SetupTest() {
 	s.Node = stack.Nodes[0]
 	s.Host = s.Node.APIServer.Address // NOTE: 0.0.0.0 will not work because we're testing TLS validation
 	s.Port = s.Node.APIServer.Port
-	s.Client, err = client.NewAPIClient(client.LegacyTLSSupport{UseTLS: true, Insecure: false}, cfg.User, s.Host, s.Port)
-	s.Require().NoError(err)
 	s.ClientV2 = clientv2.New(fmt.Sprintf("http://%s:%d", s.Host, s.Port), clientv2.WithTLS(true), clientv2.WithInsecureTLS(true))
 }
 
diff --git a/pkg/publicapi/endpoint/shared/endpoint_test.go b/pkg/publicapi/endpoint/shared/endpoint_test.go
index bd5f5d736c..4f046f4b0a 100644
--- a/pkg/publicapi/endpoint/shared/endpoint_test.go
+++ b/pkg/publicapi/endpoint/shared/endpoint_test.go
@@ -10,10 +10,10 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels/legacymodels"
-	"github.com/bacalhau-project/bacalhau/pkg/version"
 	"github.com/labstack/echo/v4"
 	"github.com/stretchr/testify/suite"
+
+	"github.com/bacalhau-project/bacalhau/pkg/version"
 )
 
 type EndpointSuite struct {
@@ -40,7 +40,7 @@ func (s *EndpointSuite) TestEndpointId() {
 }
 
 func (s *EndpointSuite) TestEndpointVersion() {
-	versionReq := &legacymodels.VersionRequest{ClientID: "testClient"}
+	versionReq := &VersionRequest{ClientID: "testClient"}
 	body, _ := json.Marshal(versionReq)
 
 	req := httptest.NewRequest(http.MethodPost, "/api/v1/version", bytes.NewReader(body))
@@ -49,7 +49,7 @@ func (s *EndpointSuite) TestEndpointVersion() {
 	s.router.ServeHTTP(rr, req)
 
 	expectedVersion := version.Get()
-	expectedResponse, _ := json.Marshal(legacymodels.VersionResponse{VersionInfo: expectedVersion})
+	expectedResponse, _ := json.Marshal(VersionResponse{VersionInfo: expectedVersion})
 
 	s.Equal(http.StatusOK, rr.Code)
 	s.Equal(string(expectedResponse), strings.TrimSpace(rr.Body.String()))
diff --git a/pkg/publicapi/test/requester_server_test.go b/pkg/publicapi/test/requester_server_test.go
deleted file mode 100644
index 02be261141..0000000000
--- a/pkg/publicapi/test/requester_server_test.go
+++ /dev/null
@@ -1,81 +0,0 @@
-//go:build unit || !integration
-
-package test
-
-import (
-	"context"
-	"testing"
-
-	"github.com/google/uuid"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
-
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
-
-	"github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/node"
-	testutils "github.com/bacalhau-project/bacalhau/pkg/test/utils"
-)
-
-// Define the suite, and absorb the built-in basic suite
-// functionality from testify - including a T() method which
-// returns the current testing context
-type RequesterSuite struct {
-	suite.Suite
-	node   *node.Node
-	client *client.APIClient
-}
-
-// In order for 'go test' to run this suite, we need to create
-// a normal test function and pass our suite to suite.Run
-func TestRequesterSuite(t *testing.T) {
-	suite.Run(t, new(RequesterSuite))
-}
-
-// Before each test
-func (s *RequesterSuite) SetupTest() {
-	logger.ConfigureTestLogging(s.T())
-	n, c := setupNodeForTest(s.T())
-	s.node = n
-	var err error
-	s.client, err = client.NewAPIClient(client.NoTLS, c.User, n.APIServer.Address, n.APIServer.Port)
-	s.Require().NoError(err)
-}
-
-// After each test
-func (s *RequesterSuite) TearDownTest() {
-	if s.node != nil {
-		s.node.CleanupManager.Cleanup(context.Background())
-	}
-}
-
-func (s *RequesterSuite) TestList() {
-	ctx := context.Background()
-
-	// Should have no jobs initially:
-	jobs, err := s.client.List(ctx, "", model.IncludeAny, model.ExcludeNone, 10, true, "created_at", true)
-	require.NoError(s.T(), err)
-	require.Empty(s.T(), jobs)
-
-	// Submit a random job to the node:
-	j := testutils.MakeNoopJob(s.T())
-
-	_, err = s.client.Submit(ctx, j)
-	require.NoError(s.T(), err)
-
-	// Should now have one job:
-	jobs, err = s.client.List(ctx, "", model.IncludeAny, model.ExcludeNone, 10, true, "created_at", true)
-	require.NoError(s.T(), err)
-	require.Len(s.T(), jobs, 1)
-}
-
-func (s *RequesterSuite) TestSubmitRejectsJobWithSigilHeader() {
-	j := testutils.MakeNoopJob(s.T())
-	jobID, err := uuid.NewRandom()
-	require.NoError(s.T(), err)
-
-	s.client.DefaultHeaders["X-Bacalhau-Job-ID"] = jobID.String()
-	_, err = s.client.Submit(context.Background(), j)
-	require.Error(s.T(), err)
-}
diff --git a/pkg/publicapi/test/util_test.go b/pkg/publicapi/test/util_test.go
index 00cbd1e011..1a04a25882 100644
--- a/pkg/publicapi/test/util_test.go
+++ b/pkg/publicapi/test/util_test.go
@@ -2,7 +2,6 @@ package test
 
 import (
 	"context"
-	"os"
 	"path/filepath"
 	"testing"
 
@@ -11,7 +10,6 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store/boltdb"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	boltjobstore "github.com/bacalhau-project/bacalhau/pkg/jobstore/boltdb"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/endpoint/requester"
 
 	"github.com/bacalhau-project/bacalhau/pkg/devstack"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/network"
@@ -74,9 +72,6 @@ func setupNodeForTestWithConfig(t *testing.T, apiCfg publicapi.Config) (*node.No
 		NetworkConfig:       networkConfig,
 	}
 
-	if err := os.Setenv(requester.UseDeprecatedEndpointsForTesting, "true"); err != nil {
-		t.Fatal(err)
-	}
 	n, err := node.NewNode(ctx, c, nodeConfig, repo)
 	require.NoError(t, err)
 
diff --git a/pkg/publicapi/test/websocket_test.go b/pkg/publicapi/test/websocket_test.go
deleted file mode 100644
index 8c5a316ecc..0000000000
--- a/pkg/publicapi/test/websocket_test.go
+++ /dev/null
@@ -1,115 +0,0 @@
-//go:build unit || !integration
-
-package test
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/gorilla/websocket"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
-
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
-
-	"github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/node"
-	testutils "github.com/bacalhau-project/bacalhau/pkg/test/utils"
-)
-
-// Define the suite, and absorb the built-in basic suite
-// functionality from testify - including a T() method which
-// returns the current testing context
-type WebsocketSuite struct {
-	suite.Suite
-	node   *node.Node
-	client *client.APIClient
-}
-
-// In order for 'go test' to run this suite, we need to create
-// a normal test function and pass our suite to suite.Run
-func TestWebsocketSuite(t *testing.T) {
-	suite.Run(t, new(WebsocketSuite))
-}
-
-// Before each test
-func (s *WebsocketSuite) SetupTest() {
-	logger.ConfigureTestLogging(s.T())
-	n, c := setupNodeForTest(s.T())
-	s.node = n
-	var err error
-	s.client, err = client.NewAPIClient(client.NoTLS, c.User, n.APIServer.Address, n.APIServer.Port)
-	s.Require().NoError(err)
-}
-
-// After each test
-func (s *WebsocketSuite) TearDownTest() {
-	s.node.CleanupManager.Cleanup(context.Background())
-}
-
-func (s *WebsocketSuite) TestWebsocketEverything() {
-	ctx := context.Background()
-	// string.Replace http with ws in c.BaseURI
-	url := *s.client.BaseURI
-	url.Scheme = "ws"
-	wurl := url.JoinPath("api", "v1", "requester", "websocket", "events")
-
-	conn, _, err := websocket.DefaultDialer.Dial(wurl.String(), nil)
-	require.NoError(s.T(), err)
-	s.T().Cleanup(func() {
-		s.NoError(conn.Close())
-	})
-
-	eventChan := make(chan model.JobEvent)
-	go func() {
-		defer close(eventChan)
-		for {
-			var event model.JobEvent
-			err = conn.ReadJSON(&event)
-			if err != nil && strings.Contains(err.Error(), net.ErrClosed.Error()) {
-				return
-			}
-			require.NoError(s.T(), err)
-			eventChan <- event
-		}
-	}()
-
-	// Pause to ensure the websocket connects _before_ we submit the job
-	time.Sleep(100 * time.Millisecond)
-
-	genericJob := testutils.MakeNoopJob(s.T())
-	_, err = s.client.Submit(ctx, genericJob)
-	require.NoError(s.T(), err)
-
-	event := <-eventChan
-	require.Equal(s.T(), model.JobEventCreated, event.EventName)
-
-}
-
-func (s *WebsocketSuite) TestWebsocketSingleJob() {
-	s.T().Skip("TODO: test is flaky as by the time we connect to the websocket, " +
-		"the job has already progressed and first event is not guaranteed to be 'Created'")
-	ctx := context.Background()
-
-	genericJob := testutils.MakeNoopJob(s.T())
-	j, err := s.client.Submit(ctx, genericJob)
-	require.NoError(s.T(), err)
-
-	url := *s.client.BaseURI
-	url.Scheme = "ws"
-	wurl := url.JoinPath("websocket", "events")
-	wurl.RawQuery = fmt.Sprintf("job_id=%s", j.Metadata.ID)
-
-	conn, _, err := websocket.DefaultDialer.Dial(wurl.String(), nil)
-	require.NoError(s.T(), err)
-
-	var event model.JobEvent
-	err = conn.ReadJSON(&event)
-	require.NoError(s.T(), err)
-	require.Equal(s.T(), "Created", event.EventName.String())
-}
diff --git a/pkg/requester/jobtransform/migrate_engine_test.go b/pkg/requester/jobtransform/migrate_engine_test.go
deleted file mode 100644
index 793b7167ee..0000000000
--- a/pkg/requester/jobtransform/migrate_engine_test.go
+++ /dev/null
@@ -1,206 +0,0 @@
-//go:build unit || !integration
-
-package jobtransform
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-)
-
-func TestEngineMigrator(t *testing.T) {
-	ctx := context.Background()
-	engineMigrator := NewEngineMigrator()
-
-	t.Run("migrate docker", func(t *testing.T) {
-		image := "my_image"
-		entrypoint := []string{"sh", "-c"}
-		envVars := []string{"VAR1=value1", "VAR2=value2"}
-		workingDir := "/app"
-		params := []string{"1", "2", "3"}
-		verify := func(t *testing.T, job model.Job) {
-			assert.Equal(t, model.EngineDocker, job.Spec.Engine)
-			assert.Equal(t, image, job.Spec.Docker.Image)
-			assert.Equal(t, params, job.Spec.Docker.Parameters)
-			assert.Equal(t, entrypoint, job.Spec.Docker.Entrypoint)
-			assert.Equal(t, workingDir, job.Spec.Docker.WorkingDirectory)
-			assert.Equal(t, envVars, job.Spec.Docker.EnvironmentVariables)
-
-			dockerSpec, err := model.DecodeEngineSpec[model.DockerEngineSpec](job.Spec.EngineSpec)
-			require.NoError(t, err)
-
-			assert.Equal(t, model.EngineDocker.String(), job.Spec.EngineSpec.Type)
-			assert.Equal(t, image, dockerSpec.Image)
-			assert.Equal(t, params, dockerSpec.Parameters)
-			assert.Equal(t, entrypoint, dockerSpec.Entrypoint)
-			assert.Equal(t, workingDir, dockerSpec.WorkingDirectory)
-			assert.Equal(t, envVars, dockerSpec.EnvironmentVariables)
-		}
-		t.Run("down", func(t *testing.T) {
-			dockerJob := model.Job{
-				Spec: model.Spec{
-					EngineSpec: model.NewDockerEngineBuilder(image).
-						WithEntrypoint(entrypoint...).
-						WithParameters(params...).
-						WithWorkingDirectory(workingDir).
-						WithEnvironmentVariables(envVars...).
-						Build(),
-				},
-			}
-			modified, err := engineMigrator(ctx, &dockerJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, dockerJob)
-
-		})
-		t.Run("up", func(t *testing.T) {
-			dockerJob := model.Job{
-				Spec: model.Spec{
-					Engine: model.EngineDocker,
-					Docker: model.JobSpecDocker{
-						Image:                image,
-						Entrypoint:           entrypoint,
-						Parameters:           params,
-						EnvironmentVariables: envVars,
-						WorkingDirectory:     workingDir,
-					},
-				},
-			}
-			modified, err := engineMigrator(ctx, &dockerJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, dockerJob)
-		})
-	})
-	t.Run("migrate wasm", func(t *testing.T) {
-		entrypoint := "_start"
-		parameters := []string{"arg1", "arg2"}
-		envVars := map[string]string{"VAR1": "value1", "VAR2": "value2"}
-		entryModule := model.StorageSpec{
-			StorageSource: model.StorageSourceS3,
-			Name:          "w2b",
-		}
-		importModules := []model.StorageSpec{
-			{
-				StorageSource: model.StorageSourceIPFS,
-				Name:          "w3b",
-			},
-			{
-				Name:          "w4b",
-				StorageSource: model.StorageSourceIPFS,
-			},
-		}
-		verify := func(t *testing.T, job model.Job) {
-			assert.Equal(t, model.EngineWasm, job.Spec.Engine)
-			assert.Equal(t, entrypoint, job.Spec.Wasm.EntryPoint)
-			assert.Equal(t, parameters, job.Spec.Wasm.Parameters)
-			assert.Equal(t, entryModule, job.Spec.Wasm.EntryModule)
-			assert.Equal(t, importModules, job.Spec.Wasm.ImportModules)
-			assert.Equal(t, envVars, job.Spec.Wasm.EnvironmentVariables)
-
-			wasmSpec, err := model.DecodeEngineSpec[model.WasmEngineSpec](job.Spec.EngineSpec)
-			require.NoError(t, err)
-
-			assert.Equal(t, model.EngineWasm.String(), job.Spec.EngineSpec.Type)
-			assert.Equal(t, entrypoint, wasmSpec.Entrypoint)
-			assert.Equal(t, parameters, wasmSpec.Parameters)
-			assert.Equal(t, entryModule, wasmSpec.EntryModule)
-			assert.Equal(t, importModules, wasmSpec.ImportModules)
-			assert.Equal(t, envVars, wasmSpec.EnvironmentVariables)
-		}
-
-		t.Run("down", func(t *testing.T) {
-			wasmJob := model.Job{
-				Spec: model.Spec{
-					EngineSpec: model.NewWasmEngineBuilder(entryModule).
-						WithEntrypoint(entrypoint).
-						WithParameters(parameters...).
-						WithImportModules(importModules...).
-						WithEnvironmentVariables(envVars).
-						Build(),
-				},
-			}
-			modified, err := engineMigrator(ctx, &wasmJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, wasmJob)
-		})
-		t.Run("up", func(t *testing.T) {
-			wasmJob := model.Job{
-				Spec: model.Spec{
-					Engine: model.EngineWasm,
-					Wasm: model.JobSpecWasm{
-						EntryModule:          entryModule,
-						EntryPoint:           entrypoint,
-						Parameters:           parameters,
-						EnvironmentVariables: envVars,
-						ImportModules:        importModules,
-					},
-				},
-			}
-			modified, err := engineMigrator(ctx, &wasmJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, wasmJob)
-		})
-	})
-	t.Run("migrate noop", func(t *testing.T) {
-		verify := func(t *testing.T, job model.Job) {
-			assert.Equal(t, model.EngineNoop, job.Spec.Engine)
-			assert.Equal(t, model.EngineNoop.String(), job.Spec.EngineSpec.Type)
-			assert.Empty(t, job.Spec.EngineSpec.Params)
-		}
-		t.Run("down", func(t *testing.T) {
-			noopJob := model.Job{
-				Spec: model.Spec{
-					EngineSpec: model.NewEngineBuilder().
-						WithType(model.EngineNoop.String()).
-						Build(),
-				},
-			}
-			modified, err := engineMigrator(ctx, &noopJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, noopJob)
-		})
-		t.Run("up", func(t *testing.T) {
-			noopJob := model.Job{
-				Spec: model.Spec{
-					Engine: model.EngineNoop,
-				},
-			}
-			modified, err := engineMigrator(ctx, &noopJob)
-			require.NoError(t, err)
-			require.True(t, modified)
-			verify(t, noopJob)
-		})
-	})
-	t.Run("do not migrate job", func(t *testing.T) {
-		expectedType := "TestEngineSpec"
-		expectedName := "an_engine_name"
-		expectedColors := []string{"red", "blue", "green"}
-
-		spec := model.NewEngineBuilder().
-			WithType(expectedType).
-			WithParam("Name", expectedName).
-			WithParam("Colors", expectedColors).
-			Build()
-		job := model.Job{
-			Spec: model.Spec{
-				EngineSpec: spec,
-			},
-		}
-
-		modified, err := engineMigrator(ctx, &job)
-		require.NoError(t, err)
-		require.False(t, modified)
-
-		assert.Equal(t, expectedType, spec.Type, "Expected Type to be '%s', got '%s'", expectedType, spec.Type)
-		assert.Equal(t, expectedName, spec.Params["Name"], "Expected Name to be '%s', got '%s'", expectedName, spec.Params["Name"])
-		assert.Equal(t, expectedColors, spec.Params["Colors"], "Expected Color to be '%s', got '%s'", expectedColors, spec.Params["Color"])
-	})
-}
diff --git a/pkg/test/compute/resourcelimits_test.go b/pkg/test/compute/resourcelimits_test.go
index 7faffbba35..2252afb320 100644
--- a/pkg/test/compute/resourcelimits_test.go
+++ b/pkg/test/compute/resourcelimits_test.go
@@ -16,22 +16,18 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
-
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-
 	"github.com/bacalhau-project/bacalhau/pkg/devstack"
 	noop_executor "github.com/bacalhau-project/bacalhau/pkg/executor/noop"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/node"
+	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/setup"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
+	"github.com/bacalhau-project/bacalhau/pkg/test/scenario"
 	"github.com/bacalhau-project/bacalhau/pkg/test/teststack"
-	testutils "github.com/bacalhau-project/bacalhau/pkg/test/utils"
 	nodeutils "github.com/bacalhau-project/bacalhau/pkg/test/utils/node"
 )
 
@@ -342,21 +338,39 @@ func (suite *ComputeNodeResourceLimitsSuite) TestParallelGPU() {
 	// for the requester node to pick up the nodeInfo messages
 	nodeutils.WaitForNodeDiscovery(suite.T(), stack.Nodes[0].RequesterNode, nodeCount)
 
-	jobConfig := testutils.MakeJobWithOpts(suite.T(),
-		legacy_job.WithResources("", "", "", "1"),
-	)
+	job := &models.Job{
+		Name:  suite.T().Name(),
+		Type:  models.JobTypeBatch,
+		Count: 1,
+		Tasks: []*models.Task{
+			{
+				Name: suite.T().Name(),
+				Engine: &models.SpecConfig{
+					Type:   models.EngineNoop,
+					Params: make(map[string]interface{}),
+				},
+				Publisher: &models.SpecConfig{
+					Type:   models.PublisherNoop,
+					Params: make(map[string]interface{}),
+				},
+				ResourcesConfig: &models.ResourcesConfig{
+					CPU:    "",
+					Memory: "",
+					Disk:   "",
+					GPU:    "1",
+				},
+			},
+		},
+	}
+	job.Normalize()
 
-	resolver := legacy.NewStateResolver(stack.Nodes[0].RequesterNode.JobStore)
+	resolver := scenario.NewStateResolverFromStore(stack.Nodes[0].RequesterNode.JobStore)
 
 	for i := 0; i < nodeCount; i++ {
 		for j := 0; j < jobsPerNode; j++ {
-			submittedJob, err := stack.Nodes[0].RequesterNode.Endpoint.SubmitJob(ctx, model.JobCreatePayload{
-				ClientID:   "123",
-				APIVersion: jobConfig.APIVersion,
-				Spec:       &jobConfig.Spec,
-			})
+			submittedJob, err := stack.Nodes[0].RequesterNode.Endpoint.SubmitJob(ctx, &orchestrator.SubmitJobRequest{Job: job})
 			require.NoError(suite.T(), err)
-			jobIds = append(jobIds, submittedJob.Metadata.ID)
+			jobIds = append(jobIds, submittedJob.JobID)
 
 			// sleep a bit here to simulate jobs being sumbmitted over time
 			// and to give time for compute nodes to accept and run the jobs
@@ -367,7 +381,7 @@ func (suite *ComputeNodeResourceLimitsSuite) TestParallelGPU() {
 	}
 
 	for _, jobId := range jobIds {
-		err := resolver.WaitUntilComplete(ctx, jobId)
+		err = resolver.Wait(ctx, jobId, scenario.WaitForSuccessfulCompletion())
 		require.NoError(suite.T(), err)
 	}
 
@@ -376,11 +390,11 @@ func (suite *ComputeNodeResourceLimitsSuite) TestParallelGPU() {
 	allocationMap := map[string]int{}
 
 	for _, jobId := range jobIds {
-		jobState, err := resolver.GetJobState(ctx, jobId)
+		jobState, err := resolver.JobState(ctx, jobId)
 		require.NoError(suite.T(), err)
-		completedExecutionStates := legacy_job.GetCompletedExecutionStates(jobState)
+		completedExecutionStates := scenario.GetCompletedExecutionStates(jobState)
 		require.Equal(suite.T(), 1, len(completedExecutionStates))
-		require.Equal(suite.T(), model.ExecutionStateCompleted, completedExecutionStates[0].State)
+		require.Equal(suite.T(), models.ExecutionStateCompleted, completedExecutionStates[0].ComputeState.StateType)
 		allocationMap[completedExecutionStates[0].NodeID]++
 	}
 
diff --git a/pkg/test/devstack/default_publisher_test.go b/pkg/test/devstack/default_publisher_test.go
index 1d42f3c828..d8662debdd 100644
--- a/pkg/test/devstack/default_publisher_test.go
+++ b/pkg/test/devstack/default_publisher_test.go
@@ -53,7 +53,6 @@ func (s *DefaultPublisherSuite) TestNoDefaultPublisher() {
 }
 
 func (s *DefaultPublisherSuite) TestDefaultPublisher() {
-	// we are skipping this test because the orchestrator endpoint doesn't require a publisher
 	stack := scenario.StackConfig{}
 	stack.DefaultPublisher = "local"
 
diff --git a/pkg/test/devstack/submit_test.go b/pkg/test/devstack/submit_test.go
index cf1bffe921..70adf0239a 100644
--- a/pkg/test/devstack/submit_test.go
+++ b/pkg/test/devstack/submit_test.go
@@ -4,13 +4,15 @@ package devstack
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/repo"
 
-	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/bacalhau-project/bacalhau/pkg/devstack"
@@ -18,7 +20,6 @@ import (
 	testutils "github.com/bacalhau-project/bacalhau/pkg/test/teststack"
 
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
 )
 
 type DevstackSubmitSuite struct {
@@ -47,17 +48,29 @@ func (suite *DevstackSubmitSuite) TestEmptySpec() {
 	)
 
 	apiServer := stack.Nodes[0].APIServer
-	apiClient, err := client.NewAPIClient(client.NoTLS, suite.Config.User, apiServer.Address, apiServer.Port)
-	suite.Require().NoError(err)
+	apiClient := clientv2.New(fmt.Sprintf("http://%s:%d", apiServer.Address, apiServer.Port))
 
-	j := &model.Job{}
-	j.Spec.Deal = model.Deal{Concurrency: 1}
-	_, missingSpecError := apiClient.Submit(ctx, j)
+	j := &models.Job{}
+	j.Count = 1
+	resp, err := apiClient.Jobs().Put(ctx, &apimodels.PutJobRequest{
+		Job: j,
+	})
 
-	require.Error(suite.T(), missingSpecError)
+	suite.Require().Error(err)
+	suite.Require().Empty(resp)
 
-	j = &model.Job{}
-	j.Spec = model.Spec{EngineSpec: model.NewEngineBuilder().WithType(model.EngineDocker.String()).Build()}
-	_, missingDealError := apiClient.Submit(ctx, j)
-	require.Error(suite.T(), missingDealError)
+	j = &models.Job{}
+	j.Tasks = []*models.Task{
+		{
+			Engine: &models.SpecConfig{
+				Type:   models.EngineDocker,
+				Params: make(map[string]interface{}),
+			},
+		},
+	}
+	resp, err = apiClient.Jobs().Put(ctx, &apimodels.PutJobRequest{
+		Job: j,
+	})
+	suite.Require().Error(err)
+	suite.Require().Empty(resp)
 }
diff --git a/pkg/test/logstream/stream_address_test.go b/pkg/test/logstream/stream_address_test.go
index 18fdd9c27e..c6a16cf1ac 100644
--- a/pkg/test/logstream/stream_address_test.go
+++ b/pkg/test/logstream/stream_address_test.go
@@ -77,7 +77,7 @@ func (s *LogStreamTestSuite) TestStreamAddress() {
 	err = node.RequesterNode.JobStore.CreateExecution(s.ctx, *execution, models.Event{})
 	require.NoError(s.T(), err)
 
-	ch, err := node.RequesterNode.EndpointV2.ReadLogs(s.ctx, orchestrator.ReadLogsRequest{
+	ch, err := node.RequesterNode.Endpoint.ReadLogs(s.ctx, orchestrator.ReadLogsRequest{
 		JobID:       job.ID,
 		ExecutionID: execution.ID,
 		Tail:        true,
diff --git a/pkg/test/requester/node_selection_test.go b/pkg/test/requester/node_selection_test.go
index 48f655a6c4..a5b4c2de8d 100644
--- a/pkg/test/requester/node_selection_test.go
+++ b/pkg/test/requester/node_selection_test.go
@@ -80,7 +80,7 @@ func (s *NodeSelectionSuite) SetupSuite() {
 	s.compute3 = stack.Nodes[3]
 	s.api = clientv2.New(fmt.Sprintf("http://%s:%d", s.requester.APIServer.Address, s.requester.APIServer.Port))
 	s.Require().NoError(err)
-	s.stateResolver = scenario.NewStateResolver(s.api)
+	s.stateResolver = scenario.NewStateResolverFromAPI(s.api)
 	s.computeNodes = []*node.Node{s.compute1, s.compute2, s.compute3}
 
 	nodeutils.WaitForNodeDiscovery(s.T(), s.requester.RequesterNode, 4)
diff --git a/pkg/test/requester/retries_test.go b/pkg/test/requester/retries_test.go
index 2bd4cb9913..efba91eba6 100644
--- a/pkg/test/requester/retries_test.go
+++ b/pkg/test/requester/retries_test.go
@@ -15,8 +15,10 @@ import (
 	"golang.org/x/exp/slices"
 	"k8s.io/apimachinery/pkg/selection"
 
-	"github.com/bacalhau-project/bacalhau/pkg/models/migration/legacy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
+	"github.com/bacalhau-project/bacalhau/pkg/test/scenario"
 
 	"github.com/bacalhau-project/bacalhau/pkg/bidstrategy"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/math"
@@ -25,12 +27,9 @@ import (
 
 	"github.com/bacalhau-project/bacalhau/pkg/devstack"
 	noop_executor "github.com/bacalhau-project/bacalhau/pkg/executor/noop"
-	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
 	"github.com/bacalhau-project/bacalhau/pkg/node"
 	noop_publisher "github.com/bacalhau-project/bacalhau/pkg/publisher/noop"
-	testutils "github.com/bacalhau-project/bacalhau/pkg/test/utils"
 	nodeutils "github.com/bacalhau-project/bacalhau/pkg/test/utils/node"
 )
 
@@ -43,8 +42,8 @@ type RetriesSuite struct {
 	suite.Suite
 	requester     *node.Node
 	computeNodes  []*node.Node
-	client        *client.APIClient
-	stateResolver *legacy_job.StateResolver
+	client        clientv2.API
+	stateResolver *scenario.StateResolver
 }
 
 func (s *RetriesSuite) SetupSuite() {
@@ -144,9 +143,9 @@ func (s *RetriesSuite) SetupSuite() {
 	)
 
 	s.requester = stack.Nodes[0]
-	s.client, err = client.NewAPIClient(client.NoTLS, cfg.User, s.requester.APIServer.Address, s.requester.APIServer.Port)
+	s.client = clientv2.New(fmt.Sprintf("http://%s:%d", s.requester.APIServer.Address, s.requester.APIServer.Port))
 	s.Require().NoError(err)
-	s.stateResolver = legacy.NewStateResolver(s.requester.RequesterNode.JobStore)
+	s.stateResolver = scenario.NewStateResolverFromStore(s.requester.RequesterNode.JobStore)
 	nodeutils.WaitForNodeDiscovery(s.T(), s.requester.RequesterNode, len(nodeOverrides))
 }
 
@@ -169,18 +168,18 @@ func (s *RetriesSuite) TestRetry() {
 		nodes                   []string // nodes to constrain the job to
 		concurrency             int
 		failed                  bool // whether the job should fail
-		expectedJobState        model.JobStateType
+		expectedJobState        models.JobStateType
 		expectedStateCount      IntMatch
-		expectedExecutionStates map[model.ExecutionStateType]IntMatch
-		expectedExecutionErrors map[model.ExecutionStateType]string
+		expectedExecutionStates map[models.ExecutionStateType]IntMatch
+		expectedExecutionErrors map[models.ExecutionStateType]string
 	}{
 		{
 			name:               "bid-rejected-succeed-with-retry-on-good-nodes",
 			nodes:              []string{"bid-rejector", "good-guy1"},
 			expectedStateCount: NewIntMatch(2),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateCompleted:         NewIntMatch(1),
-				model.ExecutionStateAskForBidRejected: NewIntMatch(1),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateCompleted:         NewIntMatch(1),
+				models.ExecutionStateAskForBidRejected: NewIntMatch(1),
 			},
 		},
 		{
@@ -188,20 +187,20 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"bid-rejector"},
 			failed:             true,
 			expectedStateCount: NewIntMatch(1),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateAskForBidRejected: NewIntMatch(1),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateAskForBidRejected: NewIntMatch(1),
 			},
 		},
 		{
 			name:               "execution-failure-succeed-with-retry-on-good-nodes",
 			nodes:              []string{"bad-executor", "good-guy1"},
 			expectedStateCount: NewIntMatch(2),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateCompleted: NewIntMatch(1),
-				model.ExecutionStateFailed:    NewIntMatch(1),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateCompleted: NewIntMatch(1),
+				models.ExecutionStateFailed:    NewIntMatch(1),
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errExecution.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errExecution.Error(),
 			},
 		},
 		{
@@ -209,23 +208,23 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"bad-executor"},
 			failed:             true,
 			expectedStateCount: NewIntMatch(1),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateFailed: NewIntMatch(2), // we retry up to two times on the same node
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateFailed: NewIntMatch(2), // we retry up to two times on the same node
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errExecution.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errExecution.Error(),
 			},
 		},
 		{
 			name:               "publish-failure-succeed-with-retry-on-good-nodes",
 			nodes:              []string{"bad-publisher", "good-guy1"},
 			expectedStateCount: NewIntMatch(2),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateCompleted: NewIntMatch(1),
-				model.ExecutionStateFailed:    NewIntMatch(1),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateCompleted: NewIntMatch(1),
+				models.ExecutionStateFailed:    NewIntMatch(1),
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errPublish.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errPublish.Error(),
 			},
 		},
 		{
@@ -233,11 +232,11 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"bad-publisher"},
 			failed:             true,
 			expectedStateCount: NewIntMatch(1),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateFailed: NewIntMatch(2),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateFailed: NewIntMatch(2),
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errPublish.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errPublish.Error(),
 			},
 		},
 		{
@@ -245,17 +244,17 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"bad-publisher", "good-guy1"},
 			concurrency:        2,
 			failed:             true,
-			expectedJobState:   model.JobStateError,
+			expectedJobState:   models.JobStateTypeFailed,
 			expectedStateCount: NewIntMatch(2),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
 				// we cancel the good-guy1 if the two attempts on bad-publisher fail but it may
 				// have completed so we will allow 0,1 cancelled and 0,1 completed
-				model.ExecutionStateCancelled: NewIntMatch(0, 1),
-				model.ExecutionStateCompleted: NewIntMatch(0, 1),
-				model.ExecutionStateFailed:    NewIntMatch(2), // we retry up to two times on the same node
+				models.ExecutionStateCancelled: NewIntMatch(0, 1),
+				models.ExecutionStateCompleted: NewIntMatch(0, 1),
+				models.ExecutionStateFailed:    NewIntMatch(2), // we retry up to two times on the same node
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errPublish.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errPublish.Error(),
 			},
 		},
 		{
@@ -263,14 +262,14 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"slow-executor", "bad-executor"},
 			concurrency:        2,
 			failed:             true,
-			expectedJobState:   model.JobStateError,
+			expectedJobState:   models.JobStateTypeFailed,
 			expectedStateCount: NewIntMatch(2),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateFailed:    NewIntMatch(2), // we retry up to two times on the same node
-				model.ExecutionStateCancelled: NewIntMatch(1),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateFailed:    NewIntMatch(2), // we retry up to two times on the same node
+				models.ExecutionStateCancelled: NewIntMatch(1),
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed: errExecution.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed: errExecution.Error(),
 			},
 		},
 		{
@@ -278,14 +277,14 @@ func (s *RetriesSuite) TestRetry() {
 			nodes:              []string{"bid-rejector", "bad-executor", "good-guy1", "good-guy2"},
 			concurrency:        2,
 			expectedStateCount: NewIntMatch(3),
-			expectedExecutionStates: map[model.ExecutionStateType]IntMatch{
-				model.ExecutionStateAskForBidRejected: NewIntMatch(1),
-				model.ExecutionStateFailed:            NewIntMatch(1),
-				model.ExecutionStateCompleted:         NewIntMatch(2),
+			expectedExecutionStates: map[models.ExecutionStateType]IntMatch{
+				models.ExecutionStateAskForBidRejected: NewIntMatch(1),
+				models.ExecutionStateFailed:            NewIntMatch(1),
+				models.ExecutionStateCompleted:         NewIntMatch(2),
 			},
-			expectedExecutionErrors: map[model.ExecutionStateType]string{
-				model.ExecutionStateFailed:    errExecution.Error(),
-				model.ExecutionStateCancelled: errExecution.Error(),
+			expectedExecutionErrors: map[models.ExecutionStateType]string{
+				models.ExecutionStateFailed:    errExecution.Error(),
+				models.ExecutionStateCancelled: errExecution.Error(),
 			},
 		},
 	}
@@ -293,19 +292,24 @@ func (s *RetriesSuite) TestRetry() {
 		s.Run(tc.name, func() {
 			zerolog.SetGlobalLevel(zerolog.InfoLevel)
 			ctx := context.Background()
+
 			j := makeBadTargetingJob(s.T(), tc.nodes)
-			j.Spec.Deal.Concurrency = math.Max(1, tc.concurrency)
-			submittedJob, err := s.client.Submit(ctx, j)
+			j.Count = math.Max(1, tc.concurrency)
+			j.Normalize()
+
+			submittedJob, err := s.client.Jobs().Put(ctx, &apimodels.PutJobRequest{
+				Job: j,
+			})
 			s.Require().NoError(err)
 
 			if tc.failed {
-				s.Require().Error(s.stateResolver.WaitUntilComplete(ctx, submittedJob.ID()))
+				s.Require().Error(s.stateResolver.Wait(ctx, submittedJob.JobID, scenario.WaitForSuccessfulCompletion()))
 			} else {
-				s.Require().NoError(s.stateResolver.WaitUntilComplete(ctx, submittedJob.ID()))
+				s.Require().NoError(s.stateResolver.Wait(ctx, submittedJob.JobID, scenario.WaitForSuccessfulCompletion()))
 			}
-			s.Require().NoError(s.stateResolver.Wait(ctx, submittedJob.ID(), legacy_job.WaitForTerminalStates()))
+			s.Require().NoError(s.stateResolver.Wait(ctx, submittedJob.JobID, scenario.WaitForTerminalStates()))
 
-			jobState, err := s.stateResolver.GetJobState(ctx, submittedJob.ID())
+			jobState, err := s.stateResolver.JobState(ctx, submittedJob.JobID)
 			if len(tc.expectedExecutionStates) == 0 {
 				// no job state is expected to exist for this scenario
 				s.Require().Error(err)
@@ -316,13 +320,13 @@ func (s *RetriesSuite) TestRetry() {
 			// verify job state
 			if tc.expectedJobState.IsUndefined() {
 				if tc.failed {
-					tc.expectedJobState = model.JobStateError
+					tc.expectedJobState = models.JobStateTypeFailed
 				} else {
-					tc.expectedJobState = model.JobStateCompleted
+					tc.expectedJobState = models.JobStateTypeCompleted
 				}
 			}
-			s.Require().Equal(tc.expectedJobState, jobState.State,
-				"Expected job in state %s, found %s", tc.expectedJobState.String(), jobState.State.String())
+			s.Require().Equal(tc.expectedJobState, jobState.State.StateType,
+				"Expected job in state %s, found %s", tc.expectedJobState.String(), jobState.State.StateType)
 
 			// verify execution states
 			executionStates := jobState.GroupExecutionsByState()
@@ -337,30 +341,47 @@ func (s *RetriesSuite) TestRetry() {
 			// verify execution error status message
 			for state, message := range tc.expectedExecutionErrors {
 				for _, execution := range executionStates[state] {
-					s.Require().Contains(execution.Status, message)
+					s.Require().Contains(execution.ComputeState.Message, message)
 				}
 			}
 		})
 	}
 }
 
-func makeBadTargetingJob(t testing.TB, restrictedNodes []string) *model.Job {
-	j := testutils.MakeJobWithOpts(t)
-	req := []model.LabelSelectorRequirement{
+func makeBadTargetingJob(t testing.TB, restrictedNodes []string) *models.Job {
+	j := &models.Job{
+		Name:  t.Name(),
+		Type:  models.JobTypeBatch,
+		Count: 1,
+		Tasks: []*models.Task{
+			{
+				Name: t.Name(),
+				Engine: &models.SpecConfig{
+					Type:   models.EngineNoop,
+					Params: make(map[string]interface{}),
+				},
+				Publisher: &models.SpecConfig{
+					Type:   models.PublisherNoop,
+					Params: make(map[string]interface{}),
+				},
+			},
+		},
+	}
+	req := []*models.LabelSelectorRequirement{
 		{
 			Key:      "favour_name",
 			Operator: selection.NotIn,
 			Values:   []string{"good-guy1", "good-guy2"},
 		}}
 	if len(restrictedNodes) > 0 {
-		req = append(req, model.LabelSelectorRequirement{
+		req = append(req, &models.LabelSelectorRequirement{
 			Key:      "name",
 			Operator: selection.In,
 			Values:   restrictedNodes,
 		})
 	}
-	j.Spec.NodeSelectors = req
-	return &j
+	j.Constraints = req
+	return j
 }
 
 // IntMatch is a type that contains a list of numbers that are
diff --git a/pkg/test/scenario/resolver.go b/pkg/test/scenario/resolver.go
index 55ee5e8050..9fa1dad27d 100644
--- a/pkg/test/scenario/resolver.go
+++ b/pkg/test/scenario/resolver.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/rs/zerolog/log"
 
+	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
@@ -14,14 +15,73 @@ import (
 )
 
 type StateResolver struct {
-	api             client.API
+	loader          JobStateLoader
 	maxWaitAttempts int
 	waitDelay       time.Duration
 }
 
-func NewStateResolver(api client.API) *StateResolver {
+type JobStateLoader interface {
+	GetJob(ctx context.Context, id string) (*JobState, error)
+}
+
+type apiWrapper struct {
+	api client.API
+}
+
+func (a *apiWrapper) GetJob(ctx context.Context, id string) (*JobState, error) {
+	getResp, err := a.api.Jobs().Get(ctx, &apimodels.GetJobRequest{
+		JobID:   id,
+		Include: "executions",
+	})
+	if err != nil {
+		return nil, fmt.Errorf("getting job from api: %w", err)
+	}
+	return &JobState{
+		ID:         getResp.Job.ID,
+		Executions: getResp.Executions.Items,
+		State:      getResp.Job.State,
+	}, nil
+}
+
+type storeWrapper struct {
+	store jobstore.Store
+}
+
+func (s *storeWrapper) GetJob(ctx context.Context, id string) (*JobState, error) {
+	job, err := s.store.GetJob(ctx, id)
+	if err != nil {
+		return nil, fmt.Errorf("getting job from store: %w", err)
+	}
+	executions, err := s.store.GetExecutions(ctx, jobstore.GetExecutionsOptions{
+		JobID:      id,
+		IncludeJob: false,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("getting job executions from store: %w", err)
+	}
+	exe := make([]*models.Execution, 0, len(executions))
+	for _, e := range executions {
+		tmp := e
+		exe = append(exe, &tmp)
+	}
+	return &JobState{
+		ID:         job.ID,
+		Executions: exe,
+		State:      job.State,
+	}, nil
+}
+
+func NewStateResolverFromStore(s jobstore.Store) *StateResolver {
 	return &StateResolver{
-		api:             api,
+		loader:          &storeWrapper{store: s},
+		maxWaitAttempts: 1000,
+		waitDelay:       time.Millisecond * 100,
+	}
+}
+
+func NewStateResolverFromAPI(api client.API) *StateResolver {
+	return &StateResolver{
+		loader:          &apiWrapper{api: api},
 		maxWaitAttempts: 1000,
 		waitDelay:       time.Millisecond * 100,
 	}
@@ -33,22 +93,24 @@ type JobState struct {
 	State      models.State[models.JobStateType]
 }
 
+// GroupExecutionsByState groups the executions by state
+func (s *JobState) GroupExecutionsByState() map[models.ExecutionStateType][]*models.Execution {
+	result := make(map[models.ExecutionStateType][]*models.Execution)
+	for _, execution := range s.Executions {
+		result[execution.ComputeState.StateType] = append(result[execution.ComputeState.StateType], execution)
+	}
+	return result
+}
+
 type StateChecks func(s *JobState) (bool, error)
 
 func (s *StateResolver) JobState(ctx context.Context, id string) (*JobState, error) {
-	resp, err := s.api.Jobs().Get(ctx, &apimodels.GetJobRequest{
-		JobID:   id,
-		Include: "executions",
-	})
+	state, err := s.loader.GetJob(ctx, id)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get job (%s): %w", id, err)
 	}
 
-	return &JobState{
-		ID:         resp.Job.ID,
-		Executions: resp.Executions.Items,
-		State:      resp.Job.State,
-	}, nil
+	return state, nil
 }
 
 func (s *StateResolver) Wait(ctx context.Context, id string, until ...StateChecks) error {
diff --git a/pkg/test/scenario/suite.go b/pkg/test/scenario/suite.go
index 715460d2e0..c2a14c6827 100644
--- a/pkg/test/scenario/suite.go
+++ b/pkg/test/scenario/suite.go
@@ -160,7 +160,7 @@ func (s *ScenarioRunner) RunScenario(scenario Scenario) string {
 	jobID := getResp.Job.ID
 
 	s.T().Log("Waiting for job")
-	s.Require().NoError(NewStateResolver(api).Wait(s.Ctx, jobID, scenario.JobCheckers...))
+	s.Require().NoError(NewStateResolverFromAPI(api).Wait(s.Ctx, jobID, scenario.JobCheckers...))
 
 	// Check outputs
 	if scenario.ResultsChecker != nil {
diff --git a/pkg/test/teststack/stack.go b/pkg/test/teststack/stack.go
index 47501449ef..9464d578b9 100644
--- a/pkg/test/teststack/stack.go
+++ b/pkg/test/teststack/stack.go
@@ -4,7 +4,6 @@ package teststack
 
 import (
 	"context"
-	"os"
 	"testing"
 	"time"
 
@@ -19,7 +18,6 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/model"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/node"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/endpoint/requester"
 	"github.com/bacalhau-project/bacalhau/pkg/repo"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
 )
@@ -51,9 +49,6 @@ func Setup(
 	t.Cleanup(func() {
 		cm.Cleanup(ctx)
 	})
-	if err := os.Setenv(requester.UseDeprecatedEndpointsForTesting, "true"); err != nil {
-		t.Fatal(err)
-	}
 	stack, err := devstack.Setup(ctx, cfg, cm, fsr, append(testDevStackConfig().Options(), opts...)...)
 	if err != nil {
 		t.Fatalf("creating teststack: %s", err)
diff --git a/pkg/test/utils/utils.go b/pkg/test/utils/utils.go
index 5f43cdcde9..e27dbc9746 100644
--- a/pkg/test/utils/utils.go
+++ b/pkg/test/utils/utils.go
@@ -2,18 +2,15 @@ package testutils
 
 import (
 	"context"
-	"fmt"
 	"regexp"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
 	legacy_job "github.com/bacalhau-project/bacalhau/pkg/legacyjob"
-	"github.com/bacalhau-project/bacalhau/pkg/lib/marshaller"
 	"github.com/bacalhau-project/bacalhau/pkg/model"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
 )
@@ -31,36 +28,6 @@ func GetJobFromTestOutput(ctx context.Context, t *testing.T, c clientv2.API, out
 	return j.Job
 }
 
-func GetJobFromTestOutputLegacy(ctx context.Context, t *testing.T, c *client.APIClient, out string) model.Job {
-	jobID := system.FindJobIDInTestOutputLegacy(out)
-	uuidRegex := regexp.MustCompile(`[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}`)
-	require.Regexp(t, uuidRegex, jobID, "Job ID should be a UUID")
-
-	j, _, err := c.Get(ctx, jobID)
-	require.NoError(t, err)
-	require.NotNil(t, j, "Failed to get job with ID: %s", out)
-	return j.Job
-}
-
-func FirstFatalError(_ *testing.T, output string) (model.TestFatalErrorHandlerContents, error) {
-	linesInOutput := system.SplitLines(output)
-	fakeFatalError := &model.TestFatalErrorHandlerContents{}
-	for _, line := range linesInOutput {
-		err := marshaller.JSONUnmarshalWithMax([]byte(line), fakeFatalError)
-		if err != nil {
-			return model.TestFatalErrorHandlerContents{}, err
-		} else {
-			return *fakeFatalError, nil
-		}
-	}
-	return model.TestFatalErrorHandlerContents{}, fmt.Errorf("no fatal error found in output")
-}
-
-func MakeNoopJob(t testing.TB) *model.Job {
-	j := MakeJobWithOpts(t)
-	return &j
-}
-
 func MakeJobWithOpts(t testing.TB, opts ...legacy_job.SpecOpt) model.Job {
 	spec, err := legacy_job.MakeSpec(opts...)
 	if err != nil {
@@ -72,14 +39,6 @@ func MakeJobWithOpts(t testing.TB, opts ...legacy_job.SpecOpt) model.Job {
 	return *j
 }
 
-func MakeSpecWithOpts(t testing.TB, opts ...legacy_job.SpecOpt) model.Spec {
-	spec, err := legacy_job.MakeSpec(opts...)
-	if err != nil {
-		t.Fatalf("creating job spec: %s", err)
-	}
-	return spec
-}
-
 // MustHaveIPFS will skip the test if the test is running in an environment that cannot support IPFS.
 func MustHaveIPFS(t testing.TB, ipfsConnect string) {
 	if !IsIPFSEnabled(ipfsConnect) {