Add statusCode to WriteJobStatus and status stream responses

Wire Job.StatusCode through the status streaming path so plugins can
forward cluster-specific status codes (e.g. Kubernetes pod phase) to
the Launcher, matching C++ plugin behavior.

Author: dmortondev

api/types.go

@@ -401,7 +401,9 @@ type Job struct {
 	StatusMsg string `json:"statusMessage,omitempty"`
 
 	// The standard code/enum for the current status of the Job, if known.
-	// Optional.
+	// Optional. Carried in job state responses as part of the Job payload,
+	// and forwarded explicitly in job status stream responses via
+	// [launcher.StreamResponseWriter.WriteJobStatus].
 	StatusCode string `json:"statusCode,omitempty"`
 
 	// The process ID of the Job, if applicable. Optional.

cache/cache.go

@@ -249,7 +249,7 @@ func (r *JobCache) StreamJobStatus(ctx context.Context, w launcher.StreamRespons
 	done := false
 	err := r.Lookup(user, id, func(job *api.Job) {
 		//nolint:errcheck // fire-and-forget convenience wrapper
-		w.WriteJobStatus(job.ID, job.Name, job.Status, job.StatusMsg)
+		w.WriteJobStatus(job.ID, job.Name, job.Status, job.StatusCode, job.StatusMsg)
 		// Break off early if we know there will be no further updates.
 		if api.TerminalStatus(job.Status) {
 			done = true
@@ -275,7 +275,7 @@ func (r *JobCache) StreamJobStatus(ctx context.Context, w launcher.StreamRespons
 				return
 			}
 			//nolint:errcheck // fire-and-forget convenience wrapper
-			w.WriteJobStatus(j.ID, j.Name, j.Status, j.StatusMsg)
+			w.WriteJobStatus(j.ID, j.Name, j.Status, j.StatusCode, j.StatusMsg)
 		}
 	}
 }
@@ -286,7 +286,7 @@ func (r *JobCache) StreamJobStatuses(ctx context.Context, w launcher.StreamRespo
 	r.store.JobsForUser(user, nil, func(jobs []*api.Job) {
 		for _, job := range jobs {
 			//nolint:errcheck // fire-and-forget convenience wrapper
-			w.WriteJobStatus(job.ID, job.Name, job.Status, job.StatusMsg)
+			w.WriteJobStatus(job.ID, job.Name, job.Status, job.StatusCode, job.StatusMsg)
 		}
 	})
 	ch := make(chan *statusUpdate, 1)
@@ -300,7 +300,7 @@ func (r *JobCache) StreamJobStatuses(ctx context.Context, w launcher.StreamRespo
 				return
 			}
 			//nolint:errcheck // fire-and-forget convenience wrapper
-			w.WriteJobStatus(j.ID, j.Name, j.Status, j.StatusMsg)
+			w.WriteJobStatus(j.ID, j.Name, j.Status, j.StatusCode, j.StatusMsg)
 		}
 	}
 }
@@ -447,16 +447,17 @@ func (s *subManager) Close() int {
 }
 
 type statusUpdate struct {
-	ID        api.JobID
-	User      string
-	Name      string
-	Status    string
-	StatusMsg string
+	ID         api.JobID
+	User       string
+	Name       string
+	Status     string
+	StatusCode string
+	StatusMsg  string
 }
 
 func newStatusUpdateFromJob(job *api.Job) *statusUpdate {
 	return &statusUpdate{
-		job.ID, job.User, job.Name, job.Status, job.StatusMsg,
+		job.ID, job.User, job.Name, job.Status, job.StatusCode, job.StatusMsg,
 	}
 }
 

internal/protocol/rpc.go

@@ -289,19 +289,20 @@ type JobStatusStreamResponse struct {
 	ID        api.JobID        `json:"id"`
 	Name      string           `json:"name"`
 	Status    string           `json:"status"`
-	Msg       string           `json:"statusMessage,omitempty"`
 	Code      string           `json:"statusCode,omitempty"`
+	Msg       string           `json:"statusMessage,omitempty"`
 }
 
 // NewJobStatusStreamResponse creates a new job status stream response.
-func NewJobStatusStreamResponse(responseID uint64, id, name, status, msg string) *JobStatusStreamResponse {
+func NewJobStatusStreamResponse(responseID uint64, id, name, status, statusCode, msg string) *JobStatusStreamResponse {
 	base := responseBase{responseJobStatus, 0, responseID}
 	return &JobStatusStreamResponse{
 		responseBase: base,
 		Sequences:    []StreamSequence{}, // Ensure we never send null.
 		ID:           api.JobID(id),
 		Name:         name,
 		Status:       status,
+		Code:         statusCode,
 		Msg:          msg,
 	}
 }

internal/protocol/rpc_test.go

@@ -142,6 +142,62 @@ func TestNewMetricsResponse_WithLatency(t *testing.T) {
 	}
 }
 
