feat(schedulerclient): WatchJobs SSE stream + fix ListJobs filter param

Add WatchJobs(ctx) to the Client interface and HTTPClient: opens the
apiserver's /xce/v1/events SSE stream and yields JobEvent values until the
context is cancelled or the stream ends. JobEvent carries the event type,
job id, status string, exit code, and a raw job snapshot.

Also fix ListJobs' filter params: the apiserver accepts filters via the
"filter" query param in field:op:value form (e.g. queue_name:eq:normal);
the individual queue_name/submit_user params it ignores with a 400, which
broke every filtered list. Multiple conditions are brace-wrapped {a},{b}.

Verified live on t1: WatchJobs captured job.updated PEND→RUN→DONE for a
submitted job.

Co-Authored-By: Claude <noreply@anthropic.com>

src/frontend/schedulerclient/client.go

@@ -50,4 +50,12 @@ type Client interface {
 
 	// ReleaseJob resumes a suspended job (action "resume", default SIGCONT).
 	ReleaseJob(ctx context.Context, id string) error
+
+	// ── Event stream (Phase 2) ──
+
+	// WatchJobs subscribes to the live job-event SSE stream and returns a
+	// channel of lifecycle events (job.created / job.updated / job.deleted /
+	// job.finished). The channel is closed when the context is cancelled or
+	// the stream ends. The caller is responsible for cancelling ctx.
+	WatchJobs(ctx context.Context) (<-chan JobEvent, error)
 }

src/frontend/schedulerclient/httpclient.go

