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

Issue:
Add history event when the job is delayed in the queue
Add visibility on why the job is being delayed and queued up. We should only add this event once or at least if the queue reason didn't change, and not every time we re-evaluate the job, which we do every 5 minutes


Discussion / hints:

Fix patch:
diff --git a/cmd/cli/job/describe.go b/cmd/cli/job/describe.go
index 5c3a2209cd..c31cf1f363 100644
--- a/cmd/cli/job/describe.go
+++ b/cmd/cli/job/describe.go
@@ -218,8 +218,6 @@ func (o *DescribeOptions) printHistory(cmd *cobra.Command, label string, history
 	}
 	jobHistoryCols := []output.TableColumn[*models.JobHistory]{
 		timeCol,
-		historyRevisionCol,
-		historyStateCol,
 		historyTopicCol,
 		historyEventCol,
 	}
diff --git a/cmd/cli/job/history.go b/cmd/cli/job/history.go
index 8fc9507aa5..2fd5da17ab 100644
--- a/cmd/cli/job/history.go
+++ b/cmd/cli/job/history.go
@@ -2,7 +2,6 @@ package job
 
 import (
 	"fmt"
-	"strconv"
 	"time"
 
 	"github.com/jedib0t/go-pretty/v6/table"
@@ -46,7 +45,6 @@ type HistoryOptions struct {
 	cliflags.ListOptions
 	EventType   string
 	ExecutionID string
-	NodeID      string
 }
 
 // NewHistoryOptions returns initialized Options
@@ -84,8 +82,6 @@ func NewHistoryCmd() *cobra.Command {
 		"The type of history events to return. One of: all, job, execution")
 	nodeCmd.Flags().StringVar(&o.ExecutionID, "execution-id", o.ExecutionID,
 		"The execution id to filter by.")
-	nodeCmd.Flags().StringVar(&o.NodeID, "node-id", o.NodeID,
-		"The node id to filter by.")
 	nodeCmd.Flags().AddFlagSet(cliflags.ListFlags(&o.ListOptions))
 	nodeCmd.Flags().AddFlagSet(cliflags.OutputFormatFlags(&o.OutputOptions))
 	return nodeCmd
@@ -100,33 +96,16 @@ var (
 		ColumnConfig: table.ColumnConfig{Name: "Level", WidthMax: 15, WidthMaxEnforcer: text.WrapText},
 		Value:        func(jwi *models.JobHistory) string { return jwi.Type.String() },
 	}
-	historyRevisionCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "Rev.", WidthMax: 4, WidthMaxEnforcer: text.WrapText},
-		Value:        func(j *models.JobHistory) string { return strconv.FormatUint(j.NewRevision, 10) },
-	}
 	historyExecIDCol = output.TableColumn[*models.JobHistory]{
 		ColumnConfig: table.ColumnConfig{Name: "Exec. ID", WidthMax: 10, WidthMaxEnforcer: text.WrapText},
 		Value:        func(j *models.JobHistory) string { return idgen.ShortUUID(j.ExecutionID) },
 	}
-	historyNodeIDCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "Node ID", WidthMax: 10, WidthMaxEnforcer: text.WrapText},
-		Value:        func(j *models.JobHistory) string { return idgen.ShortNodeID(j.NodeID) },
-	}
-	historyStateCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "State", WidthMax: 20, WidthMaxEnforcer: text.WrapText},
-		Value: func(j *models.JobHistory) string {
-			if j.Type == models.JobHistoryTypeJobLevel {
-				return j.JobState.New.String()
-			}
-			return j.ExecutionState.New.String()
-		},
-	}
 	historyTopicCol = output.TableColumn[*models.JobHistory]{
 		ColumnConfig: table.ColumnConfig{Name: "Topic", WidthMax: 15, WidthMaxEnforcer: text.WrapSoft},
 		Value:        func(jh *models.JobHistory) string { return string(jh.Event.Topic) },
 	}
 	historyEventCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "Event", WidthMax: 60, WidthMaxEnforcer: text.WrapText},
