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

Issue:
Piping job run outputs not working
Previously we were able to do operations like:
```
# Submit and describe a job
bacalhau docker run --id-only ubuntu echo hello | xargs bacalhau job describe

# Submit and copy job id
bacalhau docker run --id-only ubuntu echo hello | pbcopy
```

But that is no longer possible. Might be related to rendering stdout for job progress visibility



job progress doesn't render correctly when window is small
Bunch of duplicate rows. Even the headers are duplicated. 

### How to reproduce
1. The easier is the run multiple nodes
2. Run a job that fails so it gets retried and prints larger rows

```
→ bacalhau-1.5.0 docker run non_existent_image
17:53:16.564 | INF pkg/config/config.go:211 > Config loaded from: [], and with data-dir /Users/walid/bacalhau-versions/v1.5.0/data-dir
Job successfully submitted. Job ID: j-3c916967-6716-4853-8bc0-db988ec5df7c
Checking job status... (Enter Ctrl+C to exit at any time, your job will continue running):

 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:20.473  e-54ade96c  Requesting Node  Could not inspect image "non_existent_image" - could be due
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:20.473  e-54ade96c  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:20.473  e-54ade96c  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:16.567              Submission       Job submitted
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:20.473  e-54ade96c  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
 TIME                 EXEC. ID    TOPIC            EVENT
 Sep 19 17:53:18.701  e-8232fd5d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:20.473  e-54ade96c  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:22.299  e-b4f5069d  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:24.154  e-934aef42  Requesting Node  Could not inspect image "non_existent_image" - could be due
                                                   to repo/image not existing, or registry needing authorizatio
                                                   n: Error response from daemon: errors:
                                                   denied: requested access to the resource is denied
                                                   unauthorized: authentication required
                                                   * Hint: If the image is private, supply the node with valid
                                                   Docker login credentials using the DOCKER_USERNAME and DOCKE
                                                   R_PASSWORD environment variables
 Sep 19 17:53:24.183              Scheduling       not enough nodes to run job. requested: 1, available: 2, sui
                                                   table: 0.
                                                   • 2 of 2 nodes: job already executed on this node more than
                                                   once

Printout canceled due to timeout (the job is still running).

To get more information at any time, run:
   bacalhau job describe j-3c916967-6716-4853-8bc0-db988ec5df7c

To cancel the job, run:
   bacalhau job stop j-3c916967-6716-4853-8bc0-db988ec5df7c

Error: job failed

Job Results By Node:

• 4 runs on 2 nodes: Could not inspect image "non_existent_image" - could be due to repo/image not existing, or registry needing authorization: Error response from daemon: errors:
                       denied: requested access to the resource is denied
                       unauthorized: authentication required
```


Discussion / hints:
It appears that all output from the command is being written to stderr rather than stdout:


```
frrist@cypress ~/w/s/g/b/bacalhau (main)> bacalhau docker run --id-only ubuntu echo hello 2&> /dev/null
frrist@cypress ~/w/s/g/b/bacalhau (main)> bacalhau docker run --id-only ubuntu echo hello > /dev/null
j-4666e2c4-7926-4ad7-a133-3f1fd2752c70
```


Fix patch:
diff --git a/cmd/cli/config/list.go b/cmd/cli/config/list.go
index 6997fe11bf..c3d970b132 100644
--- a/cmd/cli/config/list.go
+++ b/cmd/cli/config/list.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/jedib0t/go-pretty/v6/text"
+	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v3"
 
@@ -40,6 +41,7 @@ Each key shown can be used with:
 			if err != nil {
 				return err
 			}
+			log.Info().Msgf("Config loaded from: %s, and with data-dir %s", cfg.Paths(), cfg.Get(types.DataDirKey))
 			return list(cmd, cfg, o)
 		},
 	}
diff --git a/cmd/cli/job/describe.go b/cmd/cli/job/describe.go
index bc2a3450b8..16f38a7655 100644
--- a/cmd/cli/job/describe.go
+++ b/cmd/cli/job/describe.go
@@ -6,12 +6,11 @@ import (
 	"slices"
 	"time"
 
-	"github.com/jedib0t/go-pretty/v6/table"
-	"github.com/jedib0t/go-pretty/v6/text"
 	"github.com/samber/lo"
 	"github.com/spf13/cobra"
 	"k8s.io/kubectl/pkg/util/i18n"
 
+	"github.com/bacalhau-project/bacalhau/cmd/util/cols"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
@@ -211,19 +210,14 @@ func (o *DescribeOptions) printHistory(cmd *cobra.Command, label string, history
 		return nil
 	}
 
-	timeCol := output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: historyTimeCol.ColumnConfig.Name, WidthMax: 20, WidthMaxEnforcer: text.WrapText},
-		Value:        func(h *models.JobHistory) string { return h.Occurred().Format(time.DateTime) },
-	}
-
 	tableOptions := output.OutputOptions{
 		Format:  output.TableFormat,
 		NoStyle: true,
 	}
 	jobHistoryCols := []output.TableColumn[*models.JobHistory]{
-		timeCol,
-		historyTopicCol,
-		historyEventCol,
+		cols.HistoryDateTime,
+		cols.HistoryTopic,
+		cols.HistoryEvent,
 	}
 	output.Bold(cmd, fmt.Sprintf("\n%s History\n", label))
 	return output.Output(cmd, jobHistoryCols, tableOptions, history)
diff --git a/cmd/cli/job/executions.go b/cmd/cli/job/executions.go
index 6fb217e2ff..428262a8eb 100644
--- a/cmd/cli/job/executions.go
+++ b/cmd/cli/job/executions.go
@@ -125,8 +125,9 @@ var (
 		Value:        func(e *models.Execution) string { return e.DesiredState.StateType.String() },
 	}
 	executionColumnComment = output.TableColumn[*models.Execution]{
-		ColumnConfig: table.ColumnConfig{Name: "Comment", WidthMax: 40, WidthMaxEnforcer: text.WrapText},
-		Value:        func(e *models.Execution) string { return e.ComputeState.Message },
+		ColumnConfig: table.ColumnConfig{
+			Name: "Comment", WidthMax: 40, WidthMaxEnforcer: output.WrapSoftPreserveNewlines},
+		Value: func(e *models.Execution) string { return e.ComputeState.Message },
 	}
 )
 
diff --git a/cmd/cli/job/history.go b/cmd/cli/job/history.go
index 8edb7ed0a1..f78abf1242 100644
--- a/cmd/cli/job/history.go
+++ b/cmd/cli/job/history.go
@@ -3,21 +3,17 @@ package job
 import (
 	"errors"
 	"fmt"
-	"time"
 
-	"github.com/jedib0t/go-pretty/v6/table"
-	"github.com/jedib0t/go-pretty/v6/text"
-	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
 	"k8s.io/kubectl/pkg/util/i18n"
 
 	"github.com/bacalhau-project/bacalhau/cmd/util"
+	"github.com/bacalhau-project/bacalhau/cmd/util/cols"
 	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
 	"github.com/bacalhau-project/bacalhau/cmd/util/output"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 	"github.com/bacalhau-project/bacalhau/pkg/util/templates"
 )
 
@@ -90,62 +86,12 @@ func NewHistoryCmd() *cobra.Command {
 	return nodeCmd
 }
 