@@ -1,6 +1,7 @@
 package schedulerclient
 
 import (
+	"bufio"
 	"context"
 	"crypto/tls"
 	"encoding/json"
@@ -44,11 +45,19 @@ func (hc *HTTPClient) ListJobs(ctx context.Context, filter JobFilter) ([]*URMJob
 	url := hc.baseURL + "/xce/v1/jobs"
 	req := hc.rc.R().SetContext(ctx).SetHeader("Authorization", hc.token)
 
+	// The apiserver's job list accepts filters via the "filter" query param in
+	// "field:op:value" form (e.g. queue_name:eq:normal); the individual
+	// queue_name/submit_user params it ignores (400). Multiple conditions are
+	// brace-wrapped: {a},{b}.
+	var conds []string
 	if filter.Queue != "" {
-		req.SetQueryParam("queue_name", filter.Queue)
+		conds = append(conds, "queue_name:eq:"+filter.Queue)
 	}
 	if filter.Owner != "" {
-		req.SetQueryParam("submit_user", filter.Owner)
+		conds = append(conds, "submit_user:eq:"+filter.Owner)
+	}
+	if len(conds) > 0 {
+		req.SetQueryParam("filter", "{"+strings.Join(conds, "},{")+"}")
 	}
 	if filter.Limit > 0 {
 		req.SetQueryParam("limit", strconv.Itoa(filter.Limit))
@@ -503,3 +512,61 @@ type apiHost struct {
 	NumUSusp   int     `json:"num_ususp"`
 	NumDone    int     `json:"num_done"`
 }
+
+// ── Event stream (Phase 2) ──────────────────────────────────────────────────
+
+// WatchJobs subscribes to the apiserver's live job-event SSE endpoint and
+// yields lifecycle events on the returned channel. The connection is kept
+// open until ctx is cancelled or the server ends the stream; the channel is
+// closed when the reader terminates. Malformed frames are skipped.
+func (hc *HTTPClient) WatchJobs(ctx context.Context) (<-chan JobEvent, error) {
+	// SSE is a long-lived stream — no request timeout. TLS skip matches
+	// NewClient's InsecureSkipVerify behaviour for the apiserver endpoint.
+	client := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		},
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, hc.baseURL+"/xce/v1/events", nil)
+	if err != nil {
+		return nil, fmt.Errorf("watch jobs: %w", err)
+	}
+	req.Header.Set("Authorization", hc.token)
+	req.Header.Set("Accept", "text/event-stream")
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("watch jobs: %w", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		return nil, fmt.Errorf("watch jobs: apiserver returned status %d", resp.StatusCode)
+	}
+
+	ch := make(chan JobEvent, 64)
+	go func() {
+		defer close(ch)
+		defer resp.Body.Close()
+
+		scanner := bufio.NewScanner(resp.Body)
+		scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+		for scanner.Scan() {
+			line := strings.TrimSpace(scanner.Text())
+			if !strings.HasPrefix(line, "data: ") {
+				continue
+			}
+			var ev JobEvent
+			if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil {
+				continue // skip malformed frame rather than kill the stream
+			}
+			select {
+			case ch <- ev:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return ch, nil
+}

src/frontend/schedulerclient/types.go

@@ -103,3 +103,33 @@ type Metrics struct {
 	PendingJobs int     `json:"pending_jobs"`
 	Utilization float64 `json:"utilization"` // 0.0 – 1.0
 }
+
+// ── Job lifecycle events (Phase 2, SSE WatchJobs) ────────────────────────────
+
+// JobEvent is a single job lifecycle event received from the apiserver's live
+// SSE stream (/xce/v1/events). Type is one of:
+//
+//	job.created / job.updated / job.deleted / job.finished
+//
+// Status is the backend-native job status string (PEND/RUN/DONE/EXIT/...)
+// when the event carries a job snapshot. For job.finished, ExitCode holds the
+// process exit code (0 → DONE, non-zero → EXIT).
+type JobEvent struct {
+	Type     string      `json:"type"`
+	JobID    uint64      `json:"job_id,omitempty"`
+	ExitCode int         `json:"exit_code,omitempty"`
+	Status   string      `json:"job_status,omitempty"`
+	Job      *JobEventJob `json:"job,omitempty"`
+}
+
+// JobEventJob mirrors the apiserver job snapshot embedded in a JobEvent.
+// Only the fields the scheduler agent consumes are modelled.
+type JobEventJob struct {
+	UniqueJobID int64  `json:"unique_job_id"`
+	JobID       int64  `json:"job_id"`
+	SubmitUser  string `json:"submit_user"`
+	JobStatus   int    `json:"job_status"`
+	JobName     string `json:"job_name"`
+	Command     string `json:"command"`
+	QueueName   string `json:"queue_name"`
+}

src/frontend/schedulerclient/watch_test.go

@@ -0,0 +1,116 @@
+package schedulerclient
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestWatchJobsReceivesEvents verifies WatchJobs parses SSE frames from a
+// live stream into JobEvent values.
+func TestWatchJobsReceivesEvents(t *testing.T) {
+	payload := `{"type":"job.updated","job_id":123,"job_status":"RUN","job":{"unique_job_id":123,"job_id":123,"submit_user":"mcp-agent","job_name":"wtest","command":"sleep 5","queue_name":"normal"}}`
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		if got := r.Header.Get("Authorization"); got == "" {
+			t.Errorf("expected Authorization header, got empty")
+		}
+		if got := r.URL.Path; got != "/xce/v1/events" {
+			t.Errorf("unexpected path %q", got)
+		}
+		flusher, _ := w.(http.Flusher)
+		_, _ = w.Write([]byte("data: " + payload + "\n\n"))
+		flusher.Flush()
+	}))
+	defer srv.Close()
+
+	hc := NewClient(srv.URL, "xce_ak_test")
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	ch, err := hc.WatchJobs(ctx)
+	if err != nil {
+		t.Fatalf("WatchJobs: %v", err)
+	}
+
+	var got *JobEvent
+	for ev := range ch {
+		got = &ev
+		break
+	}
+	if got == nil {
+		t.Fatal("did not receive any event")
+	}
+	if got.Type != "job.updated" || got.JobID != 123 || got.Status != "RUN" {
+		t.Fatalf("unexpected event: %+v", got)
+	}
+	if got.Job == nil || got.Job.Command != "sleep 5" || got.Job.QueueName != "normal" {
+		t.Fatalf("event job payload wrong: %+v", got)
+	}
+}
+
+// TestWatchJobsSkipsMalformedFrames verifies a malformed SSE frame does not
+// kill the stream — the next valid frame is still delivered.
+func TestWatchJobsSkipsMalformedFrames(t *testing.T) {
+	payload := `{"type":"job.finished","job_id":7,"exit_code":0,"job_status":"DONE"}`
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		flusher, _ := w.(http.Flusher)
+		_, _ = w.Write([]byte("data: {not json\n\n"))
+		flusher.Flush()
+		_, _ = w.Write([]byte("data: " + payload + "\n\n"))
+		flusher.Flush()
+	}))
+	defer srv.Close()
+
+	hc := NewClient(srv.URL, "xce_ak_test")
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	ch, err := hc.WatchJobs(ctx)
+	if err != nil {
+		t.Fatalf("WatchJobs: %v", err)
+	}
+
+	var got *JobEvent
+	for ev := range ch {
+		got = &ev
+		break
+	}
+	if got == nil {
+		t.Fatal("did not receive valid event after malformed frame")
+	}
+	if got.Type != "job.finished" || got.JobID != 7 || got.ExitCode != 0 {
+		t.Fatalf("unexpected event: %+v", got)
+	}
+}
+
+// TestJobEventJSONRoundTrip ensures the JobEvent JSON shape matches what the
+// apiserver SSE endpoint emits.
+func TestJobEventJSONRoundTrip(t *testing.T) {
+	raw := `{"type":"job.created","job_id":5,"job_status":"PEND","job":{"unique_job_id":5,"job_id":5,"submit_user":"u","job_name":"n","command":"echo hi","queue_name":"q"}}`
+	var ev JobEvent
+	if err := json.Unmarshal([]byte(raw), &ev); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if ev.Type != "job.created" || ev.JobID != 5 || ev.Status != "PEND" {
+		t.Fatalf("unexpected event: %+v", ev)
+	}
+	if ev.Job == nil || ev.Job.JobName != "n" || ev.Job.Command != "echo hi" {
+		t.Fatalf("unexpected job payload: %+v", ev.Job)
+	}
+	// Re-marshal and confirm it stays a valid single-line SSE payload.
+	out, err := json.Marshal(ev)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if strings.Contains(string(out), "\n") {
+		t.Fatalf("marshalled event must not contain newline (SSE framing): %s", out)
+	}
+}