+		ColumnConfig: table.ColumnConfig{Name: "Event", WidthMax: 90, WidthMaxEnforcer: text.WrapText},
 		Value: func(h *models.JobHistory) string {
 			res := h.Event.Message
 
@@ -161,10 +140,7 @@ var (
 var historyColumns = []output.TableColumn[*models.JobHistory]{
 	historyTimeCol,
 	historyLevelCol,
-	historyRevisionCol,
 	historyExecIDCol,
-	historyNodeIDCol,
-	historyStateCol,
 	historyTopicCol,
 	historyEventCol,
 }
@@ -176,7 +152,6 @@ func (o *HistoryOptions) run(cmd *cobra.Command, args []string, api client.API)
 		JobID:       jobID,
 		EventType:   o.EventType,
 		ExecutionID: o.ExecutionID,
-		NodeID:      o.NodeID,
 		BaseListRequest: apimodels.BaseListRequest{
 			Limit:     o.Limit,
 			NextToken: o.NextToken,
diff --git a/cmd/util/printer/print.go b/cmd/util/printer/print.go
index 7e753adfc4..f778576ed8 100644
--- a/cmd/util/printer/print.go
+++ b/cmd/util/printer/print.go
@@ -155,7 +155,7 @@ func followLogs(cmd *cobra.Command, api clientv2.API, jobID string, client clien
 		if err != nil {
 			return fmt.Errorf("failed getting job: %w", err)
 		}
-		if resp.Job.State.StateType != models.JobStateTypePending {
+		if resp.Job.State.StateType != models.JobStateTypePending && resp.Job.State.StateType != models.JobStateTypeQueued {
 			break
 		}
 		// TODO: add exponential backoff if there were no state updates
@@ -413,8 +413,7 @@ func summariseHistoryEvents(history []*models.JobHistory) []models.Event {
 	for _, entry := range history {
 		hasDetails := entry.Event.Details != nil
 		failsExecution := hasDetails && entry.Event.Details[models.DetailsKeyFailsExecution] == "true"
-		terminalState := entry.ExecutionState.New.IsTermainl()
-		if (failsExecution || terminalState) && entry.Event.Message != "" {
+		if failsExecution && entry.Event.Message != "" {
 			events[entry.Event.Message] = entry.Event
 		}
 	}
diff --git a/pkg/compute/events.go b/pkg/compute/events.go
index 7129b2d459..eb4a2881d3 100644
--- a/pkg/compute/events.go
+++ b/pkg/compute/events.go
@@ -16,13 +16,8 @@ const (
 )
 
 func RespondedToBidEvent(response *bidstrategy.BidStrategyResponse) models.Event {
-	message := response.Reason
-	if message == "" && response.ShouldBid {
-		message = "Accepted job"
-	}
-
 	return models.Event{
-		Message:   message,
+		Message:   response.Reason,
 		Topic:     EventTopicExecutionBidding,
 		Timestamp: time.Now(),
 		Details:   map[string]string{},
diff --git a/pkg/jobstore/boltdb/store.go b/pkg/jobstore/boltdb/store.go
index ca87dd6b99..18d6b3f297 100644
--- a/pkg/jobstore/boltdb/store.go
+++ b/pkg/jobstore/boltdb/store.go
@@ -7,7 +7,6 @@ import (
 	"slices"
 	"sort"
 	"strings"
-	"time"
 
 	"github.com/benbjohnson/clock"
 	"github.com/imdario/mergo"
@@ -700,10 +699,6 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
 			return false
 		}
 
-		if query.NodeID != "" && !strings.HasPrefix(event.NodeID, query.NodeID) {
-			return false
-		}
-
 		if event.Time.Unix() < query.Since {
 			return false
 		}
@@ -755,7 +750,7 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
 }
 
 // CreateJob creates a new record of a job in the data store
-func (b *BoltJobStore) CreateJob(ctx context.Context, job models.Job, event models.Event) error {
+func (b *BoltJobStore) CreateJob(ctx context.Context, job models.Job) error {
 	job.State = models.NewJobState(models.JobStateTypePending)
 	job.Revision = 1
 	job.CreateTime = b.clock.Now().UTC().UnixNano()
@@ -766,7 +761,7 @@ func (b *BoltJobStore) CreateJob(ctx context.Context, job models.Job, event mode
 		return err
 	}
 	return b.update(ctx, func(tx *bolt.Tx) (err error) {
-		return b.createJob(tx, job, event)
+		return b.createJob(tx, job)
 	})
 }
 
@@ -838,7 +833,7 @@ func (b *BoltJobStore) view(ctx context.Context, view func(tx *bolt.Tx) error) e
 	return view(tx)
 }
 
-func (b *BoltJobStore) createJob(tx *bolt.Tx, job models.Job, event models.Event) error {
+func (b *BoltJobStore) createJob(tx *bolt.Tx, job models.Job) error {
 	if b.jobExists(tx, job.ID) {
 		return jobstore.NewErrJobAlreadyExists(job.ID)
 	}
@@ -899,7 +894,7 @@ func (b *BoltJobStore) createJob(tx *bolt.Tx, job models.Job, event models.Event
 		}
 	}
 
-	return b.appendJobHistory(tx, job, models.JobStateTypePending, event)
+	return nil
 }
 
 // DeleteJob removes the specified job from the system entirely
@@ -988,9 +983,8 @@ func (b *BoltJobStore) updateJobState(tx *bolt.Tx, request jobstore.UpdateJobSta
 	})
 
 	// update the job state
-	previousState := job.State.StateType
 	job.State.StateType = request.NewState
-	job.State.Message = request.Event.Message
+	job.State.Message = request.Message
 	job.Revision++
 	job.ModifyTime = b.clock.Now().UTC().UnixNano()
 
@@ -1018,28 +1012,34 @@ func (b *BoltJobStore) updateJobState(tx *bolt.Tx, request jobstore.UpdateJobSta
 		}
 	}
 
-	return b.appendJobHistory(tx, job, previousState, request.Event)
+	return nil
+}
+
+// AddJobHistory appends a new history entry to the job history
+func (b *BoltJobStore) AddJobHistory(ctx context.Context, jobID string, events ...models.Event) error {
+	return b.update(ctx, func(tx *bolt.Tx) (err error) {
+		for _, event := range events {
+			if err = b.addJobHistory(tx, jobID, event); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 }
 
-func (b *BoltJobStore) appendJobHistory(tx *bolt.Tx, updateJob models.Job, previousState models.JobStateType, event models.Event) error {
+func (b *BoltJobStore) addJobHistory(tx *bolt.Tx, jobID string, event models.Event) error {
 	historyEntry := models.JobHistory{
 		Type:  models.JobHistoryTypeJobLevel,
-		JobID: updateJob.ID,
-		JobState: &models.StateChange[models.JobStateType]{
-			Previous: previousState,
-			New:      updateJob.State.StateType,
-		},
-		NewRevision: updateJob.Revision,
-		Comment:     event.Message,
-		Event:       event,
-		Time:        time.Unix(0, updateJob.ModifyTime),
+		JobID: jobID,
+		Event: event,
+		Time:  b.clock.Now().UTC(),
 	}
 	data, err := b.marshaller.Marshal(historyEntry)
 	if err != nil {
 		return err
 	}
 
-	if bkt, err := NewBucketPath(BucketJobs, updateJob.ID, BucketJobHistory).Get(tx, true); err != nil {
+	if bkt, err := NewBucketPath(BucketJobs, jobID, BucketJobHistory).Get(tx, true); err != nil {
 		return err
 	} else {
 		seq := BucketSequenceString(tx, bkt)
@@ -1052,7 +1052,7 @@ func (b *BoltJobStore) appendJobHistory(tx *bolt.Tx, updateJob models.Job, previ
 }
 
 // CreateExecution creates a record of a new execution
-func (b *BoltJobStore) CreateExecution(ctx context.Context, execution models.Execution, event models.Event) error {
+func (b *BoltJobStore) CreateExecution(ctx context.Context, execution models.Execution) error {
 	if execution.CreateTime == 0 {
 		execution.CreateTime = b.clock.Now().UTC().UnixNano()
 	}
@@ -1070,11 +1070,11 @@ func (b *BoltJobStore) CreateExecution(ctx context.Context, execution models.Exe
 		return err
 	}
 	return b.update(ctx, func(tx *bolt.Tx) (err error) {
-		return b.createExecution(tx, execution, event)
+		return b.createExecution(tx, execution)
 	})
 }
 
-func (b *BoltJobStore) createExecution(tx *bolt.Tx, execution models.Execution, event models.Event) error {
+func (b *BoltJobStore) createExecution(tx *bolt.Tx, execution models.Execution) error {
 	if !b.jobExists(tx, execution.JobID) {
 		return jobstore.NewErrJobNotFound(execution.JobID)
 	}
@@ -1112,7 +1112,7 @@ func (b *BoltJobStore) createExecution(tx *bolt.Tx, execution models.Execution,
 		}
 	}
 
-	return b.appendExecutionHistory(tx, execution, models.ExecutionStateNew, event)
+	return nil
 }
 
 // UpdateExecution updates the state of a single execution by loading from storage,
@@ -1173,25 +1173,28 @@ func (b *BoltJobStore) updateExecution(tx *bolt.Tx, request jobstore.UpdateExecu
 		}
 	}
 
-	return b.appendExecutionHistory(tx, newExecution, existingExecution.ComputeState.StateType, request.Event)
+	return nil
+}
+
+// AddExecutionHistory appends a new history entry to the execution history
+func (b *BoltJobStore) AddExecutionHistory(ctx context.Context, jobID, executionID string, events ...models.Event) error {
+	return b.update(ctx, func(tx *bolt.Tx) (err error) {
+		for _, event := range events {
+			if err = b.addExecutionHistory(tx, jobID, executionID, event); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 }
 
-func (b *BoltJobStore) appendExecutionHistory(tx *bolt.Tx, updated models.Execution,
-	previous models.ExecutionStateType, event models.Event,
-) error {
+func (b *BoltJobStore) addExecutionHistory(tx *bolt.Tx, jobID, executionID string, event models.Event) error {
 	historyEntry := models.JobHistory{
 		Type:        models.JobHistoryTypeExecutionLevel,
-		JobID:       updated.JobID,
-		NodeID:      updated.NodeID,
-		ExecutionID: updated.ID,
-		ExecutionState: &models.StateChange[models.ExecutionStateType]{
-			Previous: previous,
-			New:      updated.ComputeState.StateType,
-		},
-		NewRevision: updated.Revision,
-		Comment:     event.Message,
+		JobID:       jobID,
+		ExecutionID: executionID,
 		Event:       event,
-		Time:        time.Unix(0, updated.ModifyTime),
+		Time:        b.clock.Now().UTC(),
 	}
 
 	data, err := b.marshaller.Marshal(historyEntry)
@@ -1201,7 +1204,7 @@ func (b *BoltJobStore) appendExecutionHistory(tx *bolt.Tx, updated models.Execut
 
 	// Get the history bucket for this job ID, which involves potentially
 	// creating the bucket (executions_history.<jobid>)
-	if bkt, err := NewBucketPath(BucketJobs, updated.JobID, BucketExecutionHistory).Get(tx, true); err != nil {
+	if bkt, err := NewBucketPath(BucketJobs, jobID, BucketExecutionHistory).Get(tx, true); err != nil {
 		return err
 	} else {
 		seq := BucketSequenceString(tx, bkt)
diff --git a/pkg/jobstore/mocks.go b/pkg/jobstore/mocks.go
index 1d0f57eee5..3e6720de25 100644
--- a/pkg/jobstore/mocks.go
+++ b/pkg/jobstore/mocks.go
@@ -144,6 +144,44 @@ func (m *MockStore) EXPECT() *MockStoreMockRecorder {
 	return m.recorder
 }
 
+// AddExecutionHistory mocks base method.
+func (m *MockStore) AddExecutionHistory(ctx context.Context, jobID, executionID string, events ...models.Event) error {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, jobID, executionID}
+	for _, a := range events {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "AddExecutionHistory", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AddExecutionHistory indicates an expected call of AddExecutionHistory.
+func (mr *MockStoreMockRecorder) AddExecutionHistory(ctx, jobID, executionID interface{}, events ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, jobID, executionID}, events...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddExecutionHistory", reflect.TypeOf((*MockStore)(nil).AddExecutionHistory), varargs...)
+}
+
+// AddJobHistory mocks base method.
+func (m *MockStore) AddJobHistory(ctx context.Context, jobID string, events ...models.Event) error {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, jobID}
+	for _, a := range events {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "AddJobHistory", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AddJobHistory indicates an expected call of AddJobHistory.
+func (mr *MockStoreMockRecorder) AddJobHistory(ctx, jobID interface{}, events ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, jobID}, events...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddJobHistory", reflect.TypeOf((*MockStore)(nil).AddJobHistory), varargs...)
+}
+
 // BeginTx mocks base method.
 func (m *MockStore) BeginTx(ctx context.Context) (TxContext, error) {
 	m.ctrl.T.Helper()
@@ -188,31 +226,31 @@ func (mr *MockStoreMockRecorder) CreateEvaluation(ctx, eval interface{}) *gomock
 }
 
 // CreateExecution mocks base method.
-func (m *MockStore) CreateExecution(ctx context.Context, execution models.Execution, event models.Event) error {
+func (m *MockStore) CreateExecution(ctx context.Context, execution models.Execution) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateExecution", ctx, execution, event)
+	ret := m.ctrl.Call(m, "CreateExecution", ctx, execution)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // CreateExecution indicates an expected call of CreateExecution.
-func (mr *MockStoreMockRecorder) CreateExecution(ctx, execution, event interface{}) *gomock.Call {
+func (mr *MockStoreMockRecorder) CreateExecution(ctx, execution interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateExecution", reflect.TypeOf((*MockStore)(nil).CreateExecution), ctx, execution, event)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateExecution", reflect.TypeOf((*MockStore)(nil).CreateExecution), ctx, execution)
 }
 
 // CreateJob mocks base method.
-func (m *MockStore) CreateJob(ctx context.Context, j models.Job, event models.Event) error {
+func (m *MockStore) CreateJob(ctx context.Context, j models.Job) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateJob", ctx, j, event)
+	ret := m.ctrl.Call(m, "CreateJob", ctx, j)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // CreateJob indicates an expected call of CreateJob.
-func (mr *MockStoreMockRecorder) CreateJob(ctx, j, event interface{}) *gomock.Call {
+func (mr *MockStoreMockRecorder) CreateJob(ctx, j interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateJob", reflect.TypeOf((*MockStore)(nil).CreateJob), ctx, j, event)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateJob", reflect.TypeOf((*MockStore)(nil).CreateJob), ctx, j)
 }
 
 // DeleteEvaluation mocks base method.
diff --git a/pkg/jobstore/types.go b/pkg/jobstore/types.go
index 0ef55fab79..dfa921e090 100644
--- a/pkg/jobstore/types.go
+++ b/pkg/jobstore/types.go
@@ -4,8 +4,9 @@ package jobstore
 import (
 	"context"
 
-	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"k8s.io/apimachinery/pkg/labels"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
 )
 
 type JobQuery struct {
@@ -36,7 +37,6 @@ type JobHistoryQuery struct {
 	ExcludeExecutionLevel bool   `json:"exclude_execution_level"`
 	ExcludeJobLevel       bool   `json:"exclude_job_level"`
 	ExecutionID           string `json:"execution_id"`
-	NodeID                string `json:"node_id"`
 	NextToken             string `json:"next_token"`
 }
 
@@ -92,7 +92,7 @@ type Store interface {
 	GetJobHistory(ctx context.Context, jobID string, options JobHistoryQuery) (*JobHistoryQueryResponse, error)
 
 	// CreateJob will create a new job and persist it in the store.
-	CreateJob(ctx context.Context, j models.Job, event models.Event) error
+	CreateJob(ctx context.Context, j models.Job) error
 
 	// GetExecutions retrieves all executions for the specified job.
 	GetExecutions(ctx context.Context, options GetExecutionsOptions) ([]models.Execution, error)
@@ -101,13 +101,19 @@ type Store interface {
 	// [UpdateJobStateRequest].
 	UpdateJobState(ctx context.Context, request UpdateJobStateRequest) error
 
+	// AddJobHistory adds a new history entry for the specified job
+	AddJobHistory(ctx context.Context, jobID string, events ...models.Event) error
+
 	// CreateExecution creates a new execution
-	CreateExecution(ctx context.Context, execution models.Execution, event models.Event) error
+	CreateExecution(ctx context.Context, execution models.Execution) error
 
 	// UpdateExecution updates the execution state according to the values
 	// within [UpdateExecutionRequest].
 	UpdateExecution(ctx context.Context, request UpdateExecutionRequest) error
 
+	// AddExecutionHistory adds a new history entry for the specified execution
+	AddExecutionHistory(ctx context.Context, jobID, executionID string, events ...models.Event) error
+
 	// DeleteJob removes all trace of the provided job from storage
 	DeleteJob(ctx context.Context, jobID string) error
 
@@ -129,14 +135,13 @@ type UpdateJobStateRequest struct {
 	JobID     string
 	Condition UpdateJobCondition
 	NewState  models.JobStateType
-	Event     models.Event
+	Message   string
 }
 
 type UpdateExecutionRequest struct {
 	ExecutionID string
 	Condition   UpdateExecutionCondition
 	NewValues   models.Execution
-	Event       models.Event
 }
 
 type UpdateJobCondition struct {
diff --git a/pkg/models/job_history.go b/pkg/models/job_history.go
index 406c895d22..bd10502055 100644
--- a/pkg/models/job_history.go
+++ b/pkg/models/job_history.go
@@ -41,16 +41,17 @@ type StateChange[StateType any] struct {
 // {Job,Event}State fields will only be present if the Type field is of
 // the matching type.
 type JobHistory struct {
-	Type           JobHistoryType                   `json:"Type"`
-	JobID          string                           `json:"JobID"`
-	NodeID         string                           `json:"NodeID,omitempty"`
-	ExecutionID    string                           `json:"ExecutionID,omitempty"`
-	JobState       *StateChange[JobStateType]       `json:"JobState,omitempty"`
+	Type        JobHistoryType `json:"Type"`
+	JobID       string         `json:"JobID"`
+	ExecutionID string         `json:"ExecutionID,omitempty"`
+	Event       Event          `json:"Event,omitempty"`
+	Time        time.Time      `json:"Time"`
+
+	// TODO: remove with v1.5
+	// Deprecated: Left for backward compatibility with v1.4.x clients
+	JobState *StateChange[JobStateType] `json:"JobState,omitempty"`
+	// Deprecated: Left for backward compatibility with v1.4.x clients
 	ExecutionState *StateChange[ExecutionStateType] `json:"ExecutionState,omitempty"`
-	NewRevision    uint64                           `json:"NewRevision"`
-	Comment        string                           `json:"Comment,omitempty"`
-	Event          Event                            `json:"Event,omitempty"`
-	Time           time.Time                        `json:"Time"`
 }
 
 // Occurred returns when the action that triggered an update to job history
diff --git a/pkg/models/plan.go b/pkg/models/plan.go
index 193e08314a..aad9095661 100644
--- a/pkg/models/plan.go
+++ b/pkg/models/plan.go
@@ -8,16 +8,15 @@ type PlanExecutionDesiredUpdate struct {
 
 // Plan holds actions as a result of processing an evaluation by the scheduler.
 type Plan struct {
-	EvalID      string `json:"EvalID"`
-	EvalReceipt string `json:"EvalReceipt"`
-	// TODO: passing the evalID should be enough once we persist evaluations
-	Eval     *Evaluation `json:"Eval,omitempty"`
-	Priority int         `json:"Priority"`
+	EvalID      string      `json:"EvalID"`
+	EvalReceipt string      `json:"EvalReceipt"`
+	Eval        *Evaluation `json:"Eval,omitempty"`
+	Priority    int         `json:"Priority"`
 
 	Job *Job `json:"Job,omitempty"`
 
 	DesiredJobState JobStateType `json:"DesiredJobState,omitempty"`
-	Event           Event        `json:"Event,omitempty"`
+	UpdateMessage   string       `json:"Message,omitempty"`
 
 	// NewExecutions holds the executions to be created.
 	NewExecutions []*Execution `json:"NewExecutions,omitempty"`
@@ -26,6 +25,9 @@ type Plan struct {
 
 	// NewEvaluations holds the evaluations to be created, such as delayed evaluations when no nodes are available.
 	NewEvaluations []*Evaluation `json:"NewEvaluations,omitempty"`
+
+	JobEvents       []Event            `json:"JobEvents,omitempty"`
+	ExecutionEvents map[string][]Event `json:"ExecutionEvents,omitempty"`
 }
 
 // NewPlan creates a new Plan instance.
@@ -38,12 +40,15 @@ func NewPlan(eval *Evaluation, job *Job) *Plan {
 		NewExecutions:     []*Execution{},
 		UpdatedExecutions: make(map[string]*PlanExecutionDesiredUpdate),
 		NewEvaluations:    []*Evaluation{},
+		JobEvents:         []Event{},
+		ExecutionEvents:   make(map[string][]Event),
 	}
 }
 
 // AppendExecution appends the execution to the plan executions.
-func (p *Plan) AppendExecution(execution *Execution) {
+func (p *Plan) AppendExecution(execution *Execution, event Event) {
 	p.NewExecutions = append(p.NewExecutions, execution)
+	p.AppendExecutionEvent(execution.ID, event)
 }
 
 // AppendStoppedExecution marks an execution to be stopped.
@@ -54,15 +59,17 @@ func (p *Plan) AppendStoppedExecution(execution *Execution, event Event) {
 		Event:        event,
 	}
 	p.UpdatedExecutions[execution.ID] = updateRequest
+	p.AppendExecutionEvent(execution.ID, event)
 }
 
 // AppendApprovedExecution marks an execution as accepted and ready to be started.
-func (p *Plan) AppendApprovedExecution(execution *Execution) {
+func (p *Plan) AppendApprovedExecution(execution *Execution, event Event) {
 	updateRequest := &PlanExecutionDesiredUpdate{
 		Execution:    execution,
 		DesiredState: ExecutionDesiredStateRunning,
 	}
 	p.UpdatedExecutions[execution.ID] = updateRequest
+	p.AppendExecutionEvent(execution.ID, event)
 }
 
 // AppendEvaluation appends the evaluation to the plan evaluations.
@@ -70,41 +77,46 @@ func (p *Plan) AppendEvaluation(eval *Evaluation) {
 	p.NewEvaluations = append(p.NewEvaluations, eval)
 }
 
-func (p *Plan) MarkJobCompleted() {
+func (p *Plan) MarkJobCompleted(event Event) {
 	p.DesiredJobState = JobStateTypeCompleted
 	p.NewExecutions = []*Execution{}
+	p.AppendJobEvent(event)
 }
 
 // MarkJobRunningIfEligible updates the job state to "Running" under certain conditions.
-func (p *Plan) MarkJobRunningIfEligible() {
+func (p *Plan) MarkJobRunningIfEligible(event Event) bool {
 	// Exit the function if DesiredJobState is already defined.
 	if !p.DesiredJobState.IsUndefined() {
-		return
+		return false
 	}
 
 	// Only proceed if the current job state is "Pending" or "Queued".
 	if p.Job.State.StateType != JobStateTypePending && p.Job.State.StateType != JobStateTypeQueued {
-		return
+		return false
 	}
 
 	// Check if there are any running executions; if not, exit the function.
 	if !p.hasRunningExecutions() {
-		return
+		return false
 	}
 
 	// All conditions met, set DesiredJobState to "Running".
 	p.DesiredJobState = JobStateTypeRunning
+	p.AppendJobEvent(event)
+	return true
 }
 
 // MarkJobQueued marks the job as pending.
 func (p *Plan) MarkJobQueued(event Event) {
 	p.DesiredJobState = JobStateTypeQueued
-	p.Event = event
+	p.UpdateMessage = event.Message
+	p.AppendJobEvent(event)
 }
 
 func (p *Plan) MarkJobFailed(event Event) {
 	p.DesiredJobState = JobStateTypeFailed
-	p.Event = event
+	p.UpdateMessage = event.Message
+	p.AppendJobEvent(event)
 
 	p.NewExecutions = []*Execution{}
 	// drop any update that is not stopping an execution
@@ -120,6 +132,19 @@ func (p *Plan) IsJobFailed() bool {
 	return p.DesiredJobState == JobStateTypeFailed
 }
 
+// AppendJobEvent appends the event to the job events.
+func (p *Plan) AppendJobEvent(event Event) {
+	p.JobEvents = append(p.JobEvents, event)
+}
+
+// AppendExecutionEvent appends the event to the execution events.
+func (p *Plan) AppendExecutionEvent(executionID string, event Event) {
+	if _, ok := p.ExecutionEvents[executionID]; !ok {
+		p.ExecutionEvents[executionID] = []Event{}
+	}
+	p.ExecutionEvents[executionID] = append(p.ExecutionEvents[executionID], event)
+}
+
 // hasRunningExecutions returns true if the plan has executions in desired state "Running".
 func (p *Plan) hasRunningExecutions() bool {
 	for _, exec := range p.NewExecutions {
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index c3f65694d3..33ce91e30b 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -39,7 +39,6 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	"github.com/bacalhau-project/bacalhau/pkg/orchestrator/selection/discovery"
 	"github.com/bacalhau-project/bacalhau/pkg/orchestrator/selection/ranking"
-	"github.com/bacalhau-project/bacalhau/pkg/requester"
 )
 
 type Requester struct {
@@ -313,12 +312,12 @@ 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{
+	callback := orchestrator.NewCallback(&orchestrator.CallbackParams{
 		ID:           nodeID,
 		EventEmitter: eventEmitter,
 		Store:        jobStore,
 	})
-	if err = transportLayer.RegisterComputeCallback(endpoint); err != nil {
+	if err = transportLayer.RegisterComputeCallback(callback); err != nil {
 		return nil, err
 	}
 
diff --git a/pkg/requester/endpoint.go b/pkg/orchestrator/callback.go
similarity index 63%
rename from pkg/requester/endpoint.go
rename to pkg/orchestrator/callback.go
index 5ae41a954b..5b55d078cc 100644
--- a/pkg/requester/endpoint.go
+++ b/pkg/orchestrator/callback.go
@@ -1,4 +1,4 @@
-package requester
+package orchestrator
 
 import (
 	"context"
@@ -10,24 +10,23 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
 )
 
-type BaseEndpointParams struct {
+type CallbackParams struct {
 	ID           string
 	Store        jobstore.Store
-	EventEmitter orchestrator.EventEmitter
+	EventEmitter EventEmitter
 }
 
-// BaseEndpoint base implementation of requester Endpoint
-type BaseEndpoint struct {
+// Callback base implementation of requester Endpoint
+type Callback struct {
 	id           string
 	store        jobstore.Store
-	eventEmitter orchestrator.EventEmitter
+	eventEmitter EventEmitter
 }
 
-func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
-	return &BaseEndpoint{
+func NewCallback(params *CallbackParams) *Callback {
+	return &Callback{
 		id:           params.ID,
 		store:        params.Store,
 		eventEmitter: params.EventEmitter,
@@ -39,9 +38,11 @@ func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
 // /////////////////////////////
 
 // OnBidComplete implements compute.Callback
-func (e *BaseEndpoint) OnBidComplete(ctx context.Context, response compute.BidResult) {
+func (e *Callback) OnBidComplete(ctx context.Context, response compute.BidResult) {
 	log.Ctx(ctx).Debug().Msgf("Requester node received bid response %+v", response)
 
+	var executionEvents []models.Event
+
 	updateRequest := jobstore.UpdateExecutionRequest{
 		ExecutionID: response.ExecutionID,
 		Condition: jobstore.UpdateExecutionCondition{
@@ -53,41 +54,75 @@ func (e *BaseEndpoint) OnBidComplete(ctx context.Context, response compute.BidRe
 		NewValues: models.Execution{
 			ComputeState: models.NewExecutionState(models.ExecutionStateAskForBidAccepted).WithMessage(response.Event.Message),
 		},
-		Event: response.Event,
 	}
 
 	if !response.Accepted {
 		updateRequest.NewValues.ComputeState.StateType = models.ExecutionStateAskForBidRejected
 		updateRequest.NewValues.DesiredState =
 			models.NewExecutionDesiredState(models.ExecutionDesiredStateStopped).WithMessage("bid rejected")
+		executionEvents = append(executionEvents, response.Event)
 	}
-	err := e.store.UpdateExecution(ctx, updateRequest)
+
+	txContext, err := e.store.BeginTx(ctx)
 	if err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnBidComplete] failed to begin transaction")
+		return
+	}
+
+	defer func() {
+		if err != nil {
+			_ = txContext.Rollback()
+		}
+	}()
+
+	if err = e.store.UpdateExecution(txContext, updateRequest); err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("[OnBidComplete] failed to update execution")
 		return
 	}
 
+	for _, event := range executionEvents {
+		if err = e.store.AddExecutionHistory(txContext, response.JobID, response.ExecutionID, event); err != nil {
+			log.Ctx(ctx).Error().Err(err).Msgf("[OnBidComplete] failed to add execution history")
+			return
+		}
+	}
+
 	// enqueue evaluation to allow the scheduler to either accept the bid, or find a new node
-	e.enqueueEvaluation(ctx, response.JobID, "OnBidComplete")
+	e.enqueueEvaluation(txContext, response.JobID, "OnBidComplete")
+
+	if err = txContext.Commit(); err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnBidComplete] failed to commit transaction")
+		return
+	}
 
 	if response.Accepted {
 		e.eventEmitter.EmitBidReceived(ctx, response)
 	}
 }
 
-func (e *BaseEndpoint) OnRunComplete(ctx context.Context, result compute.RunResult) {
+func (e *Callback) OnRunComplete(ctx context.Context, result compute.RunResult) {
 	log.Ctx(ctx).Debug().Msgf("Requester node %s received RunComplete for execution: %s from %s",
 		e.id, result.ExecutionID, result.SourcePeerID)
 	e.eventEmitter.EmitRunComplete(ctx, result)
 
-	job, err := e.store.GetJob(ctx, result.JobID)
+	txContext, err := e.store.BeginTx(ctx)
+	if err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnRunComplete] failed to begin transaction")
+		return
+	}
+
+	defer func() {
+		_ = txContext.Rollback()
+	}()
+
+	job, err := e.store.GetJob(txContext, result.JobID)
 	if err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("[OnRunComplete] failed to get job %s", result.JobID)
 		return
 	}
 
 	// update execution state
-	updateExecutionRequest := jobstore.UpdateExecutionRequest{
+	updateRequest := jobstore.UpdateExecutionRequest{
 		ExecutionID: result.ExecutionID,
 		Condition: jobstore.UpdateExecutionCondition{
 			ExpectedStates: []models.ExecutionStateType{
@@ -110,33 +145,52 @@ func (e *BaseEndpoint) OnRunComplete(ctx context.Context, result compute.RunResu
 	if job.IsLongRunning() {
 		log.Ctx(ctx).Error().Msgf(
 			"[OnRunComplete] job %s is long running, but received a RunComplete. Marking the execution as failed instead", result.JobID)
-		updateExecutionRequest.NewValues.ComputeState =
+		updateRequest.NewValues.ComputeState =
 			models.NewExecutionState(models.ExecutionStateFailed).WithMessage("execution completed unexpectedly")
-		updateExecutionRequest.NewValues.DesiredState =
+		updateRequest.NewValues.DesiredState =
 			models.NewExecutionDesiredState(models.ExecutionDesiredStateStopped).WithMessage("execution completed unexpectedly")
 	}
 
-	err = e.store.UpdateExecution(ctx, updateExecutionRequest)
-	if err != nil {
+	if err = e.store.UpdateExecution(txContext, updateRequest); err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("[OnRunComplete] failed to update execution")
 		return
 	}
 
+	if err = e.store.AddExecutionHistory(txContext, result.JobID, result.ExecutionID, ExecCompletedEvent()); err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnRunComplete] failed to add execution history")
+		return
+	}
+
 	// enqueue evaluation to allow the scheduler to mark the job as completed if all executions are completed
-	e.enqueueEvaluation(ctx, result.JobID, "OnRunComplete")
+	e.enqueueEvaluation(txContext, result.JobID, "OnRunComplete")
+
+	if err = txContext.Commit(); err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnComputeFailure] failed to commit transaction")
+		return
+	}
 }
 
-func (e *BaseEndpoint) OnCancelComplete(ctx context.Context, result compute.CancelResult) {
+func (e *Callback) OnCancelComplete(ctx context.Context, result compute.CancelResult) {
 	log.Ctx(ctx).Debug().Msgf("Requester node %s received CancelComplete for execution: %s from %s",
 		e.id, result.ExecutionID, result.SourcePeerID)
 }
 
-func (e *BaseEndpoint) OnComputeFailure(ctx context.Context, result compute.ComputeError) {
+func (e *Callback) OnComputeFailure(ctx context.Context, result compute.ComputeError) {
 	log.Ctx(ctx).Debug().Err(result).Msgf("Requester node %s received ComputeFailure for execution: %s from %s",
 		e.id, result.ExecutionID, result.SourcePeerID)
 
+	txContext, err := e.store.BeginTx(ctx)
+	if err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnRunComplete] failed to begin transaction")
+		return
+	}
+
+	defer func() {
+		_ = txContext.Rollback()
+	}()
+
 	// update execution state
-	err := e.store.UpdateExecution(ctx, jobstore.UpdateExecutionRequest{
+	if err = e.store.UpdateExecution(txContext, jobstore.UpdateExecutionRequest{
 		ExecutionID: result.ExecutionID,
 		Condition: jobstore.UpdateExecutionCondition{
 			UnexpectedStates: []models.ExecutionStateType{
@@ -148,21 +202,29 @@ func (e *BaseEndpoint) OnComputeFailure(ctx context.Context, result compute.Comp
 			ComputeState: models.NewExecutionState(models.ExecutionStateFailed).WithMessage(result.Error()),
 			DesiredState: models.NewExecutionDesiredState(models.ExecutionDesiredStateStopped).WithMessage("execution failed"),
 		},
-		Event: result.Event,
-	})
-	if err != nil {
+	}); err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("[OnComputeFailure] failed to update execution")
 		return
 	}
 
 	// enqueue evaluation to allow the scheduler find other nodes, or mark the job as failed
-	e.enqueueEvaluation(ctx, result.JobID, "OnComputeFailure")
+	e.enqueueEvaluation(txContext, result.JobID, "OnComputeFailure")
+
+	if err = e.store.AddExecutionHistory(txContext, result.JobID, result.ExecutionID, result.Event); err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnComputeFailure] failed to add execution history")
+		return
+	}
+
+	if err = txContext.Commit(); err != nil {
+		log.Ctx(ctx).Error().Err(err).Msgf("[OnComputeFailure] failed to commit transaction")
+		return
+	}
+
 	e.eventEmitter.EmitComputeFailure(ctx, result.ExecutionID, result)
 }
 
 // enqueueEvaluation enqueues an evaluation to allow the scheduler to either accept the bid, or find a new node
-// TODO: solve edge case where execution is updated, but evaluation is not enqueued
-func (e *BaseEndpoint) enqueueEvaluation(ctx context.Context, jobID, operation string) {
+func (e *Callback) enqueueEvaluation(ctx context.Context, jobID, operation string) {
 	job, err := e.store.GetJob(ctx, jobID)
 	if err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("[%s] failed to get job %s while enqueueing evaluation", operation, jobID)
@@ -187,4 +249,4 @@ func (e *BaseEndpoint) enqueueEvaluation(ctx context.Context, jobID, operation s
 }
 
 // Compile-time interface check:
-var _ compute.Callback = (*BaseEndpoint)(nil)
+var _ compute.Callback = (*Callback)(nil)
diff --git a/pkg/orchestrator/endpoint.go b/pkg/orchestrator/endpoint.go
index 9ac7315e42..60e5f7d17a 100644
--- a/pkg/orchestrator/endpoint.go
+++ b/pkg/orchestrator/endpoint.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
-	"github.com/rs/zerolog/log"
 	"sigs.k8s.io/yaml"
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
@@ -55,10 +54,6 @@ func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
 // SubmitJob submits a job to the evaluation broker.
 func (e *BaseEndpoint) SubmitJob(ctx context.Context, request *SubmitJobRequest) (*SubmitJobResponse, error) {
 	job := request.Job
-	events := []models.Event{
-		JobSubmittedEvent(),
-	}
-
 	job.Normalize()
 	warnings := job.SanitizeSubmission()
 
@@ -66,6 +61,8 @@ func (e *BaseEndpoint) SubmitJob(ctx context.Context, request *SubmitJobRequest)
 		return nil, err
 	}
 
+	var translationEvent models.Event
+
 	// We will only perform task translation in the orchestrator if we were provided with a provider
 	// that can give translators to perform the translation.
 	if e.taskTranslator != nil {
@@ -85,23 +82,33 @@ func (e *BaseEndpoint) SubmitJob(ctx context.Context, request *SubmitJobRequest)
 				return nil, errors.Wrap(err, "failure converting job to JSON")
 			} else {
 				translatedJob.Meta[models.MetaDerivedFrom] = base64.StdEncoding.EncodeToString(b)
-				events = append(events, JobTranslatedEvent(job, translatedJob))
+				translationEvent = JobTranslatedEvent(job, translatedJob)
 			}
 
 			job = translatedJob
 		}
 	}
 
-	for i, event := range events {
-		if i == 0 {
-			if err := e.store.CreateJob(ctx, *job, events[0]); err != nil {
-				return nil, err
-			}
-		} else {
-			req := jobstore.UpdateJobStateRequest{JobID: job.ID, Event: event, NewState: models.JobStateTypePending}
-			if err := e.store.UpdateJobState(ctx, req); err != nil {
-				return nil, err
-			}
+	txContext, err := e.store.BeginTx(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to begin transaction: %w", err)
+	}
+
+	defer func() {
+		if err != nil {
+			_ = txContext.Rollback()
+		}
+	}()
+
+	if err = e.store.CreateJob(txContext, *job); err != nil {
+		return nil, err
+	}
+	if err = e.store.AddJobHistory(txContext, job.ID, JobSubmittedEvent()); err != nil {
+		return nil, err
+	}
+	if translationEvent.Message != "" {
+		if err = e.store.AddJobHistory(txContext, job.ID, translationEvent); err != nil {
+			return nil, err
 		}
 	}
 
@@ -115,10 +122,11 @@ func (e *BaseEndpoint) SubmitJob(ctx context.Context, request *SubmitJobRequest)
 		ModifyTime:  job.CreateTime,
 	}
 
-	// TODO(ross): How can we create this evaluation in the same transaction that the CreateJob
-	// call uses.
-	if err := e.store.CreateEvaluation(ctx, *eval); err != nil {
-		log.Ctx(ctx).Error().Err(err).Msgf("failed to save evaluation for job %s", job.ID)
+	if err = e.store.CreateEvaluation(txContext, *eval); err != nil {
+		return nil, err
+	}
+
+	if err = txContext.Commit(); err != nil {
 		return nil, err
 	}
 
@@ -141,11 +149,24 @@ func (e *BaseEndpoint) StopJob(ctx context.Context, request *StopJobRequest) (St
 		return StopJobResponse{}, nil
 	case models.JobStateTypeCompleted:
 		return StopJobResponse{}, fmt.Errorf("cannot stop job in state %s", job.State.StateType)
+	default:
+		// continue
 	}
 
+	txContext, err := e.store.BeginTx(ctx)
+	if err != nil {
+		return StopJobResponse{}, fmt.Errorf("failed to begin transaction: %w", err)
+	}
+
+	defer func() {
+		if err != nil {
+			_ = txContext.Rollback()
+		}
+	}()
+
 	// 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{
+	if err = e.store.UpdateJobState(txContext, jobstore.UpdateJobStateRequest{
 		JobID: request.JobID,
 		Condition: jobstore.UpdateJobCondition{
 			UnexpectedStates: []models.JobStateType{
@@ -153,9 +174,12 @@ func (e *BaseEndpoint) StopJob(ctx context.Context, request *StopJobRequest) (St
 			},
 		},
 		NewState: models.JobStateTypeStopped,
-		Event:    JobStoppedEvent(request.Reason),
-	})
-	if err != nil {
+		Message:  request.Reason,
+	}); err != nil {
+		return StopJobResponse{}, err
+	}
+
+	if err = e.store.AddJobHistory(txContext, job.ID, JobStoppedEvent(request.Reason)); err != nil {
 		return StopJobResponse{}, err
 	}
 
@@ -174,14 +198,16 @@ func (e *BaseEndpoint) StopJob(ctx context.Context, request *StopJobRequest) (St
 			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 stop job %s", request.JobID)
+		if err = e.store.CreateEvaluation(txContext, *eval); err != nil {
 			return StopJobResponse{}, err
 		}
 		evalID = eval.ID
 	}
+
+	if err = txContext.Commit(); err != nil {
+		return StopJobResponse{}, err
+	}
+
 	e.eventEmitter.EmitEventSilently(ctx, models.JobEvent{
 		JobID:     request.JobID,
 		EventName: models.JobEventCanceled,
diff --git a/pkg/orchestrator/errors.go b/pkg/orchestrator/errors.go
index f3edcbee1a..aee7c8eb72 100644
--- a/pkg/orchestrator/errors.go
+++ b/pkg/orchestrator/errors.go
@@ -4,8 +4,9 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 	"github.com/samber/lo"
+
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 )
 
 // ErrSchedulerNotFound is returned when the scheduler is not found for a given evaluation type
@@ -72,12 +73,40 @@ func (e ErrNotEnoughNodes) Details() map[string]string {
 
 // ErrNoMatchingNodes is returned when no matching nodes in the network to run a job
 type ErrNoMatchingNodes struct {
+	AvailableNodes []NodeRank
 }
 
-func NewErrNoMatchingNodes() ErrNoMatchingNodes {
-	return ErrNoMatchingNodes{}
+func NewErrNoMatchingNodes(availableNodes []NodeRank) ErrNoMatchingNodes {
+	return ErrNoMatchingNodes{
+		AvailableNodes: availableNodes,
+	}
 }
 
 func (e ErrNoMatchingNodes) Error() string {
-	return "no matching nodes to run job"
+	reasons := lo.GroupBy(e.AvailableNodes, func(rank NodeRank) string { return rank.Reason })
+
+	var message strings.Builder
+	fmt.Fprintf(&message, "not matching nodes to run job out of %d available nodes.", len(e.AvailableNodes))
+	for reason, nodes := range reasons {
+		fmt.Fprint(&message, "\n• ")
+		if len(nodes) > 1 {
+			fmt.Fprintf(&message, "%d of %d nodes", len(nodes), len(e.AvailableNodes))
+		} else {
+			fmt.Fprintf(&message, "Node %s", idgen.ShortNodeID(nodes[0].NodeInfo.ID()))
+		}
+		fmt.Fprintf(&message, ": %s", reason)
+	}
+	return message.String()
+}
+
+func (e ErrNoMatchingNodes) Details() map[string]string {
+	return map[string]string{
+		"NodesAvailable": fmt.Sprint(len(e.AvailableNodes)),
+	}
+}
+
+func (e ErrNoMatchingNodes) Retryable() bool {
+	return lo.ContainsBy(e.AvailableNodes, func(rank NodeRank) bool {
+		return !rank.MeetsRequirement() && rank.Retryable
+	})
 }
diff --git a/pkg/orchestrator/events.go b/pkg/orchestrator/events.go
index 85a1bc4cd3..ad3c116726 100644
--- a/pkg/orchestrator/events.go
+++ b/pkg/orchestrator/events.go
@@ -2,17 +2,21 @@ package orchestrator
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 )
 
 const (
 	EventTopicJobSubmission    models.EventTopic = "Submission"
+	EventTopicJobStateUpdate   models.EventTopic = "State Update"
 	EventTopicJobScheduling    models.EventTopic = "Scheduling"
 	EventTopicJobQueueing      models.EventTopic = "Queueing"
 	EventTopicExecutionTimeout models.EventTopic = "Exec Timeout"
 	EventTopicJobTimeout       models.EventTopic = "Job Timeout"
+	EventTopicExecution        models.EventTopic = "Execution"
 )
 
 const (
@@ -22,13 +26,15 @@ const (
 	jobStopRequestedMessage    = "Job requested to stop before completion"
 	jobExhaustedRetriesMessage = "Job failed because it has been retried too many times"
 	JobTimeoutMessage          = "Job timed out"
+	jobExecutionsFailedMessage = "Job failed because one or more executions failed"
 
+	execCompletedMessage                 = "Completed successfully"
+	execRunningMessage                   = "Running"
 	execStoppedByJobStopMessage          = "Execution stop requested because job has been stopped"
 	execStoppedByNodeUnhealthyMessage    = "Execution stop requested because node has disappeared"
 	execStoppedByNodeRejectedMessage     = "Execution stop requested because node has been rejected"
 	execStoppedByOversubscriptionMessage = "Execution stop requested because there are more executions than needed"
-	execRejectedByNodeMessage            = "Node responded to execution run request"
-	execFailedMessage                    = "Execution did not complete successfully"
+	execStoppedDueToJobFailureMessage    = "Execution stopped due to job failure"
 
 	executionTimeoutMessage = "Execution timed out"
 
@@ -56,6 +62,16 @@ func JobTranslatedEvent(old, new *models.Job) models.Event {
 	})
 }
 
+func JobStateUpdateEvent(new models.JobStateType, message ...string) models.Event {
+	eventMessage := new.String()
+	if len(message) > 0 && message[0] != "" {
+		eventMessage += fmt.Sprintf(". %s", strings.Join(message, ". "))
+	}
+	return *models.NewEvent(EventTopicJobStateUpdate).
+		WithMessage(eventMessage).
+		WithDetail("NewState", new.String())
+}
+
 func JobStoppedEvent(reason string) models.Event {
 	return event(EventTopicJobScheduling, jobStopRequestedMessage, map[string]string{
 		"Reason": reason,
@@ -74,6 +90,10 @@ func JobTimeoutEvent(timeout time.Duration) models.Event {
 	return *e
 }
 
+func JobExecutionsFailedEvent() models.Event {
+	return event(EventTopicJobScheduling, jobExecutionsFailedMessage, map[string]string{})
+}
+
 func JobQueueingEvent(reason string) models.Event {
 	message := jobQueuedMessage
 	if reason != "" {
@@ -82,6 +102,20 @@ func JobQueueingEvent(reason string) models.Event {
 	return *models.NewEvent(EventTopicJobQueueing).WithMessage(message)
 }
 
+func ExecCreatedEvent(execution *models.Execution) models.Event {
+	return *models.NewEvent(EventTopicJobScheduling).
+		WithMessage(fmt.Sprintf("Requested execution on %s", idgen.ShortNodeID(execution.NodeID))).
+		WithDetail("NodeID", execution.NodeID)
+}
+
+func ExecCompletedEvent() models.Event {
+	return *models.NewEvent(EventTopicExecution).WithMessage(execCompletedMessage)
+}
+
+func ExecRunningEvent() models.Event {
+	return *models.NewEvent(EventTopicExecution).WithMessage(execRunningMessage)
+}
+
 func ExecStoppedByJobStopEvent() models.Event {
 	return event(EventTopicJobScheduling, execStoppedByJobStopMessage, map[string]string{})
 }
@@ -105,3 +139,7 @@ func ExecStoppedByNodeRejectedEvent() models.Event {
 func ExecStoppedByOversubscriptionEvent() models.Event {
 	return event(EventTopicJobScheduling, execStoppedByOversubscriptionMessage, map[string]string{})
 }
+
+func ExecStoppedDueToJobFailureEvent() models.Event {
+	return *models.NewEvent(EventTopicJobScheduling).WithMessage(execStoppedDueToJobFailureMessage)
+}
diff --git a/pkg/orchestrator/planner/compute_forwarder.go b/pkg/orchestrator/planner/compute_forwarder.go
index 714fdf8e40..d966058f9b 100644
--- a/pkg/orchestrator/planner/compute_forwarder.go
+++ b/pkg/orchestrator/planner/compute_forwarder.go
@@ -3,10 +3,11 @@ package planner
 import (
 	"context"
 
+	"github.com/rs/zerolog/log"
+
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/rs/zerolog/log"
 )
 
 type ComputeForwarder struct {
@@ -170,7 +171,7 @@ func (s *ComputeForwarder) notifyCancel(ctx context.Context, message string, exe
 // updateExecutionState updates the execution state in the job store.
 func (s *ComputeForwarder) updateExecutionState(ctx context.Context, execution *models.Execution,
 	newState models.ExecutionStateType, expectedStates ...models.ExecutionStateType) {
-	err := s.jobStore.UpdateExecution(ctx, jobstore.UpdateExecutionRequest{
+	if err := s.jobStore.UpdateExecution(ctx, jobstore.UpdateExecutionRequest{
 		ExecutionID: execution.ID,
 		NewValues: models.Execution{
 			ComputeState: models.State[models.ExecutionStateType]{
@@ -180,8 +181,7 @@ func (s *ComputeForwarder) updateExecutionState(ctx context.Context, execution *
 		Condition: jobstore.UpdateExecutionCondition{
 			ExpectedStates: expectedStates,
 		},
-	})
-	if err != nil {
+	}); err != nil {
 		log.Ctx(ctx).Error().Err(err).Msgf("Failed to update execution %s to state %s", execution.ID, newState)
 	}
 }
diff --git a/pkg/orchestrator/planner/event_emitter.go b/pkg/orchestrator/planner/event_emitter.go
index 6b7c2b5aca..e7ea6bea28 100644
--- a/pkg/orchestrator/planner/event_emitter.go
+++ b/pkg/orchestrator/planner/event_emitter.go
@@ -2,6 +2,7 @@ package planner
 
 import (
 	"context"
+	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
@@ -41,9 +42,9 @@ func (s *EventEmitter) Process(ctx context.Context, plan *models.Plan) error {
 		s.eventEmitter.EmitEventSilently(ctx, models.JobEvent{
 			SourceNodeID: s.id,
 			JobID:        plan.Job.ID,
-			Status:       plan.Event.Message,
+			Status:       plan.UpdateMessage,
 			EventName:    eventName,
-			EventTime:    plan.Event.Timestamp,
+			EventTime:    time.Now(),
 		})
 	}
 	return nil
diff --git a/pkg/orchestrator/planner/logging_planner.go b/pkg/orchestrator/planner/logging_planner.go
index 834bdf7994..f14dab35d5 100644
--- a/pkg/orchestrator/planner/logging_planner.go
+++ b/pkg/orchestrator/planner/logging_planner.go
@@ -3,10 +3,11 @@ package planner
 import (
 	"context"
 
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
 )
 
 type LoggingPlanner struct {
@@ -18,8 +19,14 @@ func NewLoggingPlanner() *LoggingPlanner {
 
 func (s *LoggingPlanner) Process(ctx context.Context, plan *models.Plan) error {
 	dict := zerolog.Dict()
-	for key, value := range plan.Event.Details {
-		dict = dict.Str(key, value)
+	var eventMessage string
+	var delim string
+	for _, event := range plan.JobEvents {
+		for key, value := range event.Details {
+			dict = dict.Str(key, value)
+		}
+		eventMessage += delim + event.Message
+		delim = ". "
 	}
 
 	level := zerolog.TraceLevel
@@ -34,7 +41,11 @@ func (s *LoggingPlanner) Process(ctx context.Context, plan *models.Plan) error {
 	default:
 	}
 
-	log.Ctx(ctx).WithLevel(level).Dict("Details", dict).Str("Event", plan.Event.Message).Str("JobID", plan.Job.ID).Msg(message)
+	log.Ctx(ctx).WithLevel(level).
+		Dict("Details", dict).
+		Str("Event", eventMessage).
+		Str("JobID", plan.Job.ID).
+		Msg(message)
 	return nil
 }
 
diff --git a/pkg/orchestrator/planner/state_updater.go b/pkg/orchestrator/planner/state_updater.go
index 6ae15e5dcf..c81346e2d7 100644
--- a/pkg/orchestrator/planner/state_updater.go
+++ b/pkg/orchestrator/planner/state_updater.go
@@ -25,6 +25,8 @@ func NewStateUpdater(store jobstore.Store) *StateUpdater {
 }
 
 // Process updates the state of the executions in the plan according to the scheduler's desired state.
+//
+//nolint:gocyclo
 func (s *StateUpdater) Process(ctx context.Context, plan *models.Plan) error {
 	// If there are no new or updated executions
 	// and the job state is not being updated, there is nothing to do.
@@ -48,15 +50,14 @@ func (s *StateUpdater) Process(ctx context.Context, plan *models.Plan) error {
 
 	// Create new executions
 	for _, exec := range plan.NewExecutions {
-		err = s.store.CreateExecution(txContext, *exec, plan.Event)
-		if err != nil {
+		if err = s.store.CreateExecution(txContext, *exec); err != nil {
 			return err
 		}
 	}
 
 	// Update existing executions
 	for _, u := range plan.UpdatedExecutions {
-		err = s.store.UpdateExecution(txContext, jobstore.UpdateExecutionRequest{
+		if err = s.store.UpdateExecution(txContext, jobstore.UpdateExecutionRequest{
 			ExecutionID: u.Execution.ID,
 			NewValues: models.Execution{
 				DesiredState: models.State[models.ExecutionDesiredStateType]{
@@ -67,35 +68,46 @@ func (s *StateUpdater) Process(ctx context.Context, plan *models.Plan) error {
 			Condition: jobstore.UpdateExecutionCondition{
 				ExpectedRevision: u.Execution.Revision,
 			},
-			Event: u.Event,
-		})
-		if err != nil {
+		}); err != nil {
 			return err
 		}
 	}
 
 	// Update job state if necessary
 	if !plan.DesiredJobState.IsUndefined() {
-		err = s.store.UpdateJobState(txContext, jobstore.UpdateJobStateRequest{
+		if err = s.store.UpdateJobState(txContext, jobstore.UpdateJobStateRequest{
 			JobID:    plan.Job.ID,
 			NewState: plan.DesiredJobState,
-			Event:    plan.Event,
+			Message:  plan.UpdateMessage,
 			Condition: jobstore.UpdateJobCondition{
 				ExpectedRevision: plan.Job.Revision,
 			},
-		})
-		if err != nil {
+		}); err != nil {
 			return err
 		}
 	}
 
 	// Create follow-up evaluations, if any
 	for _, eval := range plan.NewEvaluations {
-		err = s.store.CreateEvaluation(txContext, *eval)
-		if err != nil {
+		if err = s.store.CreateEvaluation(txContext, *eval); err != nil {
 			return err
 		}
 	}
+
+	// Add history events
+	if len(plan.JobEvents) > 0 {
+		if err = s.store.AddJobHistory(txContext, plan.Job.ID, plan.JobEvents...); err != nil {
+			return err
+		}
+	}
+	if len(plan.ExecutionEvents) > 0 {
+		for executionID, events := range plan.ExecutionEvents {
+			if err = s.store.AddExecutionHistory(txContext, plan.Job.ID, executionID, events...); err != nil {
+				return err
+			}
+		}
+	}
+
 	return txContext.Commit()
 }
 
diff --git a/pkg/orchestrator/scheduler/batch_service_job.go b/pkg/orchestrator/scheduler/batch_service_job.go
index 68cc962d7d..a28d588705 100644
--- a/pkg/orchestrator/scheduler/batch_service_job.go
+++ b/pkg/orchestrator/scheduler/batch_service_job.go
@@ -77,7 +77,7 @@ func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *mode
 
 	// early exit if the job is stopped
 	if job.IsTerminal() {
-		nonTerminalExecs.markStopped(orchestrator.ExecStoppedByJobStopEvent(), plan)
+		nonTerminalExecs.markStopped(plan, orchestrator.ExecStoppedByJobStopEvent())
 		return b.planner.Process(ctx, plan)
 	}
 
@@ -92,7 +92,7 @@ func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *mode
 
 	// Mark executions that are running on nodes that are not healthy as failed
 	nonTerminalExecs, lost := nonTerminalExecs.filterByNodeHealth(nodeInfos)
-	lost.markStopped(orchestrator.ExecStoppedByNodeUnhealthyEvent(), plan)
+	lost.markStopped(plan, orchestrator.ExecStoppedByNodeUnhealthyEvent())
 	allFailedExecs = allFailedExecs.union(lost)
 
 	nonTerminalExecs, allFailedExecs = b.handleTimeouts(plan, nonTerminalExecs, allFailedExecs)
@@ -108,8 +108,8 @@ func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *mode
 
 	// Approve/Reject nodes
 	execsByApprovalStatus := nonTerminalExecs.filterByApprovalStatus(desiredRemainingCount)
-	execsByApprovalStatus.toApprove.markApproved(plan)
-	execsByApprovalStatus.toReject.markStopped(orchestrator.ExecStoppedByNodeRejectedEvent(), plan)
+	execsByApprovalStatus.toApprove.markApproved(plan, orchestrator.ExecRunningEvent())
+	execsByApprovalStatus.toReject.markStopped(plan, orchestrator.ExecStoppedByNodeRejectedEvent())
 
 	// create new executions if needed
 	remainingExecutionCount := desiredRemainingCount - execsByApprovalStatus.activeCount()
@@ -130,18 +130,18 @@ func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *mode
 
 	// if the plan's job state if terminal, stop all active executions
 	if plan.IsJobFailed() {
-		nonTerminalExecs.markStopped(plan.Event, plan)
+		nonTerminalExecs.markStopped(plan, orchestrator.ExecStoppedDueToJobFailureEvent())
 	} else {
 		// stop executions if we over-subscribed and exceeded the desired number of executions
 		_, overSubscriptions := execsByApprovalStatus.running.filterByOverSubscriptions(desiredRemainingCount)
-		overSubscriptions.markStopped(orchestrator.ExecStoppedByOversubscriptionEvent(), plan)
+		overSubscriptions.markStopped(plan, orchestrator.ExecStoppedByOversubscriptionEvent())
 
 		// Check the job's state and update it accordingly.
 		if desiredRemainingCount <= 0 {
 			// If there are no remaining tasks to be done, mark the job as completed.
-			plan.MarkJobCompleted()
+			plan.MarkJobCompleted(orchestrator.JobStateUpdateEvent(models.JobStateTypeCompleted))
 		}
-		plan.MarkJobRunningIfEligible()
+		plan.MarkJobRunningIfEligible(orchestrator.JobStateUpdateEvent(models.JobStateTypeRunning))
 	}
 	return b.planner.Process(ctx, plan)
 }
@@ -165,7 +165,7 @@ func (b *BatchServiceJobScheduler) handleTimeouts(
 			expirationTime := b.clock.Now().Add(-timeout)
 			var timedOut execSet
 			nonTerminalExecs, timedOut = nonTerminalExecs.filterByExecutionTimeout(expirationTime)
-			timedOut.markStopped(orchestrator.ExecStoppedByExecutionTimeoutEvent(timeout), plan)
+			timedOut.markStopped(plan, orchestrator.ExecStoppedByExecutionTimeoutEvent(timeout))
 			allFailedExecs = allFailedExecs.union(timedOut)
 		}
 	}
@@ -210,7 +210,7 @@ func (b *BatchServiceJobScheduler) createMissingExecs(
 			DesiredState: models.NewExecutionDesiredState(models.ExecutionDesiredStatePending),
 		}
 		execution.Normalize()
-		plan.AppendExecution(execution)
+		plan.AppendExecution(execution, orchestrator.ExecCreatedEvent(execution))
 	}
 
 	if len(matching) < remainingExecutionCount {
diff --git a/pkg/orchestrator/scheduler/daemon_job.go b/pkg/orchestrator/scheduler/daemon_job.go
index a6c08538c3..5d22f1b1af 100644
--- a/pkg/orchestrator/scheduler/daemon_job.go
+++ b/pkg/orchestrator/scheduler/daemon_job.go
@@ -58,7 +58,7 @@ func (b *DaemonJobScheduler) Process(ctx context.Context, evaluation *models.Eva
 
 	// early exit if the job is stopped
 	if job.IsTerminal() {
-		nonTerminalExecs.markStopped(orchestrator.ExecStoppedByJobStopEvent(), plan)
+		nonTerminalExecs.markStopped(plan, orchestrator.ExecStoppedByJobStopEvent())
 		return b.planner.Process(ctx, plan)
 	}
 
@@ -70,7 +70,7 @@ func (b *DaemonJobScheduler) Process(ctx context.Context, evaluation *models.Eva
 
 	// Mark executions that are running on nodes that are not healthy as failed
 	_, lost := nonTerminalExecs.filterByNodeHealth(nodeInfos)
-	lost.markStopped(orchestrator.ExecStoppedByNodeUnhealthyEvent(), plan)
+	lost.markStopped(plan, orchestrator.ExecStoppedByNodeUnhealthyEvent())
 
 	// Look for new matching nodes and create new executions every time we evaluate the job
 	_, err = b.createMissingExecs(ctx, &job, plan, existingExecs)
@@ -78,7 +78,8 @@ func (b *DaemonJobScheduler) Process(ctx context.Context, evaluation *models.Eva
 		return fmt.Errorf("failed to find/create missing executions: %w", err)
 	}
 
-	plan.MarkJobRunningIfEligible()
+	runningEvent := orchestrator.JobStateUpdateEvent(models.JobStateTypeRunning)
+	plan.MarkJobRunningIfEligible(runningEvent)
 	return b.planner.Process(ctx, plan)
 }
 
@@ -116,7 +117,7 @@ func (b *DaemonJobScheduler) createMissingExecs(
 		newExecs[execution.ID] = execution
 	}
 	for _, exec := range newExecs {
-		plan.AppendExecution(exec)
+		plan.AppendExecution(exec, orchestrator.ExecCreatedEvent(exec))
 	}
 	return newExecs, nil
 }
diff --git a/pkg/orchestrator/scheduler/ops_job.go b/pkg/orchestrator/scheduler/ops_job.go
index a5abf479f4..66470b2d02 100644
--- a/pkg/orchestrator/scheduler/ops_job.go
+++ b/pkg/orchestrator/scheduler/ops_job.go
@@ -2,7 +2,6 @@ package scheduler
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"github.com/benbjohnson/clock"
@@ -68,7 +67,7 @@ func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evalua
 
 	// early exit if the job is stopped
 	if job.IsTerminal() {
-		nonTerminalExecs.markStopped(orchestrator.ExecStoppedByJobStopEvent(), plan)
+		nonTerminalExecs.markStopped(plan, orchestrator.ExecStoppedByJobStopEvent())
 		return b.planner.Process(ctx, plan)
 	}
 
@@ -83,7 +82,7 @@ func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evalua
 
 	// Mark executions that are running on nodes that are not healthy as failed
 	nonTerminalExecs, lost := nonTerminalExecs.filterByNodeHealth(nodeInfos)
-	lost.markStopped(orchestrator.ExecStoppedByNodeUnhealthyEvent(), plan)
+	lost.markStopped(plan, orchestrator.ExecStoppedByNodeUnhealthyEvent())
 	allFailedExecs = allFailedExecs.union(lost)
 
 	nonTerminalExecs, allFailedExecs = b.handleTimeouts(plan, nonTerminalExecs, allFailedExecs)
@@ -92,27 +91,25 @@ func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evalua
 	// the first time we are evaluating the job
 	var newExecs execSet
 	if len(existingExecs) == 0 {
-		newExecs, err = b.createMissingExecs(ctx, &job, plan)
-		if err != nil {
-			b.handleFailure(nonTerminalExecs, allFailedExecs, plan, err)
-			return b.planner.Process(ctx, plan)
+		if newExecs, err = b.createMissingExecs(ctx, &job, plan); err != nil {
+			return err // internal error
 		}
 	}
 
 	// if the plan's job state is failed, stop all active executions
 	if plan.IsJobFailed() {
-		nonTerminalExecs.markStopped(plan.Event, plan)
+		nonTerminalExecs.markStopped(plan, orchestrator.ExecStoppedDueToJobFailureEvent())
 	} else {
 		// mark job as completed if there are no more active or new executions
 		if len(nonTerminalExecs) == 0 && len(newExecs) == 0 {
 			if len(allFailedExecs) > 0 {
-				b.handleFailure(nonTerminalExecs, allFailedExecs, plan, errors.New(""))
+				plan.MarkJobFailed(orchestrator.JobExecutionsFailedEvent())
 			} else {
-				plan.MarkJobCompleted()
+				plan.MarkJobCompleted(orchestrator.JobStateUpdateEvent(models.JobStateTypeCompleted))
 			}
 		}
 	}
-	plan.MarkJobRunningIfEligible()
+	plan.MarkJobRunningIfEligible(orchestrator.JobStateUpdateEvent(models.JobStateTypeRunning))
 	return b.planner.Process(ctx, plan)
 }
 
@@ -133,7 +130,7 @@ func (b *OpsJobScheduler) handleTimeouts(
 		expirationTime := b.clock.Now().Add(-timeout)
 		var timedOut execSet
 		nonTerminalExecs, timedOut = nonTerminalExecs.filterByExecutionTimeout(expirationTime)
-		timedOut.markStopped(orchestrator.ExecStoppedByExecutionTimeoutEvent(timeout), plan)
+		timedOut.markStopped(plan, orchestrator.ExecStoppedByExecutionTimeoutEvent(timeout))
 		allFailedExecs = allFailedExecs.union(timedOut)
 	}
 	return nonTerminalExecs, allFailedExecs
@@ -141,15 +138,17 @@ func (b *OpsJobScheduler) handleTimeouts(
 
 func (b *OpsJobScheduler) createMissingExecs(
 	ctx context.Context, job *models.Job, plan *models.Plan) (execSet, error) {
-	newExecs := execSet{}
-	nodes, _, err := b.selector.MatchingNodes(ctx, job)
+	matching, rejected, err := b.selector.MatchingNodes(ctx, job)
 	if err != nil {
-		return newExecs, err
+		return nil, err
 	}
-	if len(nodes) == 0 {
-		return nil, orchestrator.ErrNoMatchingNodes{}
+	if len(matching) == 0 {
+		plan.MarkJobFailed(
+			models.EventFromError(orchestrator.EventTopicJobScheduling, orchestrator.NewErrNoMatchingNodes(rejected)))
+		return nil, nil
 	}
-	for _, node := range nodes {
+	newExecs := execSet{}
+	for _, node := range matching {
 		execution := &models.Execution{
 			JobID:        job.ID,
 			Job:          job,
@@ -161,22 +160,11 @@ func (b *OpsJobScheduler) createMissingExecs(
 			NodeID:       node.NodeInfo.ID(),
 		}
 		execution.Normalize()
+		plan.AppendExecution(execution, orchestrator.ExecCreatedEvent(execution))
 		newExecs[execution.ID] = execution
 	}
-	for _, exec := range newExecs {
-		plan.AppendExecution(exec)
-	}
 	return newExecs, nil
 }
 
-func (b *OpsJobScheduler) handleFailure(nonTerminalExecs execSet, failed execSet, plan *models.Plan, err error) {
-	// mark all non-terminal executions as failed
-	nonTerminalExecs.markStopped(plan.Event, plan)
-
-	// mark the job as failed, using the error message of the latest failed execution, if any, or use
-	// the error message passed by the scheduler
-	plan.MarkJobFailed(plan.Event)
-}
-
 // compile-time assertion that OpsJobScheduler satisfies the Scheduler interface
 var _ orchestrator.Scheduler = &OpsJobScheduler{}
diff --git a/pkg/orchestrator/scheduler/types.go b/pkg/orchestrator/scheduler/types.go
index 12f6d287c2..e7b1a8316f 100644
--- a/pkg/orchestrator/scheduler/types.go
+++ b/pkg/orchestrator/scheduler/types.go
@@ -6,8 +6,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/rs/zerolog/log"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
 )
 
 // execSet is a set of executions with a series of helper functions defined
@@ -208,16 +209,16 @@ func (set execSet) filterByApprovalStatus(desiredCount int) executionsByApproval
 }
 
 // markStopped
-func (set execSet) markStopped(event models.Event, plan *models.Plan) {
+func (set execSet) markStopped(plan *models.Plan, event models.Event) {
 	for _, exec := range set {
 		plan.AppendStoppedExecution(exec, event)
 	}
 }
 
 // markStopped
-func (set execSet) markApproved(plan *models.Plan) {
+func (set execSet) markApproved(plan *models.Plan, event models.Event) {
 	for _, exec := range set {
-		plan.AppendApprovedExecution(exec)
+		plan.AppendApprovedExecution(exec, event)
 	}
 }
 
diff --git a/pkg/orchestrator/selection/selector/node_selector.go b/pkg/orchestrator/selection/selector/node_selector.go
index e0c95ee30b..99916b51f7 100644
--- a/pkg/orchestrator/selection/selector/node_selector.go
+++ b/pkg/orchestrator/selection/selector/node_selector.go
@@ -2,7 +2,6 @@ package selector
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sort"
 
@@ -88,10 +87,6 @@ func (n NodeSelector) rankAndFilterNodes(
 		return true
 	})
 
-	if len(nodeStates) == 0 {
-		return nil, nil, errors.New("unable to find any connected and approved nodes")
-	}
-
 	// extract the nodeInfo from the slice of node states for ranking
 	nodeInfos := make([]models.NodeInfo, 0, len(nodeStates))
 	for _, ns := range nodeStates {
diff --git a/pkg/publicapi/apimodels/job.go b/pkg/publicapi/apimodels/job.go
index e07df27107..91652e8625 100644
--- a/pkg/publicapi/apimodels/job.go
+++ b/pkg/publicapi/apimodels/job.go
@@ -107,7 +107,6 @@ type ListJobHistoryRequest struct {
 	Since       int64  `query:"since" validate:"min=0"`
 	EventType   string `query:"event_type" validate:"omitempty,oneof=all job execution"`
 	ExecutionID string `query:"execution_id" validate:"omitempty"`
-	NodeID      string `query:"node_id" validate:"omitempty"`
 }
 
 // ToHTTPRequest is used to convert the request to an HTTP request
@@ -123,9 +122,6 @@ func (o *ListJobHistoryRequest) ToHTTPRequest() *HTTPRequest {
 	if o.ExecutionID != "" {
 		r.Params.Set("execution_id", o.ExecutionID)
 	}
-	if o.NodeID != "" {
-		r.Params.Set("node_id", o.NodeID)
-	}
 	return r
 }
 
diff --git a/pkg/publicapi/endpoint/orchestrator/job.go b/pkg/publicapi/endpoint/orchestrator/job.go
index 0f60c02cdd..ba362f858a 100644
--- a/pkg/publicapi/endpoint/orchestrator/job.go
+++ b/pkg/publicapi/endpoint/orchestrator/job.go
@@ -103,6 +103,7 @@ func (e *Endpoint) getJob(c echo.Context) error { //nolint: gocyclo
 			for i := range history {
 				response.History.Items[i] = &history[i]
 			}
+			backwardCompatibleHistoryIfNecessary(c, response.History.Items)
 		case "executions":
 			// ignore if user requested executions twice
 			if response.Executions != nil {
@@ -272,7 +273,6 @@ func (e *Endpoint) stopJob(c echo.Context) error {
 //	@Param			since			query		string	false	"Only return history since this time"
 //	@Param			event_type		query		string	false	"Only return history of this event type"
 //	@Param			execution_id	query		string	false	"Only return history of this execution ID"
-//	@Param			node_id			query		string	false	"Only return history of this node ID"
 //	@Param			next_token		query		string	false	"Token to get the next page of the jobs"
 //	@Success		200				{object}	apimodels.ListJobHistoryResponse
 //	@Failure		400				{object}	string
@@ -294,7 +294,6 @@ func (e *Endpoint) jobHistory(c echo.Context) error {
 		ExcludeExecutionLevel: args.EventType == "job",
 		ExcludeJobLevel:       args.EventType == "execution",
 		ExecutionID:           args.ExecutionID,
-		NodeID:                args.NodeID,
 		Limit:                 args.Limit,
 		NextToken:             args.NextToken,
 	}
@@ -314,6 +313,7 @@ func (e *Endpoint) jobHistory(c echo.Context) error {
 	for i := range jobHistoryQueryResponse.JobHistory {
 		res.Items[i] = &jobHistoryQueryResponse.JobHistory[i]
 	}
+	backwardCompatibleHistoryIfNecessary(c, res.Items)
 
 	return c.JSON(http.StatusOK, res)
 }
diff --git a/pkg/publicapi/endpoint/orchestrator/utils.go b/pkg/publicapi/endpoint/orchestrator/utils.go
index 9ab7f1067c..13405621c1 100644
--- a/pkg/publicapi/endpoint/orchestrator/utils.go
+++ b/pkg/publicapi/endpoint/orchestrator/utils.go
@@ -7,6 +7,10 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/labstack/echo/v4"
 	"k8s.io/apimachinery/pkg/labels"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
+	"github.com/bacalhau-project/bacalhau/pkg/version"
 )
 
 var upgrader = websocket.Upgrader{}
@@ -23,3 +27,26 @@ func parseLabels(c echo.Context) (labels.Selector, error) {
 	}
 	return selector, nil
 }
+
+// backwardCompatibleHistoryIfNecessary sets the state change fields to non-nil for backward compatibility
+// with v1.4.x clients. Otherwise, nil exceptions will be thrown when the client tries to describe job or list history.
+//
+//nolint:staticcheck
+func backwardCompatibleHistoryIfNecessary(c echo.Context, items []*models.JobHistory) {
+	clientVersionStr := c.Request().Header.Get(apimodels.HTTPHeaderBacalhauGitVersion)
+
+	// If the client version is not explicit or is greater than or equal to v1.5, we don't need to do anything.
+	if !version.IsVersionExplicit(clientVersionStr) || clientVersionStr >= "v1.5." {
+		return
+	}
+
+	// If the client version is not explicit, or is less than v1.5, we need to set the state change fields to nil.
+	for i := range items {
+		history := items[i]
+		if history.Type == models.JobHistoryTypeJobLevel {
+			history.JobState = &models.StateChange[models.JobStateType]{}
+		} else {
+			history.ExecutionState = &models.StateChange[models.ExecutionStateType]{}
+		}
+	}
+}
diff --git a/pkg/publicapi/middleware/version.go b/pkg/publicapi/middleware/version.go
index 34666e849f..a2dc3e458d 100644
--- a/pkg/publicapi/middleware/version.go
+++ b/pkg/publicapi/middleware/version.go
@@ -95,9 +95,7 @@ func VersionCheckMiddleware(serverVersion, minVersion semver.Version) echo.Middl
 	return func(next echo.HandlerFunc) echo.HandlerFunc {
 		return func(c echo.Context) error {
 			clientVersionStr := c.Request().Header.Get(apimodels.HTTPHeaderBacalhauGitVersion)
-			if clientVersionStr == "" ||
-				clientVersionStr == version.DevelopmentGitVersion ||
-				clientVersionStr == version.UnknownGitVersion {
+			if !version.IsVersionExplicit(clientVersionStr) {
 				// allow the request to pass through
 				return next(c)
 			}
diff --git a/pkg/version/version.go b/pkg/version/version.go
index a9e39031fe..04a1d45f4a 100644
--- a/pkg/version/version.go
+++ b/pkg/version/version.go
@@ -36,6 +36,14 @@ var (
 	GITVERSION = DevelopmentGitVersion
 )
 
+// IsVersionExplicit checks if the client version is a specific, known version.
+// It returns false for empty, development, or unknown versions.
+func IsVersionExplicit(clientVersionStr string) bool {
+	return clientVersionStr != "" &&
+		clientVersionStr != DevelopmentGitVersion &&
+		clientVersionStr != UnknownGitVersion
+}
+
 // Get returns the overall codebase version. It's for detecting what code a binary was built from.
 func Get() *models.BuildVersionInfo {
 	revision, revisionTime, err := getBuildInformation()


Test patch:
diff --git a/pkg/jobstore/boltdb/store_test.go b/pkg/jobstore/boltdb/store_test.go
index 3972fbb63d..2fa13cb94f 100644
--- a/pkg/jobstore/boltdb/store_test.go
+++ b/pkg/jobstore/boltdb/store_test.go
@@ -122,8 +122,8 @@ func (s *BoltJobstoreTestSuite) SetupTest() {
 		job.Type = fixture.jobType
 		job.Labels = fixture.tags
 		job.Namespace = fixture.client
-		err := s.store.CreateJob(s.ctx, *job, models.Event{})
-		s.Require().NoError(err)
+		s.Require().NoError(s.store.CreateJob(s.ctx, *job))
+		s.Require().NoError(s.store.AddJobHistory(s.ctx, fixture.id, *models.NewEvent("test").WithMessage("job created")))
 
 		for i, state := range fixture.jobStates {
 			s.clock.Add(1 * time.Second)
@@ -140,10 +140,9 @@ func (s *BoltJobstoreTestSuite) SetupTest() {
 					ExpectedState:    oldState,
 					ExpectedRevision: uint64(i + 1),
 				},
-				Event: models.Event{},
 			}
-			err = s.store.UpdateJobState(s.ctx, request)
-			s.Require().NoError(err)
+			s.Require().NoError(s.store.UpdateJobState(s.ctx, request))
+			s.Require().NoError(s.store.AddJobHistory(s.ctx, fixture.id, *models.NewEvent("test").WithMessage(state.String())))
 		}
 
 		for _, executionStates := range fixture.executions {
@@ -153,8 +152,9 @@ func (s *BoltJobstoreTestSuite) SetupTest() {
 			// clear out CreateTime and ModifyTime from the mocked execution to let the job store fill those
 			execution.CreateTime = 0
 			execution.ModifyTime = 0
-			err = s.store.CreateExecution(s.ctx, *execution, models.Event{})
-			s.Require().NoError(err)
+			s.Require().NoError(s.store.CreateExecution(s.ctx, *execution))
+			s.Require().NoError(s.store.AddExecutionHistory(s.ctx, fixture.id, execution.ID, *models.NewEvent("test").WithMessage("execution created")))
+
 			for i, state := range executionStates {
 
 				s.clock.Add(1 * time.Second)
@@ -175,11 +175,10 @@ func (s *BoltJobstoreTestSuite) SetupTest() {
 						ExpectedRevision: uint64(i + 1),
 					},
 					NewValues: *execution,
-					Event:     models.Event{},
 				}
 
-				err = s.store.UpdateExecution(s.ctx, request)
-				s.Require().NoError(err)
+				s.Require().NoError(s.store.UpdateExecution(s.ctx, request))
+				s.Require().NoError(s.store.AddExecutionHistory(s.ctx, fixture.id, execution.ID, *models.NewEvent("test").WithMessage(state.String())))
 			}
 		}
 
@@ -299,31 +298,6 @@ func (s *BoltJobstoreTestSuite) TestExecutionFilteredJobHistory() {
 	}
 }
 
-func (s *BoltJobstoreTestSuite) TestNodeFilteredJobHistory() {
-	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{})
-	require.NoError(s.T(), err)
-
-	var nodeID string
-	for _, h := range jobHistoryQueryResponse.JobHistory {
-		if h.NodeID != "" {
-			nodeID = h.NodeID
-			break
-		}
-	}
-	require.NotEmpty(s.T(), nodeID, "failed to find node ID")
-
-	options := jobstore.JobHistoryQuery{
-		NodeID: nodeID,
-	}
-
-	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", options)
-	require.NoError(s.T(), err, "failed to get job history")
-
-	for _, h := range jobHistoryQueryResponse.JobHistory {
-		require.Equal(s.T(), nodeID, h.NodeID)
-	}
-}
-
 func (s *BoltJobstoreTestSuite) TestLevelFilteredJobHistory() {
 	jobOptions := jobstore.JobHistoryQuery{
 		ExcludeExecutionLevel: true,
@@ -335,7 +309,6 @@ func (s *BoltJobstoreTestSuite) TestLevelFilteredJobHistory() {
 	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobOptions)
 	s.Require().NoError(err, "failed to get job history")
 	s.Require().Equal(4, len(jobHistoryQueryResponse.JobHistory))
-	s.Require().Equal(models.JobStateTypePending, jobHistoryQueryResponse.JobHistory[1].JobState.New)
 
 	count := lo.Reduce(jobHistoryQueryResponse.JobHistory, func(agg int, item models.JobHistory, _ int) int {
 		if item.Type == models.JobHistoryTypeJobLevel {
@@ -348,7 +321,6 @@ func (s *BoltJobstoreTestSuite) TestLevelFilteredJobHistory() {
 	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", execOptions)
 	s.Require().NoError(err, "failed to get job history")
 	s.Require().Equal(4, len(jobHistoryQueryResponse.JobHistory))
-	s.Require().Equal(models.ExecutionStateAskForBid, jobHistoryQueryResponse.JobHistory[1].ExecutionState.New)
 
 	count = lo.Reduce(jobHistoryQueryResponse.JobHistory, func(agg int, item models.JobHistory, _ int) int {
 		if item.Type == models.JobHistoryTypeExecutionLevel {
@@ -507,7 +479,7 @@ func (s *BoltJobstoreTestSuite) TestDeleteJob() {
 	job.ID = "deleteme"
 	job.Namespace = "client1"
 
-	err := s.store.CreateJob(s.ctx, *job, models.Event{})
+	err := s.store.CreateJob(s.ctx, *job)
 	s.Require().NoError(err)
 
 	err = s.store.DeleteJob(s.ctx, job.ID)
@@ -526,8 +498,8 @@ func (s *BoltJobstoreTestSuite) TestGetJob() {
 func (s *BoltJobstoreTestSuite) TestCreateExecution() {
 	job := mock.Job()
 	execution := mock.ExecutionForJob(job)
-	s.Require().NoError(s.store.CreateJob(s.ctx, *job, models.Event{}))
-	s.Require().NoError(s.store.CreateExecution(s.ctx, *execution, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(s.ctx, *job))
+	s.Require().NoError(s.store.CreateExecution(s.ctx, *execution))
 
 	// Ensure that the execution is created
 	exec, err := s.store.GetExecutions(s.ctx, jobstore.GetExecutionsOptions{
@@ -554,7 +526,7 @@ func (s *BoltJobstoreTestSuite) TestGetExecutions() {
 	})
 	s.Require().NoError(err)
 	s.NotNil(state)
-	s.Equal(len(state), 1)
+	s.Equal(1, len(state))
 	s.Nil(state[0].Job)
 
 	state, err = s.store.GetExecutions(s.ctx, jobstore.GetExecutionsOptions{
@@ -715,7 +687,7 @@ func (s *BoltJobstoreTestSuite) TestShortIDs() {
 	s.Require().IsType(err, &bacerrors.JobNotFound{})
 
 	// Create and fetch the single entry
-	err = s.store.CreateJob(s.ctx, *job, models.Event{})
+	err = s.store.CreateJob(s.ctx, *job)
 	s.Require().NoError(err)
 
 	j, err := s.store.GetJob(s.ctx, shortString)
@@ -724,7 +696,7 @@ func (s *BoltJobstoreTestSuite) TestShortIDs() {
 
 	// Add a record that will also match and expect an appropriate error
 	job.ID = uuidString2
-	err = s.store.CreateJob(s.ctx, *job, models.Event{})
+	err = s.store.CreateJob(s.ctx, *job)
 	s.Require().NoError(err)
 
 	_, err = s.store.GetJob(s.ctx, shortString)
@@ -746,7 +718,7 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 	var execution models.Execution
 
 	s.Run("job create event", func() {
-		err := s.store.CreateJob(s.ctx, *job, models.Event{})
+		err := s.store.CreateJob(s.ctx, *job)
 		s.Require().NoError(err)
 
 		// Read an event, it should be a jobcreate
@@ -764,7 +736,7 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 		execution = *mock.Execution()
 		execution.JobID = "10"
 		execution.ComputeState = models.State[models.ExecutionStateType]{StateType: models.ExecutionStateNew}
-		err := s.store.CreateExecution(s.ctx, execution, models.Event{})
+		err := s.store.CreateExecution(s.ctx, execution)
 		s.Require().NoError(err)
 
 		// Read an event, it should be a ExecutionForJob Create
@@ -780,7 +752,6 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 			Condition: jobstore.UpdateJobCondition{
 				ExpectedState: models.JobStateTypePending,
 			},
-			Event: models.Event{Message: "event test"},
 		}
 		_ = s.store.UpdateJobState(s.ctx, request)
 		ev := <-watcher.Channel()
@@ -797,7 +768,6 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 				ExpectedStates: []models.ExecutionStateType{models.ExecutionStateNew},
 			},
 			NewValues: execution,
-			Event:     models.Event{Message: "event test"},
 		})
 		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.UpdateEvent)
@@ -852,8 +822,8 @@ func (s *BoltJobstoreTestSuite) TestTransactionsWithTxContext() {
 	job := mock.Job()
 	execution := mock.ExecutionForJob(job)
 	evaluation := mock.EvalForJob(job)
-	s.Require().NoError(s.store.CreateJob(txCtx, *job, models.Event{}))
-	s.Require().NoError(s.store.CreateExecution(txCtx, *execution, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(txCtx, *job))
+	s.Require().NoError(s.store.CreateExecution(txCtx, *execution))
 	s.Require().NoError(s.store.CreateEvaluation(txCtx, *evaluation))
 
 	// Commit the transaction
@@ -890,8 +860,8 @@ func (s *BoltJobstoreTestSuite) TestTransactionsWithTxContextRollback() {
 	job := mock.Job()
 	execution := mock.ExecutionForJob(job)
 	evaluation := mock.EvalForJob(job)
-	s.Require().NoError(s.store.CreateJob(txCtx, *job, models.Event{}))
-	s.Require().NoError(s.store.CreateExecution(txCtx, *execution, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(txCtx, *job))
+	s.Require().NoError(s.store.CreateExecution(txCtx, *execution))
 	s.Require().NoError(s.store.CreateEvaluation(txCtx, *evaluation))
 
 	// Rollback the transaction
@@ -923,8 +893,8 @@ func (s *BoltJobstoreTestSuite) TestTransactionsWithTxContextCancellation() {
 	job := mock.Job()
 	execution := mock.ExecutionForJob(job)
 	evaluation := mock.EvalForJob(job)
-	s.Require().NoError(s.store.CreateJob(txCtx, *job, models.Event{}))
-	s.Require().NoError(s.store.CreateExecution(txCtx, *execution, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(txCtx, *job))
+	s.Require().NoError(s.store.CreateExecution(txCtx, *execution))
 	s.Require().NoError(s.store.CreateEvaluation(txCtx, *evaluation))
 
 	// cancel the context
@@ -950,14 +920,14 @@ func (s *BoltJobstoreTestSuite) TestTransactionsWithTxContextCancellation() {
 func (s *BoltJobstoreTestSuite) TestTransactionsReadDuringWrite() {
 	// Create a job outside the transaction
 	oldJob := mock.Job()
-	s.Require().NoError(s.store.CreateJob(s.ctx, *oldJob, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(s.ctx, *oldJob))
 
 	txCtx, err := s.store.BeginTx(s.ctx)
 	s.Require().NoError(err)
 	s.Require().NotNil(txCtx)
 
 	job := mock.Job()
-	s.Require().NoError(s.store.CreateJob(txCtx, *job, models.Event{}))
+	s.Require().NoError(s.store.CreateJob(txCtx, *job))
 
 	// make sure we can read existing data during transaction
 	readOldJob, err := s.store.GetJob(txCtx, oldJob.ID)
diff --git a/pkg/models/plan_test.go b/pkg/models/plan_test.go
index a806529190..5a1d430308 100644
--- a/pkg/models/plan_test.go
+++ b/pkg/models/plan_test.go
@@ -5,9 +5,10 @@ package models_test
 import (
 	"testing"
 
+	"github.com/stretchr/testify/suite"
+
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
-	"github.com/stretchr/testify/suite"
 )
 
 type PlanTestSuite struct {
@@ -34,52 +35,65 @@ func (s *PlanTestSuite) TestNewPlan() {
 func (s *PlanTestSuite) TestMarkJobFailed() {
 	s.plan.MarkJobFailed(models.Event{Message: "Test failure"})
 	s.Equal(models.JobStateTypeFailed, s.plan.DesiredJobState)
-	s.Equal("Test failure", s.plan.Event.Message)
+	s.Equal("Test failure", s.plan.UpdateMessage)
+	s.Equal(1, len(s.plan.JobEvents))
+	s.Equal("Test failure", s.plan.JobEvents[0].Message)
 }
 
 func (s *PlanTestSuite) TestMarkJobRunningIfApplicable() {
 	// Test when DesiredJobState is already defined
 	s.plan = models.NewPlan(s.eval, s.job)
 	s.plan.DesiredJobState = models.JobStateTypeCompleted
-	s.plan.MarkJobRunningIfEligible()
+	s.plan.MarkJobRunningIfEligible(models.Event{Message: "job running"})
 	s.Equal(models.JobStateTypeCompleted, s.plan.DesiredJobState, "Should not change DesiredJobState if already defined")
+	s.Equal(0, len(s.plan.JobEvents), "Should not append JobEvents if DesiredJobState is already defined")
 
 	// Test when JobStateType is not Pending
 	s.job.State = models.NewJobState(models.JobStateTypeRunning)
 	s.plan = models.NewPlan(s.eval, s.job)
-	s.plan.MarkJobRunningIfEligible()
+	s.plan.MarkJobRunningIfEligible(models.Event{Message: "job running"})
 	s.Equal(models.JobStateTypeUndefined, s.plan.DesiredJobState, "Should remain Undefined if Job is not in Pending state")
+	s.Equal(0, len(s.plan.JobEvents), "Should not append JobEvents if Job is not in Pending state")
 
 	// Test when there are no running executions
 	s.job.State = models.NewJobState(models.JobStateTypePending)
 	s.plan = models.NewPlan(s.eval, s.job)
-	s.plan.MarkJobRunningIfEligible()
+	s.plan.MarkJobRunningIfEligible(models.Event{Message: "job running"})
 	s.Equal(models.JobStateTypeUndefined, s.plan.DesiredJobState, "Should remain Undefined if no running executions")
+	s.Equal(0, len(s.plan.JobEvents), "Should not append JobEvents if no running executions")
 
 	// Test when conditions are met to set DesiredJobState to Running
 	s.job.State = models.NewJobState(models.JobStateTypePending)
 	s.plan = models.NewPlan(s.eval, s.job)
-	s.plan.AppendExecution(&models.Execution{DesiredState: models.NewExecutionDesiredState(models.ExecutionDesiredStateRunning)})
-	s.plan.MarkJobRunningIfEligible()
+	s.plan.AppendExecution(&models.Execution{DesiredState: models.NewExecutionDesiredState(models.ExecutionDesiredStateRunning)}, models.Event{})
+	s.plan.MarkJobRunningIfEligible(models.Event{Message: "job running"})
 	s.Equal(models.JobStateTypeRunning, s.plan.DesiredJobState, "Should set to Running when all conditions are met")
-
+	s.Equal(1, len(s.plan.JobEvents), "Should append JobEvents when DesiredJobState is set to Running")
+	s.Equal("job running", s.plan.JobEvents[0].Message)
 	// Test when conditions are met to set DesiredJobState to Running
 	s.job.State = models.NewJobState(models.JobStateTypePending)
 	s.plan = models.NewPlan(s.eval, s.job)
-	s.plan.AppendApprovedExecution(mock.ExecutionForJob(s.plan.Job))
-	s.plan.MarkJobRunningIfEligible()
+	s.plan.AppendApprovedExecution(mock.ExecutionForJob(s.plan.Job), models.Event{Message: "job approved"})
+	s.plan.MarkJobRunningIfEligible(models.Event{Message: "job running"})
 	s.Equal(models.JobStateTypeRunning, s.plan.DesiredJobState, "Should set to Running when all conditions are met")
+	s.Equal(1, len(s.plan.JobEvents), "Should append JobEvents when DesiredJobState is set to Running")
+	s.Equal("job running", s.plan.JobEvents[0].Message)
 }
 
 func (s *PlanTestSuite) TestAppendExecution() {
-	s.plan.AppendExecution(mock.ExecutionForJob(s.job))
+	exec := mock.ExecutionForJob(s.job)
+	s.plan.AppendExecution(exec, models.Event{Message: "execution created"})
 	s.Equal(1, len(s.plan.NewExecutions), "NewExecutions should have length 1 after appending")
+	s.NotEmpty(s.plan.ExecutionEvents[exec.ID], "ExecutionEvents should have an entry for the new execution")
+	s.Equal("execution created", s.plan.ExecutionEvents[exec.ID][0].Message, "Should append event to ExecutionEvents")
 }
 
 func (s *PlanTestSuite) TestMarkJobCompleted() {
-	s.plan.MarkJobCompleted()
+	s.plan.MarkJobCompleted(models.Event{Message: "job completed"})
 	s.Equal(models.JobStateTypeCompleted, s.plan.DesiredJobState, "Should set DesiredJobState to Completed")
 	s.Equal(0, len(s.plan.NewExecutions), "NewExecutions should be empty after marking job as completed")
+	s.Equal(1, len(s.plan.JobEvents), "Should append JobEvents when DesiredJobState is set to Completed")
+	s.Equal("job completed", s.plan.JobEvents[0].Message)
 }
 
 func TestRunPlanTestSuite(t *testing.T) {
diff --git a/pkg/orchestrator/evaluation/watcher_test.go b/pkg/orchestrator/evaluation/watcher_test.go
index 653310b658..1c8c702638 100644
--- a/pkg/orchestrator/evaluation/watcher_test.go
+++ b/pkg/orchestrator/evaluation/watcher_test.go
@@ -116,7 +116,7 @@ func (s *WatcherTestSuite) createEvaluation() (*models.Evaluation, error) {
 	eval.Type = job.Type
 
 	// Create job
-	err := s.store.CreateJob(context.Background(), *job, models.Event{})
+	err := s.store.CreateJob(context.Background(), *job)
 	s.Require().NoError(err)
 
 	// Create an evaluation
diff --git a/pkg/orchestrator/planner/compute_forwarder_test.go b/pkg/orchestrator/planner/compute_forwarder_test.go
index 4fa21212fc..fb157ea131 100644
--- a/pkg/orchestrator/planner/compute_forwarder_test.go
+++ b/pkg/orchestrator/planner/compute_forwarder_test.go
@@ -8,12 +8,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/suite"
+	"go.uber.org/mock/gomock"
+
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
-	"github.com/stretchr/testify/suite"
-	"go.uber.org/mock/gomock"
 )
 
 type ComputeForwarderSuite struct {
diff --git a/pkg/orchestrator/planner/state_updater_test.go b/pkg/orchestrator/planner/state_updater_test.go
index 197a0b9b3f..305ee1247c 100644
--- a/pkg/orchestrator/planner/state_updater_test.go
+++ b/pkg/orchestrator/planner/state_updater_test.go
@@ -37,8 +37,8 @@ func (suite *StateUpdaterSuite) TestStateUpdater_Process_CreateExecutions_Succes
 	execution1, execution2 := mockCreateExecutions(plan)
 
 	suite.mockStore.EXPECT().BeginTx(suite.ctx).Return(suite.mockTxContext, nil).Times(1)
-	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1, models.Event{}).Times(1)
-	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution2, models.Event{}).Times(1)
+	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1).Times(1)
+	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution2).Times(1)
 	suite.mockTxContext.EXPECT().Commit()
 
 	suite.NoError(suite.stateUpdater.Process(suite.ctx, plan))
@@ -50,7 +50,7 @@ func (suite *StateUpdaterSuite) TestStateUpdater_Process_CreateExecutions_Error(
 
 	// no attempt to create execution2
 	suite.mockStore.EXPECT().BeginTx(suite.ctx).Return(suite.mockTxContext, nil).Times(1)
-	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1, models.Event{}).Return(errors.New("create error")).Times(1)
+	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1).Return(errors.New("create error")).Times(1)
 	suite.mockTxContext.EXPECT().Rollback()
 
 	suite.Error(suite.stateUpdater.Process(suite.ctx, plan))
@@ -124,8 +124,8 @@ func (suite *StateUpdaterSuite) TestStateUpdater_Process_MultiOp() {
 	plan.DesiredJobState = models.JobStateTypeCompleted
 
 	suite.mockStore.EXPECT().BeginTx(suite.ctx).Return(suite.mockTxContext, nil).Times(1)
-	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1, models.Event{}).Times(1)
-	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution2, models.Event{}).Times(1)
+	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution1).Times(1)
+	suite.mockStore.EXPECT().CreateExecution(suite.mockTxContext, *execution2).Times(1)
 	suite.mockStore.EXPECT().UpdateExecution(suite.mockTxContext, NewUpdateExecutionMatcherFromPlanUpdate(suite.T(), update1)).Times(1)
 	suite.mockStore.EXPECT().UpdateExecution(suite.mockTxContext, NewUpdateExecutionMatcherFromPlanUpdate(suite.T(), update2)).Times(1)
 	suite.mockStore.EXPECT().UpdateJobState(suite.mockTxContext, NewUpdateJobMatcherFromPlanUpdate(suite.T(), plan)).Times(1)
diff --git a/pkg/orchestrator/planner/utils_test.go b/pkg/orchestrator/planner/utils_test.go
index 4eca3774a5..461e68bd2a 100644
--- a/pkg/orchestrator/planner/utils_test.go
+++ b/pkg/orchestrator/planner/utils_test.go
@@ -61,7 +61,6 @@ type UpdateExecutionMatcher struct {
 	desiredStateComment string
 	expectedState       models.ExecutionStateType
 	expectedRevision    uint64
-	expectedEvent       models.Event
 }
 
 type UpdateExecutionMatcherParams struct {
@@ -71,7 +70,6 @@ type UpdateExecutionMatcherParams struct {
 	DesiredStateComment string
 	ExpectedState       models.ExecutionStateType
 	ExpectedRevision    uint64
-	ExpectedEvent       models.Event
 }
 
 func NewUpdateExecutionMatcher(t *testing.T, execution *models.Execution, params UpdateExecutionMatcherParams) *UpdateExecutionMatcher {
@@ -84,7 +82,6 @@ func NewUpdateExecutionMatcher(t *testing.T, execution *models.Execution, params
 		desiredStateComment: params.DesiredStateComment,
 		expectedState:       params.ExpectedState,
 		expectedRevision:    params.ExpectedRevision,
-		expectedEvent:       params.ExpectedEvent,
 	}
 }
 
@@ -93,7 +90,6 @@ func NewUpdateExecutionMatcherFromPlanUpdate(t *testing.T, update *models.PlanEx
 		NewDesiredState:     update.DesiredState,
 		DesiredStateComment: update.Event.Message,
 		ExpectedRevision:    update.Execution.Revision,
-		ExpectedEvent:       update.Event,
 	})
 }
 
@@ -113,7 +109,6 @@ func (m *UpdateExecutionMatcher) Matches(x interface{}) bool {
 		Condition: jobstore.UpdateExecutionCondition{
 			ExpectedRevision: m.expectedRevision,
 		},
-		Event: m.expectedEvent,
 	}
 
 	// set expected state if present
@@ -134,13 +129,13 @@ type UpdateJobMatcher struct {
 	t                *testing.T
 	job              *models.Job
 	newState         models.JobStateType
-	event            models.Event
+	comment          string
 	expectedRevision uint64
 }
 
 type UpdateJobMatcherParams struct {
 	NewState         models.JobStateType
-	Event            models.Event
+	Comment          string
 	ExpectedRevision uint64
 }
 
@@ -149,7 +144,7 @@ func NewUpdateJobMatcher(t *testing.T, job *models.Job, params UpdateJobMatcherP
 		t:                t,
 		job:              job,
 		newState:         params.NewState,
-		event:            params.Event,
+		comment:          params.Comment,
 		expectedRevision: params.ExpectedRevision,
 	}
 }
@@ -157,7 +152,7 @@ func NewUpdateJobMatcher(t *testing.T, job *models.Job, params UpdateJobMatcherP
 func NewUpdateJobMatcherFromPlanUpdate(t *testing.T, plan *models.Plan) *UpdateJobMatcher {
 	return NewUpdateJobMatcher(t, plan.Job, UpdateJobMatcherParams{
 		NewState:         plan.DesiredJobState,
-		Event:            plan.Event,
+		Comment:          plan.UpdateMessage,
 		ExpectedRevision: plan.Job.Revision,
 	})
 }
@@ -172,7 +167,7 @@ func (m *UpdateJobMatcher) Matches(x interface{}) bool {
 	expectedRequest := jobstore.UpdateJobStateRequest{
 		JobID:    m.job.ID,
 		NewState: m.newState,
-		Event:    m.event,
+		Message:  m.comment,
 		Condition: jobstore.UpdateJobCondition{
 			ExpectedRevision: m.expectedRevision,
 		},
diff --git a/pkg/test/logstream/stream_address_test.go b/pkg/test/logstream/stream_address_test.go
index c6a16cf1ac..ac396ee2f0 100644
--- a/pkg/test/logstream/stream_address_test.go
+++ b/pkg/test/logstream/stream_address_test.go
@@ -34,7 +34,7 @@ func (s *LogStreamTestSuite) TestStreamAddress() {
 	execution.NodeID = node.ID
 	execution.AllocateResources(task.Name, models.Resources{})
 
-	err = node.RequesterNode.JobStore.CreateJob(s.ctx, *job, models.Event{})
+	err = node.RequesterNode.JobStore.CreateJob(s.ctx, *job)
 	require.NoError(s.T(), err)
 
 	exec, err := node.ComputeNode.Executors.Get(s.ctx, models.EngineDocker)
@@ -74,7 +74,7 @@ func (s *LogStreamTestSuite) TestStreamAddress() {
 	node.ComputeNode.ExecutionStore.CreateExecution(s.ctx, *localExecutionState)
 
 	execution.ComputeState.StateType = models.ExecutionStateBidAccepted
-	err = node.RequesterNode.JobStore.CreateExecution(s.ctx, *execution, models.Event{})
+	err = node.RequesterNode.JobStore.CreateExecution(s.ctx, *execution)
 	require.NoError(s.T(), err)
 
 	ch, err := node.RequesterNode.Endpoint.ReadLogs(s.ctx, orchestrator.ReadLogsRequest{