-var (
-	historyTimeCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "Time", WidthMax: len(time.StampMilli), WidthMaxEnforcer: output.ShortenTime},
-		Value:        func(j *models.JobHistory) string { return j.Occurred().Format(time.StampMilli) },
-	}
-	historyLevelCol = output.TableColumn[*models.JobHistory]{
-		ColumnConfig: table.ColumnConfig{Name: "Level", WidthMax: 15, WidthMaxEnforcer: text.WrapText},
-		Value:        func(jwi *models.JobHistory) string { return jwi.Type.String() },
-	}
-	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) },
-	}
-	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: 90, WidthMaxEnforcer: text.WrapText},
-		Value: func(h *models.JobHistory) string {
-			res := h.Event.Message
-
-			if h.Event.Details != nil {
-				// if is error, then the event is in red
-				if h.Event.Details[models.DetailsKeyIsError] == "true" {
-					res = output.RedStr(res)
-				}
-
-				// print hint in green
-				if h.Event.Details[models.DetailsKeyHint] != "" {
-					res += "\n" + fmt.Sprintf(
-						"%s %s", output.BoldStr(output.GreenStr("* Hint:")), h.Event.Details[models.DetailsKeyHint])
-				}
-
-				// print all other details in debug mode
-				if zerolog.GlobalLevel() <= zerolog.DebugLevel {
-					for k, v := range h.Event.Details {
-						// don't print hint and error since they are already represented
-						if k == models.DetailsKeyHint || k == models.DetailsKeyIsError {
-							continue
-						}
-						res += "\n" + fmt.Sprintf("* %s %s", output.BoldStr(k+":"), v)
-					}
-				}
-			}
-			return res
-		},
-	}
-)
-
 var historyColumns = []output.TableColumn[*models.JobHistory]{
-	historyTimeCol,
-	historyLevelCol,
-	historyExecIDCol,
-	historyTopicCol,
-	historyEventCol,
+	cols.HistoryTime,
+	cols.HistoryLevel,
+	cols.HistoryExecID,
+	cols.HistoryTopic,
+	cols.HistoryEvent,
 }
 
 func (o *HistoryOptions) run(cmd *cobra.Command, args []string, api client.API) error {
diff --git a/cmd/cli/root.go b/cmd/cli/root.go
index df88fdc761..0786dd8763 100644
--- a/cmd/cli/root.go
+++ b/cmd/cli/root.go
@@ -130,6 +130,11 @@ func Execute(ctx context.Context) {
 	rootCmd.SetContext(ctx)
 	rootCmd.SilenceErrors = true
 	rootCmd.SilenceUsage = true
+
+	// this is needed as cobra defaults to Stderr if no output is set.
+	rootCmd.SetOut(os.Stdout)
+	rootCmd.SetErr(os.Stderr)
+
 	if err := rootCmd.Execute(); err != nil {
 		util.Fatal(rootCmd, err, 1)
 	}
diff --git a/cmd/cli/serve/serve.go b/cmd/cli/serve/serve.go
index 3ef163f338..19e226fba0 100644
--- a/cmd/cli/serve/serve.go
+++ b/cmd/cli/serve/serve.go
@@ -92,10 +92,14 @@ func NewCmd() *cobra.Command {
 			return configflags.BindFlags(viper.GetViper(), serveFlags)
 		},
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			cfg, err := util.SetupConfig(cmd)
+			cfg, rawCfg, err := util.SetupConfigs(cmd)
 			if err != nil {
 				return fmt.Errorf("failed to setup config: %w", err)
 			}
+
+			log.Info().Msgf("Config loaded from: %s, and with data-dir %s",
+				rawCfg.Paths(), rawCfg.Get(types.DataDirKey))
+
 			// create or open the bacalhau repo and load the config
 			fsr, err := setup.SetupBacalhauRepo(cfg)
 			if err != nil {
diff --git a/cmd/util/cols/formats.go b/cmd/util/cols/formats.go
new file mode 100644
index 0000000000..73fbd6d7f4
--- /dev/null
+++ b/cmd/util/cols/formats.go
@@ -0,0 +1,5 @@
+package cols
+
+const (
+	TimeOnlyMilli = "15:04:05.000"
+)
diff --git a/cmd/util/cols/history.go b/cmd/util/cols/history.go
new file mode 100644
index 0000000000..39ec335399
--- /dev/null
+++ b/cmd/util/cols/history.go
@@ -0,0 +1,76 @@
+package cols
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/jedib0t/go-pretty/v6/table"
+	"github.com/jedib0t/go-pretty/v6/text"
+	"github.com/rs/zerolog"
+
+	"github.com/bacalhau-project/bacalhau/cmd/util/output"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
+)
+
+var (
+	HistoryTime = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Time", WidthMax: len(time.StampMilli), WidthMaxEnforcer: output.ShortenTime},
+		Value:        func(j *models.JobHistory) string { return j.Occurred().Format(time.StampMilli) },
+	}
+	HistoryTimeOnly = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Time", WidthMax: len(TimeOnlyMilli), WidthMaxEnforcer: text.Trim},
+		Value:        func(j *models.JobHistory) string { return j.Occurred().Format(TimeOnlyMilli) },
+	}
+	HistoryDateTime = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Time", WidthMax: 20, WidthMaxEnforcer: text.WrapText},
+		Value:        func(j *models.JobHistory) string { return j.Occurred().Format(time.DateTime) },
+	}
+	HistoryLevel = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Level", WidthMax: 15, WidthMaxEnforcer: text.WrapText},
+		Value:        func(jwi *models.JobHistory) string { return jwi.Type.String() },
+	}
+	HistoryExecID = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Exec. ID", WidthMax: 10, WidthMaxEnforcer: text.WrapText},
+		Value: func(j *models.JobHistory) string {
+			if j.ExecutionID == "" {
+				return "/"
+			}
+			return idgen.ShortUUID(j.ExecutionID)
+		},
+	}
+	HistoryTopic = 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) },
+	}
+	HistoryEvent = output.TableColumn[*models.JobHistory]{
+		ColumnConfig: table.ColumnConfig{Name: "Event", WidthMax: 90, WidthMaxEnforcer: output.WrapSoftPreserveNewlines},
+		Value: func(h *models.JobHistory) string {
+			res := h.Event.Message
+
+			if h.Event.Details != nil {
+				if h.Event.Details[models.DetailsKeyIsError] == "true" {
+					res = output.BoldStr(output.RedStr("Error: ")) + res
+				}
+
+				// print hint in green
+				if h.Event.Details[models.DetailsKeyHint] != "" {
+					res +=
+						"\n" + output.BoldStr(output.GreenStr("Hint: ")) + h.Event.Details[models.DetailsKeyHint]
+				}
+
+				// print all other details in debug mode
+				if zerolog.GlobalLevel() <= zerolog.DebugLevel {
+					for k, v := range h.Event.Details {
+						// don't print hint and error since they are already represented
+						if k == models.DetailsKeyHint || k == models.DetailsKeyIsError {
+							continue
+						}
+						res += "\n" + fmt.Sprintf("* %s %s", output.BoldStr(k+":"), v)
+					}
+				}
+			}
+			return res
+		},
+	}
+)
diff --git a/cmd/util/flags/cliflags/runtime.go b/cmd/util/flags/cliflags/runtime.go
index b9bc47a91d..08eceec71e 100644
--- a/cmd/util/flags/cliflags/runtime.go
+++ b/cmd/util/flags/cliflags/runtime.go
@@ -19,6 +19,7 @@ type RunTimeSettings struct {
 	PrintNodeDetails      bool
 	Follow                bool // Follow along with the output of the job
 	DryRun                bool // iff true do not submit the job, but instead print out what will be submitted.
+	GroupEvents           bool // Group events by execution
 }
 
 const DefaultRunWaitSeconds = 600
@@ -38,6 +39,8 @@ func NewRunTimeSettingsFlags(settings *RunTimeSettings) *pflag.FlagSet {
 	flags.BoolVar(
 		&settings.DryRun, "dry-run", settings.DryRun,
 		`Do not submit the job, but instead print out what will be submitted`)
+	flags.BoolVar(&settings.GroupEvents, "group-events", false,
+		"Group events by execution")
 
 	return flags
 }
diff --git a/cmd/util/live_table_writer.go b/cmd/util/live_table_writer.go
index 573c350fdc..bb20d29913 100644
--- a/cmd/util/live_table_writer.go
+++ b/cmd/util/live_table_writer.go
@@ -34,15 +34,16 @@ func NewLiveTableWriter() *LiveTableWriter {
 	}
 }
 
-// flush writes out the table and resets the buffer. It should be called after
+// Flush writes out the table and resets the buffer. It should be called after
 // the last call to Write to ensure that any data buffered in the Writer is
 // written to the output. An error is returned if the contents of the buffer
 // cannot be written to the underlying output stream.