+func TestNewJobStatusStreamResponse(t *testing.T) {
+	t.Run("all fields present", func(t *testing.T) {
+		resp := NewJobStatusStreamResponse(5, "job-1", "My Job", "Running", "PodRunning", "all good")
+
+		data, err := json.Marshal(resp)
+		if err != nil {
+			t.Fatalf("json.Marshal() error = %v", err)
+		}
+
+		var got map[string]interface{}
+		if err := json.Unmarshal(data, &got); err != nil {
+			t.Fatalf("json.Unmarshal() error = %v", err)
+		}
+
+		if mt := int(got["messageType"].(float64)); mt != 3 {
+			t.Errorf("messageType = %d, want 3", mt)
+		}
+		if id := got["id"].(string); id != "job-1" {
+			t.Errorf("id = %q, want %q", id, "job-1")
+		}
+		if name := got["name"].(string); name != "My Job" {
+			t.Errorf("name = %q, want %q", name, "My Job")
+		}
+		if status := got["status"].(string); status != "Running" {
+			t.Errorf("status = %q, want %q", status, "Running")
+		}
+		if code := got["statusCode"].(string); code != "PodRunning" {
+			t.Errorf("statusCode = %q, want %q", code, "PodRunning")
+		}
+		if msg := got["statusMessage"].(string); msg != "all good" {
+			t.Errorf("statusMessage = %q, want %q", msg, "all good")
+		}
+	})
+
+	t.Run("statusCode omitted when empty", func(t *testing.T) {
+		resp := NewJobStatusStreamResponse(5, "job-1", "My Job", "Running", "", "")
+
+		data, err := json.Marshal(resp)
+		if err != nil {
+			t.Fatalf("json.Marshal() error = %v", err)
+		}
+
+		var got map[string]interface{}
+		if err := json.Unmarshal(data, &got); err != nil {
+			t.Fatalf("json.Unmarshal() error = %v", err)
+		}
+
+		if _, ok := got["statusCode"]; ok {
+			t.Error("statusCode should be omitted when empty")
+		}
+		if _, ok := got["statusMessage"]; ok {
+			t.Error("statusMessage should be omitted when empty")
+		}
+	})
+}
+
 func TestNewConfigReloadResponse_ErrorTypes(t *testing.T) {
 	tests := []struct {
 		name      string

launcher/launcher.go

@@ -296,7 +296,7 @@ type ResponseWriter interface {
 // StreamResponseWriter is the interface for writing streaming responses.
 type StreamResponseWriter interface {
 	ResponseWriter
-	WriteJobStatus(id api.JobID, name, status, msg string) error
+	WriteJobStatus(id api.JobID, name, status, statusCode, msg string) error
 	WriteJobOutput(output string, outputType api.JobOutput) error
 	WriteJobResourceUtil(cpuPercent float64, cpuTime float64,
 		residentMem float64, virtualMem float64) error
@@ -610,13 +610,13 @@ func (w *defaultResponseWriter) WriteJobs(jobs []*api.Job) error {
 // non-stream response writer.
 var errNotStreamWriter = fmt.Errorf("method called on non-stream response writer")
 
-func (w *defaultResponseWriter) WriteJobStatus(id api.JobID, name, status, msg string) error {
+func (w *defaultResponseWriter) WriteJobStatus(id api.JobID, name, status, statusCode, msg string) error {
 	if w.store == nil {
 		return errNotStreamWriter
 	}
 	rid := w.req.ID()
 	resp := protocol.NewJobStatusStreamResponse(nextResponseID(), string(id),
-		name, status, msg)
+		name, status, statusCode, msg)
 	resp.Sequences = []protocol.StreamSequence{
 		{RequestID: rid, SequenceID: w.store.SequenceID(rid)},
 	}

plugintest/example_test.go

@@ -176,8 +176,8 @@ func TestMockStreamResponseWriter(t *testing.T) {
 	t.Run("captures status updates", func(t *testing.T) {
 		w := plugintest.NewMockStreamResponseWriter()
 
-		w.WriteJobStatus("job-1", "My Job", api.StatusRunning, "Job started")
-		w.WriteJobStatus("job-1", "My Job", api.StatusFinished, "Job completed")
+		w.WriteJobStatus("job-1", "My Job", api.StatusRunning, "", "Job started")
+		w.WriteJobStatus("job-1", "My Job", api.StatusFinished, "", "Job completed")
 
 		plugintest.AssertStatusCount(t, w, 2)
 

plugintest/mocks.go

@@ -212,10 +212,11 @@ type MockStreamResponseWriter struct {
 
 // StatusUpdate represents a job status update.
 type StatusUpdate struct {
-	ID      api.JobID
-	Name    string
-	Status  string
-	Message string
+	ID         api.JobID
+	Name       string
+	Status     string
+	StatusCode string
+	Message    string
 }
 
 // OutputChunk represents a chunk of job output.
@@ -249,14 +250,15 @@ func NewMockStreamResponseWriter() *MockStreamResponseWriter {
 }
 
 // WriteJobStatus implements launcher.StreamResponseWriter.
-func (m *MockStreamResponseWriter) WriteJobStatus(id api.JobID, name, status, msg string) error {
+func (m *MockStreamResponseWriter) WriteJobStatus(id api.JobID, name, status, statusCode, msg string) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.Statuses = append(m.Statuses, StatusUpdate{
-		ID:      id,
-		Name:    name,
-		Status:  status,
-		Message: msg,
+		ID:         id,
+		Name:       name,
+		Status:     status,
+		StatusCode: statusCode,
+		Message:    msg,
 	})
 	return nil
 }