-func (w *LiveTableWriter) flush() error {
+func (w *LiveTableWriter) Flush() error {
 	if len(w.buf.Bytes()) == 0 {
 		return nil
 	}
 
+	defer w.buf.Reset()
 	w.clearRows()
 
 	rows := 0
@@ -53,7 +54,11 @@ func (w *LiveTableWriter) flush() error {
 	}
 	w.rowCount = rows
 	_, err := w.Out.Write(w.buf.Bytes())
-	w.buf.Reset()
+	if err == nil {
+		// Add a new line at the end
+		_, err = w.Out.Write([]byte("\n"))
+		w.rowCount++
+	}
 	return err
 }
 
@@ -62,7 +67,7 @@ func (w *LiveTableWriter) flush() error {
 func (w *LiveTableWriter) Write(buf []byte) (n int, err error) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	_ = w.flush()
+	_ = w.Flush()
 	return w.buf.Write(buf)
 }
 
diff --git a/cmd/util/output/output.go b/cmd/util/output/output.go
index 31fabc9312..4201d9846a 100644
--- a/cmd/util/output/output.go
+++ b/cmd/util/output/output.go
@@ -4,11 +4,12 @@ import (
 	"encoding/json"
 	"fmt"
 
-	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
 	"github.com/ghodss/yaml"
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/samber/lo"
 	"github.com/spf13/cobra"
+
+	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
 )
 
 type OutputFormat string
@@ -21,10 +22,16 @@ const (
 )
 
 const (
-	bold  = "\033[1m"
-	red   = "\033[31m"
-	green = "\033[32m"
-	reset = "\033[0m"
+	// Escape sequence start and end
+	escapeStart = "\x1b["
+	escapeEnd   = "m"
+
+	// Styling
+	bold   = "\033[1m"
+	italic = "\033[3m"
+	red    = "\033[31m"
+	green  = "\033[32m"
+	reset  = "\033[0m"
 )
 
 var AllFormats = append([]OutputFormat{TableFormat, CSVFormat}, NonTabularFormats...)
@@ -154,6 +161,16 @@ func BoldStr(s any) string {
 	return bold + fmt.Sprint(s) + reset
 }
 
+// Italic prints the given string in italic
+func Italic(cmd *cobra.Command, s string) {
+	cmd.Print(italic + s + reset)
+}
+
+// ItalicStr returns the given string in italic
+func ItalicStr(s any) string {
+	return italic + fmt.Sprint(s) + reset
+}
+
 // RedStr returns the given string in red
 func RedStr(s any) string {
 	return red + fmt.Sprint(s) + reset
diff --git a/cmd/util/output/utils.go b/cmd/util/output/utils.go
index eb84205150..96348850e3 100644
--- a/cmd/util/output/utils.go
+++ b/cmd/util/output/utils.go
@@ -2,7 +2,9 @@ package output
 
 import (
 	"fmt"
+	"strings"
 	"time"
+	"unicode"
 )
 
 const (
@@ -60,3 +62,57 @@ func Elapsed(t time.Time) string {
 
 	return result + " ago"
 }
+
+func WrapSoftPreserveNewlines(s string, width int) string {
+	lines := strings.Split(s, "\n")
+	var result strings.Builder
+
+	for i, line := range lines {
+		if i > 0 {
+			result.WriteString("\n")
+		}
+		result.WriteString(wrapLine(line, width))
+	}
+
+	return result.String()
+}
+
+func wrapLine(line string, width int) string {
+	var wrapped strings.Builder
+	var current strings.Builder
+	var styling strings.Builder
+	inEscape := false
+	lineLength := 0
+	escapeStartRune := []rune(escapeStart)[0]
+	escapeEndRune := []rune(escapeEnd)[0]
+
+	for _, r := range line {
+		if r == escapeStartRune {
+			inEscape = true
+			styling.WriteRune(r)
+		} else if inEscape {
+			styling.WriteRune(r)
+			if r == escapeEndRune {
+				inEscape = false
+				current.WriteString(styling.String())
+				if styling.String() == escapeStart+"0"+escapeEnd {
+					styling.Reset()
+				}
+			}
+		} else {
+			current.WriteRune(r)
+			lineLength++
+
+			if lineLength >= width && unicode.IsSpace(r) {
+				wrapped.WriteString(current.String())
+				wrapped.WriteString("\n")
+				current.Reset()
+				current.WriteString(styling.String())
+				lineLength = 0
+			}
+		}
+	}
+
+	wrapped.WriteString(current.String())
+	return wrapped.String()
+}
diff --git a/cmd/util/printer/event_printer.go b/cmd/util/printer/event_printer.go
new file mode 100644
index 0000000000..8cc3f72fc1
--- /dev/null
+++ b/cmd/util/printer/event_printer.go
@@ -0,0 +1,220 @@
+// Package printer provides functionality for printing job events and progress.
+// It offers different types of printers to handle various output formats and requirements.
+package printer
+
+import (
+	"io"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/bacalhau-project/bacalhau/cmd/util"
+	"github.com/bacalhau-project/bacalhau/cmd/util/cols"
+	"github.com/bacalhau-project/bacalhau/cmd/util/output"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+)
+
+// eventPrinter defines the interface for different types of event printers.
+type eventPrinter interface {
+	// printEvent prints a single job history event.
+	printEvent(history *models.JobHistory) error
+	// close performs any necessary cleanup operations.
+	close() error
+}
+
+// quitePrinter is an eventPrinter that discards all output.
+type quitePrinter struct {
+}
+
+// newQuitePrinter creates a new quitePrinter and sets the command output to discard.
+func newQuitePrinter(cmd *cobra.Command) *quitePrinter {
+	return &quitePrinter{}
+}
+
+// printEvent implements eventPrinter interface but does nothing for quitePrinter.
+func (quitePrinter) printEvent(*models.JobHistory) error { return nil }
+
+// close implements eventPrinter interface but does nothing for quitePrinter.
+func (quitePrinter) close() error { return nil }
+
+// sequentialEventPrinter prints job events sequentially as they occur.
+type sequentialEventPrinter struct {
+	cmd                 *cobra.Command
+	columns             []output.TableColumn[*models.JobHistory]
+	lineCount           int
+	spinner             *FishSpinner
+	seenExecutionErrors bool
+}
+
+// newSequentialEventPrinter creates a new sequentialEventPrinter.
+func newSequentialEventPrinter(cmd *cobra.Command) *sequentialEventPrinter {
+	spinner := NewFishSpinner(cmd.OutOrStdout())
+	spinner.Start()
+
+	eventsCols := []output.TableColumn[*models.JobHistory]{
+		cols.HistoryTimeOnly,
+		cols.HistoryExecID,
+		cols.HistoryTopic,
+		cols.HistoryEvent,
+	}
+
+	// since we print each row as a separate table, we need to set the
+	// min width of the columns to avoid inconsistent column widths.
+	// will use the same as max width for now, except for the last column.
+	for i := 0; i < len(eventsCols)-1; i++ {
+		eventsCols[i].WidthMin = eventsCols[i].WidthMax
+	}
+
+	return &sequentialEventPrinter{
+		cmd:     cmd,
+		columns: eventsCols,
+		spinner: spinner,
+	}
+}
+
+// filterEvent determines whether an event should be printed.
+// It filters out events that can be too noisy.
+func (p *sequentialEventPrinter) filterEvent(event *models.JobHistory) bool {
+	// Always print the first event indicating job submission
+	if p.lineCount == 0 {
+		return true
+	}
+
+	// Check for execution level errors
+	if event.IsExecutionLevel() && event.Event.HasError() {
+		p.seenExecutionErrors = true
+		return true
+	}
+
+	// Print job level errors only if we haven't seen execution level errors
+	if !event.IsExecutionLevel() && event.Event.HasError() && !p.seenExecutionErrors {
+		return true
+	}
+
+	// Print all non-error execution level events
+	if event.IsExecutionLevel() {
+		return true
+	}
+
+	return false
+}
+
+// printEvent prints a single job history event if it passes the filter.
+func (p *sequentialEventPrinter) printEvent(event *models.JobHistory) error {
+	if !p.filterEvent(event) {
+		return nil
+	}
+
+	options := output.OutputOptions{Format: output.TableFormat, NoStyle: true, HideHeader: true}
+	if p.lineCount == 0 {
+		options.HideHeader = false
+	}
+
+	p.spinner.Pause() // Pause the spinner while printing the event
+	p.spinner.Clear() // Clear the spinner
+
+	err := output.Output(p.cmd, p.columns, options, []*models.JobHistory{event})
+	if err != nil {
+		return err
+	}
+
+	p.spinner.Resume() // Resume the spinner
+	p.lineCount++
+
+	return nil
+}
+
+// close clears the loading message when closing the printer.
+func (p *sequentialEventPrinter) close() error {
+	p.spinner.Stop()
+	p.spinner.Clear()
+	return nil
+}
+
+// groupedEventPrinter is an experimental printer that groups events by execution.
+// UNSTABLE: This printer is still in development and may change in future versions.
+type groupedEventPrinter struct {
+	cmd         *cobra.Command
+	columns     []output.TableColumn[*models.JobHistory]
+	existingOut io.Writer
+	jobEvent    *models.JobHistory
+	executions  []*executionGroup
+}
+
+type executionGroup struct {
+	ExecutionID   string
+	LatestEvent   *models.JobHistory
+	DiscoveryTime time.Time
+}
+
+// newGroupedEventPrinter creates a new groupedEventPrinter.
+func newGroupedEventPrinter(cmd *cobra.Command) *groupedEventPrinter {
+	existingOut := cmd.OutOrStdout()
+	cmd.SetOut(util.NewLiveTableWriter())
+
+	eventsCols := []output.TableColumn[*models.JobHistory]{
+		cols.HistoryTimeOnly,
+		cols.HistoryExecID,
+		cols.HistoryTopic,
+		cols.HistoryEvent,
+	}
+
+	return &groupedEventPrinter{
+		cmd:         cmd,
+		existingOut: existingOut,
+		columns:     eventsCols,
+		executions:  make([]*executionGroup, 0),
+	}
+}
+
+// printEvent adds or updates an event and renders the table.
+func (p *groupedEventPrinter) printEvent(event *models.JobHistory) error {
+	if event.ExecutionID == "" {
+		p.jobEvent = event
+	} else {
+		p.updateOrAddExecution(event)
+	}
+	return p.renderTable()
+}
+
+// updateOrAddExecution updates an existing execution or adds a new one.
+func (p *groupedEventPrinter) updateOrAddExecution(event *models.JobHistory) {
+	for _, group := range p.executions {
+		if group.ExecutionID == event.ExecutionID {
+			group.LatestEvent = event
+			return
+		}
+	}
+
+	// If execution not found, add a new one
+	newGroup := &executionGroup{
+		ExecutionID:   event.ExecutionID,
+		LatestEvent:   event,
+		DiscoveryTime: time.Now(),
+	}
+	p.executions = append(p.executions, newGroup)
+}
+
+// renderTable outputs all events as a table with stable execution order.
+func (p *groupedEventPrinter) renderTable() error {
+	var entries []*models.JobHistory
+
+	// Add execution events in the order they were discovered
+	for _, group := range p.executions {
+		entries = append(entries, group.LatestEvent)
+	}
+
+	// Add job event last
+	if p.jobEvent != nil {
+		entries = append(entries, p.jobEvent)
+	}
+
+	options := output.OutputOptions{Format: output.TableFormat, NoStyle: true}
+	return output.Output(p.cmd, p.columns, options, entries)
+}
+
+// close resets the command's output and flushes the live writer.
+func (p *groupedEventPrinter) close() error {
+	p.cmd.SetOut(p.existingOut)
+	return nil
+}
diff --git a/cmd/util/printer/fish_spinner.go b/cmd/util/printer/fish_spinner.go
new file mode 100644
index 0000000000..c5de731e27
--- /dev/null
+++ b/cmd/util/printer/fish_spinner.go
@@ -0,0 +1,150 @@
+package printer
+
+import (
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/bacalhau-project/bacalhau/cmd/util/output"
+)
+
+// ANSI escape codes for cursor control
+const (
+	hideCursor = "\033[?25l"
+	showCursor = "\033[?25h"
+)
+
+// FishSpinner represents a simple fish emoji spinner.
+// It provides a visual indicator of ongoing processing while allowing
+// concurrent event printing.
+//
+// The FishSpinner operates in its own goroutine, continuously updating
+// and printing the spinner animation. However, it's designed to work
+// in conjunction with an event printer, which may need to write to the
+// console periodically.
+//
+// To prevent the spinner and event printer from writing to the console
+// simultaneously and causing overlapping output, the FishSpinner implements
+// a synchronization mechanism:
+//
+//  1. The event printer should call Pause() before printing a new event row.
+//     This pauses the spinner animation
+//  2. The event printer then calls clear() to remove the spinner from the console
+//  2. The event printer then prints its event row.
+//  3. After printing, the event printer should call Resume() to restart
+//     the spinner animation from where it left before pausing.
+//
+// This synchronization ensures that the spinner and event printer's outputs
+// don't interfere with each other, maintaining clean and readable console output.
+type FishSpinner struct {
+	frames      []string
+	index       int
+	position    int
+	width       int
+	stopChan    chan struct{}
+	writer      io.Writer
+	spinnerText string
+	mu          sync.Mutex
+	paused      bool
+}
+
+// NewFishSpinner creates a new FishSpinner
+func NewFishSpinner(writer io.Writer) *FishSpinner {
+	return &FishSpinner{
+		frames:      []string{"🐟", "🐠", "🐡"},
+		index:       0,
+		position:    0,
+		width:       21, // Width of the animation area
+		stopChan:    make(chan struct{}),
+		writer:      writer,
+		spinnerText: output.ItalicStr(" Processing   "),
+	}
+}
+
+// Start begins the spinner animation
+func (s *FishSpinner) Start() {
+	s.hideCursor()
+	go s.run()
+}
+
+// Stop stops the spinner animation
+func (s *FishSpinner) Stop() {
+	close(s.stopChan)
+	s.showCursor()
+}
+
+// Pause temporarily stops the spinner animation and clears its output
+// This function is called before the event printer wants to print a new row
+func (s *FishSpinner) Pause() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.paused = true
+}
+
+// Resume restarts the spinner animation after it has been paused
+// This function is called after the event printer has finished printing
+func (s *FishSpinner) Resume() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.paused {
+		s.paused = false
+		s.print() // Print immediately when resuming
+	}
+}
+
+// run continuously updates the spinner animation
+func (s *FishSpinner) run() {
+	ticker := time.NewTicker(200 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			s.mu.Lock()
+			if !s.paused {
+				s.update()
+				s.print()
+			}
+			s.mu.Unlock()
+		case <-s.stopChan:
+			return
+		}
+	}
+}
+
+// update advances the spinner's state
+func (s *FishSpinner) update() {
+	s.position++
+	if s.position >= s.width {
+		s.position = 0
+		s.index = (s.index + 1) % len(s.frames)
+	}
+}
+
+// print displays the current frame of the spinner
+func (s *FishSpinner) print() {
+	frame := s.frames[s.index]
+	dots := strings.Repeat(".", s.width)
+	animation := dots[:s.position] + frame + dots[s.position+1:]
+
+	fmt.Fprintf(s.writer, "\r%s %s", s.spinnerText, animation)
+}
+
+// Clear removes the spinner from the console
+func (s *FishSpinner) Clear() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	fmt.Fprint(s.writer, "\r"+strings.Repeat(" ", len(s.spinnerText)+s.width+2)+"\r")
+}
+
+// hideCursor hides the cursor
+func (s *FishSpinner) hideCursor() {
+	fmt.Fprint(s.writer, hideCursor)
+}
+
+// showCursor shows the cursor
+func (s *FishSpinner) showCursor() {
+	fmt.Fprint(s.writer, showCursor)
+}
diff --git a/cmd/util/printer/job_progress_printer.go b/cmd/util/printer/job_progress_printer.go
deleted file mode 100644
index 765444e1a2..0000000000
--- a/cmd/util/printer/job_progress_printer.go
+++ /dev/null
@@ -1,427 +0,0 @@
-package printer
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"os/signal"
-	"sort"
-	"strings"
-	"time"
-
-	"github.com/jedib0t/go-pretty/v6/table"
-	"github.com/jedib0t/go-pretty/v6/text"
-	"github.com/pkg/errors"
-	"github.com/rs/zerolog"
-	"github.com/rs/zerolog/log"
-	"github.com/samber/lo"
-	"github.com/spf13/cobra"
-
-	"github.com/bacalhau-project/bacalhau/cmd/util"
-	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
-	"github.com/bacalhau-project/bacalhau/cmd/util/output"
-	"github.com/bacalhau-project/bacalhau/pkg/bacerrors"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
-	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
-)
-
-const PrintoutCanceledButRunningNormally string = "printout canceled but running normally"
-const PrintoutTimeoutButRunningNormally string = "but running normally"
-
-type JobProgressPrinter struct {
-	client          clientv2.API
-	runtimeSettings *cliflags.RunTimeSettings
-}
-
-type jobProgressEvent struct {
-	jobID       string
-	occurred    time.Time
-	executionID string
-	event       models.Event
-}
-
-var (
-	jobProgressEventTimeCol = output.TableColumn[*jobProgressEvent]{
-		ColumnConfig: table.ColumnConfig{Name: "Time", WidthMax: len(time.StampMilli), WidthMaxEnforcer: output.ShortenTime},
-		Value:        func(j *jobProgressEvent) string { return j.occurred.Format(time.StampMilli) },
-	}
-
-	jobProgressEventExecIDCol = output.TableColumn[*jobProgressEvent]{
-		ColumnConfig: table.ColumnConfig{Name: "Exec. ID", WidthMax: 10, WidthMaxEnforcer: text.WrapText},
-		Value:        func(j *jobProgressEvent) string { return idgen.ShortUUID(j.executionID) },
-	}
-
-	jobProgressEventTopicCol = output.TableColumn[*jobProgressEvent]{
-		ColumnConfig: table.ColumnConfig{Name: "Topic", WidthMax: 20, WidthMaxEnforcer: text.WrapSoft},
-		Value: func(j *jobProgressEvent) string {
-			return string(j.event.Topic)
-		},
-	}
-
-	jobProgressEventEventCol = output.TableColumn[*jobProgressEvent]{
-		ColumnConfig: table.ColumnConfig{Name: "Event", WidthMax: 60, WidthMaxEnforcer: text.WrapText},
-		Value: func(j *jobProgressEvent) string {
-			res := j.event.Message
-
-			if j.event.Details != nil {
-				// if is error, then the event is in red
-				if j.event.Details[models.DetailsKeyIsError] == "true" {
-					res = output.RedStr(res)
-				}
-
-				// print hint in green
-				if j.event.Details[models.DetailsKeyHint] != "" {
-					res += "\n" + fmt.Sprintf(
-						"%s %s", output.BoldStr(output.GreenStr("* Hint:")), j.event.Details[models.DetailsKeyHint])
-				}
-
-				// print all other details in debug mode
-				if zerolog.GlobalLevel() <= zerolog.DebugLevel {
-					for k, v := range j.event.Details {
-						// don't print hint and error since they are already represented
-						if k == models.DetailsKeyHint || k == models.DetailsKeyIsError {
-							continue
-						}
-						res += "\n" + fmt.Sprintf("* %s %s", output.BoldStr(k+":"), v)
-					}
-				}
-			}
-			return res
-		},
-	}
-)
-
-var jobProgressEventCols = []output.TableColumn[*jobProgressEvent]{
-	jobProgressEventTimeCol,
-	jobProgressEventExecIDCol,
-	jobProgressEventTopicCol,
-	jobProgressEventEventCol,
-}
-
-func NewJobProgressPrinter(client clientv2.API, runtimeSettings *cliflags.RunTimeSettings) *JobProgressPrinter {
-	return &JobProgressPrinter{
-		client:          client,
-		runtimeSettings: runtimeSettings,
-	}
-}
-
-// PrintJobProgress displays the job progress depending upon on cli runtime
-// settings
-func (j *JobProgressPrinter) PrintJobProgress(ctx context.Context, job *models.Job, cmd *cobra.Command) error {
-	jobID := job.ID
-	// If we are in `--wait=false` print the jobID and then exit.
-	// All the code after this point is to show the progress of the job.
-	if !j.runtimeSettings.WaitForJobToFinish {
-		cmd.Print(jobID + "\n")
-		return nil
-	}
-
-	// If we are n `id-only` mode - print the id
-	if j.runtimeSettings.PrintJobIDOnly {
-		cmd.Print(jobID + "\n")
-	}
-
-	// Follow Logs
-	if j.runtimeSettings.Follow {
-		return j.followLogs(jobID, cmd)
-	}
-
-	// If we are only printing the id, set the rest of the output to "quiet",
-	// i.e. don't print
-	quiet := j.runtimeSettings.PrintJobIDOnly
-
-	jobErr := j.followProgress(ctx, jobID, cmd, quiet)
-	if jobErr != nil {
-		if jobErr.Error() == PrintoutCanceledButRunningNormally {
-			return nil
-		}
-
-		history, err := j.client.Jobs().History(ctx, &apimodels.ListJobHistoryRequest{
-			JobID:     jobID,
-			EventType: "execution",
-		})
-		if err != nil {
-			return fmt.Errorf("failed getting job history: %w", err)
-		}
-
-		historySummary := SummariseHistoryEvents(history.Items)
-		if len(historySummary) > 0 {
-			for _, event := range historySummary {
-				PrintEvent(cmd, event)
-			}
-		} else {
-			PrintError(cmd, jobErr)
-		}
-	}
-
-	if j.runtimeSettings.PrintNodeDetails || jobErr != nil {
-		executions, err := j.client.Jobs().Executions(ctx, &apimodels.ListJobExecutionsRequest{
-			JobID: jobID,
-		})
-		if err != nil {
-			return fmt.Errorf("failed getting job executions: %w", err)
-		}
-		summary := SummariseExecutions(executions.Items)
-		if len(summary) > 0 {
-			cmd.Println("\nJob Results By Node:")
-			for message, runs := range summary {
-				nodes := len(lo.Uniq(runs))
-				prefix := fmt.Sprintf("• Node %s: ", runs[0])
-				if len(runs) > 1 {
-					prefix = fmt.Sprintf("• %d runs on %d nodes: ", len(runs), nodes)
-				}
-				printIndentedString(cmd, prefix, strings.Trim(message, "\n"), none, 0)
-			}
-		} else {
-			cmd.Println()
-		}
-	}
-
-	if !quiet {
-		cmd.Println()
-		cmd.Println("To get more details about the run, execute:")
-		cmd.Println("\t" + os.Args[0] + " job describe " + jobID)
-
-		cmd.Println()
-		cmd.Println("To get more details about the run executions, execute:")
-		cmd.Println("\t" + os.Args[0] + " job executions " + jobID)
-
-		// only print help for downloading the job if it contained a publisher.
-		if !lo.IsEmpty(job.Task().Publisher.Type) {
-			cmd.Println()
-			cmd.Println("To download the results, execute:")
-			cmd.Println("\t" + os.Args[0] + " job get " + jobID)
-		}
-	}
-
-	return nil
-}
-
-func (j *JobProgressPrinter) followLogs(jobID string, cmd *cobra.Command) error {
-	cmd.Printf("Job successfully submitted. Job ID: %s\n", jobID)
-	cmd.Printf("Waiting for logs... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
-
-	// Wait until the job has actually been accepted and started, otherwise this will fail waiting for
-	// the execution to appear.
-	for i := 0; i < 10; i++ {
-		resp, err := j.client.Jobs().Get(cmd.Context(), &apimodels.GetJobRequest{
-			JobID: jobID,
-		})
-		if err != nil {
-			return fmt.Errorf("failed getting job: %w", err)
-		}
-		if resp.Job.State.StateType != models.JobStateTypePending {
-			break
-		}
-		// TODO: add exponential backoff if there were no state updates
-		time.Sleep(time.Duration(1) * time.Second)
-	}
-
-	return util.Logs(cmd, j.client, util.LogOptions{
-		JobID:  jobID,
-		Follow: true,
-	})
-}
-
-//nolint:gocyclo,funlen
-func (j *JobProgressPrinter) followProgress(ctx context.Context, jobID string, cmd *cobra.Command, quiet bool) error {
-	getMoreInfoString := fmt.Sprintf(`
-To get more information at any time, run:
-   bacalhau job describe %s`, jobID)
-
-	cancelString := fmt.Sprintf(`
-To cancel the job, run:
-   bacalhau job stop %s`, jobID)
-
-	if !quiet {
-		cmd.Printf("Job successfully submitted. Job ID: %s\n", jobID)
-		cmd.Print("Checking job status... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
-	}
-
-	cmdShuttingDown := false
-	var returnError error = nil
-
-	// Capture Ctrl + C if the user wants to finish the job early
-	ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
-	signalChan := make(chan os.Signal, 2)
-	signal.Notify(signalChan, util.ShutdownSignals...)
-	defer func() {
-		signal.Stop(signalChan)
-		cancel()
-	}()
-
-	liveTableWriter := util.NewLiveTableWriter()
-	if quiet {
-		cmd.SetOut(io.Discard)
-	} else {
-		cmd.SetOut(liveTableWriter)
-	}
-
-	// goroutine for handling SIGINT from the signal channel, or context completion messages.
-	go func() {
-		for {
-			select {
-			case s := <-signalChan:
-				log.Ctx(ctx).Debug().Msgf("Captured %v. Exiting...", s)
-				if s == os.Interrupt {
-					cmdShuttingDown = true
-					cmd.SetOut(os.Stdout)
-
-					if !quiet {
-						cmd.Println("\n\n\rPrintout canceled (the job is still running).")
-						cmd.Println(getMoreInfoString)
-						cmd.Println(cancelString)
-					}
-					returnError = fmt.Errorf("%s", PrintoutCanceledButRunningNormally)
-				} else {
-					cmd.Println("Unexpected signal received. Exiting.")
-				}
-				cancel()
-			case <-ctx.Done():
-				cmdShuttingDown = true
-				cmd.SetOut(os.Stdout)
-
-				if !quiet {
-					cmd.Println("\n\n\rPrintout canceled due to timeout (the job is still running).")
-					cmd.Println(getMoreInfoString)
-					cmd.Println(cancelString)
-				}
-				returnError = fmt.Errorf("%s", PrintoutCanceledButRunningNormally)
-			}
-			return
-		}
-	}()
-
-	var currentJobState models.JobStateType
-	nextToken := ""
-	jobProgressEvents := make(map[string]*jobProgressEvent)
-	tableOptions := output.OutputOptions{
-		Format:  output.TableFormat,
-		NoStyle: true,
-	}
-
-	timeFilter := time.Now().Unix()
-
-	for !cmdShuttingDown {
-		// With new history events, we no longer support job state. Thus we need
-		// to fetch the job to get the job state, and determine if the job has completed or not.
-		resp, err := j.client.Jobs().Get(ctx, &apimodels.GetJobRequest{
-			JobID: jobID,
-		})
-
-		if err != nil {
-			if _, ok := err.(*bacerrors.ContextCanceledError); ok {
-				// We're done, the user canceled the job
-				cmdShuttingDown = true
-				continue
-			} else {
-				return errors.Wrap(err, "Error getting job")
-			}
-		}
-
-		currentJobState = resp.Job.State.StateType
-
-		// Have we been cancel(l)ed?
-		if condition := ctx.Err(); condition != nil {
-			break
-		}
-
-		// We are separating out 2 concerns here
-		// 1. If the Job is in terminal state, we just fetch all the remaining history events, as
-		//    the command will shut down after it.
-		// 2. If the Job is not in terminal state, we fetch single events, this is better while observing
-		//    history as only one line gets updated. Hence a good user experience.
-		var jobHistoryRequest apimodels.ListJobHistoryRequest
-		if currentJobState.IsTerminal() {
-			jobHistoryRequest = apimodels.ListJobHistoryRequest{
-				JobID:     jobID,
-				EventType: "all",
-				Since:     timeFilter,
-			}
-			if currentJobState != models.JobStateTypeCompleted {
-				returnError = errors.New("job failed")
-			}
-			cmdShuttingDown = true
-		} else {
-			jobHistoryRequest = apimodels.ListJobHistoryRequest{
-				JobID:     jobID,
-				EventType: "all",
-				Since:     timeFilter,
-				BaseListRequest: apimodels.BaseListRequest{
-					NextToken: nextToken,
-					Limit:     1,
-				},
-			}
-		}
-
-		jobHistoryResponse, _ := j.client.Jobs().History(ctx, &jobHistoryRequest)
-		// We need to make sure if the items returned are zero, then we update timeFilter to current time
-		// As or else the timeFilter would be an older history event, we would keep on getting already
-		// displayed history events.
-		if len(jobHistoryResponse.Items) == 0 {
-			timeFilter = time.Now().Unix()
-		}
-
-		nextToken = jobHistoryResponse.NextToken
-
-		for _, history := range jobHistoryResponse.Items {
-			timeFilter = history.Time.Unix()
-
-			// We group based at 2 levels
-			// 1. Per Execution
-			// 2. Job Level State Changes
-			var eventID string
-			if history.Type == models.JobHistoryTypeExecutionLevel {
-				eventID = history.ExecutionID
-			} else {
-				eventID = history.Type.String()
-			}
-
-			jobProgressEvents[eventID] = &jobProgressEvent{
-				jobID:       jobID,
-				occurred:    history.Occurred(),
-				executionID: history.ExecutionID,
-				event:       history.Event,
-			}
-		}
-
-		// Displays Job Progress Output in Table Format
-		entries := lo.Entries(jobProgressEvents)
-
-		// We do custom sorting mainly because, there is a chance that
-		// both execution level and job level states have same timestamp. In that scenarion,
-		// we need to make sure the order is still determinant. The table sorted does not
-		// support this hence we do custom sorting. rd
-		sort.Slice(entries, func(i, j int) bool {
-			if entries[i].Value.occurred.Before(entries[j].Value.occurred) {
-				return true
-			} else if entries[i].Value.occurred.After(entries[j].Value.occurred) {
-				return false
-			} else {
-				return entries[i].Value.event.Topic < entries[j].Value.event.Topic
-			}
-		})
-		if err := output.Output(cmd, jobProgressEventCols, tableOptions,
-			lo.Map(entries, func(item lo.Entry[string, *jobProgressEvent], index int) *jobProgressEvent {
-				return item.Value
-			})); err != nil {
-			return fmt.Errorf("failed to print job progress: %w", err)
-		}
-
-		// Have we been cancel(l)ed?
-		if condition := ctx.Err(); condition != nil {
-			break
-		}
-
-		time.Sleep(time.Millisecond * 500) //nolint:mnd // 500ms sleep
-	}
-
-	// This is needed as while printing progress, we delegate printing job progress table
-	// to Live Table Writer. We eventually close it. Now since it is closed, we want the
-	// command to redirect the output to Stdout again.
-	cmd.SetOut(os.Stdout)
-	return returnError
-}
diff --git a/cmd/util/printer/progress_printer.go b/cmd/util/printer/progress_printer.go
new file mode 100644
index 0000000000..4a0ce86162
--- /dev/null
+++ b/cmd/util/printer/progress_printer.go
@@ -0,0 +1,277 @@
+package printer
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/samber/lo"
+	"github.com/spf13/cobra"
+
+	"github.com/bacalhau-project/bacalhau/cmd/util"
+	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/publicapi/apimodels"
+	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
+)
+
+const progressTimeout = 5 * time.Minute
+
+type JobProgressPrinter struct {
+	client          clientv2.API
+	runtimeSettings *cliflags.RunTimeSettings
+}
+
+func NewJobProgressPrinter(client clientv2.API, runtimeSettings *cliflags.RunTimeSettings) *JobProgressPrinter {
+	return &JobProgressPrinter{
+		client:          client,
+		runtimeSettings: runtimeSettings,
+	}
+}
+
+// PrintJobProgress displays the job progress based on CLI runtime settings
+func (j *JobProgressPrinter) PrintJobProgress(ctx context.Context, job *models.Job, cmd *cobra.Command) error {
+	// If we are in `--wait=false` print the job.ID and then exit.
+	// All the code after this point is to show the progress of the job.
+	if !j.runtimeSettings.WaitForJobToFinish {
+		cmd.Print(job.ID + "\n")
+		return nil
+	}
+
+	// If we are n `id-only` mode - print the id
+	if j.runtimeSettings.PrintJobIDOnly {
+		cmd.Print(job.ID + "\n")
+	}
+
+	// Follow Logs
+	if j.runtimeSettings.Follow {
+		return j.followLogs(ctx, job, cmd)
+	}
+
+	// Follow Progress
+	if err := j.followProgress(ctx, job, cmd); err != nil {
+		PrintError(cmd, err)
+	}
+
+	if err := j.printNodeDetails(ctx, cmd, job); err != nil {
+		return err
+	}
+
+	j.printJobDetailsInstructions(cmd, job)
+
+	return nil
+}
+
+func (j *JobProgressPrinter) followLogs(ctx context.Context, job *models.Job, cmd *cobra.Command) error {
+	if !j.isQuiet() {
+		cmd.Printf("Job successfully submitted. Job ID: %s\n", job.ID)
+		cmd.Printf("Waiting for logs... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
+	}
+
+	// Wait until the job has actually been accepted and started, otherwise this will fail waiting for
+	// the execution to appear.
+	for i := 0; i < 10; i++ {
+		resp, err := j.client.Jobs().Get(ctx, &apimodels.GetJobRequest{JobID: job.ID})
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return nil
+			}
+			return fmt.Errorf("failed getting job: %w", err)
+		}
+		if resp.Job.State.StateType != models.JobStateTypePending {
+			break
+		}
+		// TODO: add exponential backoff if there were no state updates
+		time.Sleep(time.Second)
+	}
+
+	return util.Logs(cmd, j.client, util.LogOptions{JobID: job.ID, Follow: true})
+}
+
+func (j *JobProgressPrinter) followProgress(ctx context.Context, job *models.Job, cmd *cobra.Command) error {
+	ctx, cancel := context.WithTimeout(ctx, progressTimeout)
+	defer cancel()
+
+	if !j.isQuiet() {
+		cmd.Printf("Job successfully submitted. Job ID: %s\n", job.ID)
+		cmd.Print("Checking job status... (Enter Ctrl+C to exit at any time, your job will continue running):\n\n")
+	}
+
+	eventPrinter := j.createEventPrinter(cmd)
+
+	eventChan := make(chan *models.JobHistory)
+	errChan := make(chan error, 1)
+	go j.fetchEvents(ctx, job, eventChan, errChan)
+
+	// process events until the context is canceled or the job is done
+	if err := j.handleEvents(ctx, eventPrinter, eventChan, errChan); err != nil {
+		if errors.Is(err, context.DeadlineExceeded) {
+			j.printTimeoutMessage(cmd)
+			return nil
+		} else if errors.Is(err, context.Canceled) {
+			j.printCancellationMessage(cmd)
+			return nil
+		}
+		return err
+	}
+
+	return j.checkFinalJobState(ctx, job, cmd)
+}
+
+func (j *JobProgressPrinter) fetchEvents(ctx context.Context, job *models.Job, eventChan chan<- *models.JobHistory, errChan chan<- error) {
+	defer close(eventChan)
+
+	// Create a ticker that ticks every 500 milliseconds
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+
+	var nextToken string
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			response, err := j.client.Jobs().History(ctx, &apimodels.ListJobHistoryRequest{
+				JobID:     job.ID,
+				EventType: "all",
+				BaseListRequest: apimodels.BaseListRequest{
+					NextToken: nextToken,
+					Limit:     10,
+				},
+			})
+			if err != nil {
+				errChan <- err
+				return
+			}
+
+			for _, event := range response.Items {
+				eventChan <- event
+				// If the job is long running, we only care when an execution is created
+				if job.IsLongRunning() && event.IsExecutionLevel() {
+					return
+				}
+			}
+
+			if response.NextToken == "" {
+				return
+			}
+			nextToken = response.NextToken
+		}
+	}
+}
+
+func (j *JobProgressPrinter) handleEvents(ctx context.Context, printer eventPrinter, eventChan <-chan *models.JobHistory, errChan <-chan error) error {
+	defer printer.close() // close the printer to clear any spinner before exiting
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case event, ok := <-eventChan:
+			if !ok {
+				return nil // All events processed
+			}
+			if err := printer.printEvent(event); err != nil {
+				return err
+			}
+		case err := <-errChan:
+			return err
+		}
+	}
+}
+
+func (j *JobProgressPrinter) checkFinalJobState(ctx context.Context, job *models.Job, cmd *cobra.Command) error {
+	resp, err := j.client.Jobs().Get(ctx, &apimodels.GetJobRequest{JobID: job.ID})
+	if err != nil {
+		return fmt.Errorf("failed getting job: %w", err)
+	}
+	if resp.Job.State.StateType == models.JobStateTypeFailed {
+		return fmt.Errorf("job failed")
+	}
+	return nil
+}
+
+func (j *JobProgressPrinter) createEventPrinter(cmd *cobra.Command) eventPrinter {
+	if j.isQuiet() {
+		return newQuitePrinter(cmd)
+	}
+	if j.runtimeSettings.GroupEvents {
+		return newGroupedEventPrinter(cmd)
+	}
+	return newSequentialEventPrinter(cmd)
+}
+
+func (j *JobProgressPrinter) printNodeDetails(ctx context.Context, cmd *cobra.Command, job *models.Job) error {
+	if !j.runtimeSettings.PrintNodeDetails {
+		return nil
+	}
+
+	executions, err := j.client.Jobs().Executions(ctx, &apimodels.ListJobExecutionsRequest{JobID: job.ID})
+	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			return nil
+		}
+		return fmt.Errorf("failed getting job executions: %w", err)
+	}
+
+	summary := SummariseExecutions(executions.Items)
+	if len(summary) > 0 {
+		cmd.Println("\nJob Results By Node:")
+		for message, runs := range summary {
+			nodes := len(lo.Uniq(runs))
+			prefix := fmt.Sprintf("• Node %s: ", runs[0])
+			if len(runs) > 1 {
+				prefix = fmt.Sprintf("• %d runs on %d nodes: ", len(runs), nodes)
+			}
+			printIndentedString(cmd, prefix, strings.TrimSpace(message), none, 0)
+		}
+	} else {
+		cmd.Println()
+	}
+
+	return nil
+}
+
+func (j *JobProgressPrinter) printJobDetailsInstructions(cmd *cobra.Command, job *models.Job) {
+	if j.isQuiet() {
+		return
+	}
+	cmd.Println()
+	cmd.Println("To get more details about the run, execute:")
+	cmd.Printf("\t%s job describe %s\n", os.Args[0], job.ID)
+
+	cmd.Println()
+	cmd.Println("To get more details about the run executions, execute:")
+	cmd.Printf("\t%s job executions %s\n", os.Args[0], job.ID)
+
+	// Only print help for downloading the job if it contained a publisher.
+	if !lo.IsEmpty(job.Task().Publisher.Type) {
+		cmd.Println()
+		cmd.Println("To download the results, execute:")
+		cmd.Printf("\t%s job get %s\n", os.Args[0], job.ID)
+	}
+}
+
+func (j *JobProgressPrinter) printCancellationMessage(cmd *cobra.Command) {
+	if !j.isQuiet() {
+		cmd.Println()
+		PrintWarning(cmd, "Progress tracking canceled. The job is still running.")
+	}
+}
+
+func (j *JobProgressPrinter) printTimeoutMessage(cmd *cobra.Command) {
+	if !j.isQuiet() {
+		cmd.Println()
+		PrintWarning(cmd, fmt.Sprintf("Progress tracking timed out after %s. The job is still running.\n", progressTimeout))
+	}
+}
+
+// isQuiet returns true if the printer is in quite mode
+// If we are only printing the id, set the rest of the output to "quiet",
+// i.e. don't print
+func (j *JobProgressPrinter) isQuiet() bool {
+	return j.runtimeSettings.PrintJobIDOnly
+}
diff --git a/cmd/util/printer/print.go b/cmd/util/printer/utils.go
similarity index 86%
rename from cmd/util/printer/print.go
rename to cmd/util/printer/utils.go
index 275ce9eb94..f9ddbb3c01 100644
--- a/cmd/util/printer/print.go
+++ b/cmd/util/printer/utils.go
@@ -13,19 +13,22 @@ import (
 	"golang.org/x/exp/maps"
 	"golang.org/x/term"
 
+	"github.com/bacalhau-project/bacalhau/cmd/util/output"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 )
 
 var (
-	none  = color.New(color.Reset)
-	red   = color.New(color.FgRed)
-	green = color.New(color.FgGreen)
+	none   = color.New(color.Reset)
+	red    = color.New(color.FgRed)
+	yellow = color.New(color.FgYellow)
+	green  = color.New(color.FgGreen)
 )
 
 const (
-	errorPrefix = "Error: "
-	hintPrefix  = "Hint: "
+	errorPrefix   = "Error: "
+	warningPrefix = "Warning: "
+	hintPrefix    = "Hint: "
 )
 
 var terminalWidth int
@@ -53,6 +56,10 @@ func PrintError(cmd *cobra.Command, err error) {
 	printIndentedString(cmd, errorPrefix, err.Error(), red, 0)
 }
 
+func PrintWarning(cmd *cobra.Command, msg string) {
+	printIndentedString(cmd, warningPrefix, msg, yellow, 0)
+}
+
 // Groups the executions in the job state, returning a map of printable messages
 // to node(s) that generated that message.
 func SummariseExecutions(executions []*models.Execution) map[string][]string {
@@ -102,7 +109,7 @@ func printIndentedString(cmd *cobra.Command, prefix, msg string, prefixColor *co
 
 	cmd.PrintErrln()
 	cmd.PrintErr(strings.Repeat(" ", int(startIndent)))
-	prefixColor.Fprintf(cmd.ErrOrStderr(), "%s", prefix)
+	prefixColor.Fprint(cmd.ErrOrStderr(), output.BoldStr(prefix))
 	for i, line := range strings.Split(wordwrap.WrapString(msg, blockTextWidth), "\n") {
 		if i > 0 {
 			cmd.PrintErr(strings.Repeat(" ", blockIndent))
diff --git a/pkg/compute/bidder.go b/pkg/compute/bidder.go
index 3e6218af5e..0d4cfe138a 100644
--- a/pkg/compute/bidder.go
+++ b/pkg/compute/bidder.go
@@ -108,7 +108,7 @@ func (b Bidder) RunBidding(ctx context.Context, bidRequest *BidderRequest) {
 		b.callback.OnComputeFailure(ctx, ComputeError{
 			RoutingMetadata:   routingMetadata,
 			ExecutionMetadata: executionMetadata,
-			Event:             models.EventFromError(EventTopicExecutionBidding, err),
+			Event:             models.EventFromError(EventTopicExecutionScanning, err),
 		})
 		return
 	}
@@ -148,7 +148,7 @@ func (b Bidder) handleBidResult(
 			b.callback.OnComputeFailure(ctx, ComputeError{
 				RoutingMetadata:   routingMetadata,
 				ExecutionMetadata: executionMetadata,
-				Event:             models.EventFromError(EventTopicExecutionBidding, err),
+				Event:             models.EventFromError(EventTopicExecutionScanning, err),
 			})
 		}
 		handleBidComplete = func(ctx context.Context, result *bidStrategyResponse) {
diff --git a/pkg/compute/events.go b/pkg/compute/events.go
index eb4a2881d3..84a5f5a497 100644
--- a/pkg/compute/events.go
+++ b/pkg/compute/events.go
@@ -8,7 +8,7 @@ import (
 )
 
 const (
-	EventTopicExecutionBidding     models.EventTopic = "Requesting Node"
+	EventTopicExecutionScanning    models.EventTopic = "Exec Scanning"
 	EventTopicExecutionDownloading models.EventTopic = "Downloading Inputs"
 	EventTopicExecutionPreparing   models.EventTopic = "Preparing Environment"
 	EventTopicExecutionRunning     models.EventTopic = "Running Execution"
@@ -18,7 +18,7 @@ const (
 func RespondedToBidEvent(response *bidstrategy.BidStrategyResponse) models.Event {
 	return models.Event{
 		Message:   response.Reason,
-		Topic:     EventTopicExecutionBidding,
+		Topic:     EventTopicExecutionScanning,
 		Timestamp: time.Now(),
 		Details:   map[string]string{},
 	}
diff --git a/pkg/config/config.go b/pkg/config/config.go
index af60d86b63..b8b702cd7b 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -113,6 +113,18 @@ func New(opts ...Option) (*Config, error) {
 		return nil, err
 	}
 
+	// To absolute paths for better logging. This is best effort and will not return an error if it fails.
+	for i, path := range c.paths {
+		if !filepath.IsAbs(path) {
+			absPath, err := filepath.Abs(path)
+			if err != nil {
+				log.Debug().Msgf("failed to resolve absolute path for %s: %v", path, err)
+			} else {
+				c.paths[i] = absPath
+			}
+		}
+	}
+
 	// merge the config files in the order they were passed.
 	for _, path := range c.paths {
 		if err := c.Merge(path); err != nil {
@@ -200,15 +212,7 @@ func New(opts ...Option) (*Config, error) {
 		}
 	}
 
-	// log the resolved config paths
-	absoluteConfigPaths := make([]string, len(c.paths))
-	for i, path := range c.paths {
-		absoluteConfigPaths[i], err = filepath.Abs(path)
-		if err != nil {
-			absoluteConfigPaths[i] = path
-		}
-	}
-	log.Info().Msgf("Config loaded from: %s, and with data-dir %s", absoluteConfigPaths, c.base.Get(types.DataDirKey))
+	log.Debug().Msgf("Config loaded from: %s, and with data-dir %s", c.paths, c.base.Get(types.DataDirKey))
 	return c, nil
 }
 
@@ -264,6 +268,12 @@ func (c *Config) ConfigFileUsed() string {
 	return c.base.ConfigFileUsed()
 }
 
+// Paths returns the paths to the configuration files merged
+// from lower index to higher index
+func (c *Config) Paths() []string {
+	return c.paths
+}
+
 // Unmarshal returns the current configuration.
 // Unmarshal returns an error if the configuration cannot be unmarshalled.
 func (c *Config) Unmarshal(out interface{}) error {
diff --git a/pkg/models/error.go b/pkg/models/error.go
index 7d980b84b5..2a0843daf7 100644
--- a/pkg/models/error.go
+++ b/pkg/models/error.go
@@ -13,6 +13,7 @@ const (
 	DetailsKeyHint           = "Hint"
 	DetailsKeyRetryable      = "Retryable"
 	DetailsKeyFailsExecution = "FailsExecution"
+	DetailsKeyNewState       = "NewState"
 )
 
 const (
diff --git a/pkg/models/event.go b/pkg/models/event.go
index f72930bd65..1f4235554c 100644
--- a/pkg/models/event.go
+++ b/pkg/models/event.go
@@ -97,6 +97,29 @@ func (e *Event) WithDetail(key, value string) *Event {
 	return e
 }
 
+// HasError returns true if the event is an error.
+func (e *Event) HasError() bool {
+	return e.Details[DetailsKeyIsError] == "true"
+}
+
+// HasStateUpdate returns true if the event is a state update.
+func (e *Event) HasStateUpdate() bool {
+	_, ok := e.Details[DetailsKeyNewState]
+	return ok
+}
+
+// GetJobStateIfPresent returns the job state from the event, if it is a state update.
+func (e *Event) GetJobStateIfPresent() (JobStateType, error) {
+	if !e.HasStateUpdate() {
+		return JobStateTypeUndefined, nil
+	}
+	var jobState JobStateType
+	if err := jobState.UnmarshalText([]byte(e.Details[DetailsKeyNewState])); err != nil {
+		return JobStateTypeUndefined, err
+	}
+	return jobState, nil
+}
+
 // EventFromError converts an error into an Event tagged with the passed event
 // topic.
 //
diff --git a/pkg/models/job_history.go b/pkg/models/job_history.go
index ed79cc47ff..561c8fd7ec 100644
--- a/pkg/models/job_history.go
+++ b/pkg/models/job_history.go
@@ -68,3 +68,13 @@ func (jh JobHistory) Occurred() time.Time {
 	}
 	return jh.Time
 }
+
+// IsJobLevel returns true if the JobHistory is at the job level.
+func (jh JobHistory) IsJobLevel() bool {
+	return jh.Type == JobHistoryTypeJobLevel
+}
+
+// IsExecutionLevel returns true if the JobHistory is at the execution level.
+func (jh JobHistory) IsExecutionLevel() bool {
+	return jh.Type == JobHistoryTypeExecutionLevel
+}
diff --git a/pkg/orchestrator/events.go b/pkg/orchestrator/events.go
index ad3c116726..3f10539ef0 100644
--- a/pkg/orchestrator/events.go
+++ b/pkg/orchestrator/events.go
@@ -69,7 +69,7 @@ func JobStateUpdateEvent(new models.JobStateType, message ...string) models.Even
 	}
 	return *models.NewEvent(EventTopicJobStateUpdate).
 		WithMessage(eventMessage).
-		WithDetail("NewState", new.String())
+		WithDetail(models.DetailsKeyNewState, new.String())
 }
 
 func JobStoppedEvent(reason string) models.Event {


Test patch:
diff --git a/pkg/models/event_test.go b/pkg/models/event_test.go
index 041e8a004a..45d82d0d0f 100644
--- a/pkg/models/event_test.go
+++ b/pkg/models/event_test.go
@@ -119,6 +119,48 @@ func (suite *EventTestSuite) TestEventFromSimpleError() {
 	suite.Len(event.Details, 1)
 }
 
+func (suite *EventTestSuite) TestHasError() {
+	// Test case for an event with an error
+	eventWithError := NewEvent(suite.topic).WithError(fmt.Errorf("Test error"))
+	suite.True(eventWithError.HasError())
+
+	// Test case for an event without an error
+	eventWithoutError := NewEvent(suite.topic)
+	suite.False(eventWithoutError.HasError())
+}
+
+func (suite *EventTestSuite) TestHasStateUpdate() {
+	// Test case for an event with a state update
+	eventWithStateUpdate := NewEvent(suite.topic).WithDetail(DetailsKeyNewState, "Running")
+	suite.True(eventWithStateUpdate.HasStateUpdate())
+
+	// Test case for an event without a state update
+	eventWithoutStateUpdate := NewEvent(suite.topic)
+	suite.False(eventWithoutStateUpdate.HasStateUpdate())
+}
+
+func (suite *EventTestSuite) TestGetJobStateIfPresent() {
+	// Test case for an event with a valid state update
+	validState := JobStateTypeRunning
+	eventWithValidState := NewEvent(suite.topic).WithDetail(DetailsKeyNewState, validState.String())
+	state, err := eventWithValidState.GetJobStateIfPresent()
+	suite.NoError(err)
+	suite.Equal(validState, state)
+
+	// Test case for an event without a state update
+	eventWithoutState := NewEvent(suite.topic)
+	state, err = eventWithoutState.GetJobStateIfPresent()
+	suite.NoError(err)
+	suite.Equal(JobStateTypeUndefined, state)
+
+	// Test case for an event with an invalid state update
+	invalidState := "InvalidState"
+	eventWithInvalidState := NewEvent(suite.topic).WithDetail(DetailsKeyNewState, invalidState)
+	state, err = eventWithInvalidState.GetJobStateIfPresent()
+	suite.NoError(err) // JobStateType.UnmarshallText() does not return an error for invalid states
+	suite.Equal(JobStateTypeUndefined, state)
+}
+
 func TestEventTestSuite(t *testing.T) {
 	suite.Run(t, new(EventTestSuite))
 }