Add Run control event stream

platform/service/control_events.go

@@ -0,0 +1,77 @@
+package service
+
+import (
+	"sync"
+
+	"browser.local/platform/domain"
+	"browser.local/platform/validator"
+)
+
+const defaultRunControlEventRetrySeconds = 5
+
+type RunControlEventSubscription struct {
+	Initial *domain.RunControlEvent
+	Events  <-chan domain.RunControlEvent
+	Cancel  func()
+}
+
+func (svc *CoreService) SubscribeRunControlEvents(request domain.RunControlStreamRequest) (RunControlEventSubscription, error) {
+	request = domain.CopyRunControlStreamRequest(request)
+	if err := validator.ValidateRunControlStreamRequest(request); err != nil {
+		return RunControlEventSubscription{}, err
+	}
+	if err := svc.validateRunSession(request.RunEndpointID, request.SessionToken); err != nil {
+		return RunControlEventSubscription{}, err
+	}
+
+	updates := make(chan domain.RunControlEvent, 8)
+	var initial *domain.RunControlEvent
+	svc.controlStreamMu.Lock()
+	if latest, exists := svc.controlStreamEvents[request.RunEndpointID]; exists && latest.Sequence > request.LastEventSeq {
+		copy := domain.CopyRunControlEvent(latest)
+		initial = &copy
+	}
+	svc.controlStreamWaiters[request.RunEndpointID] = append(svc.controlStreamWaiters[request.RunEndpointID], updates)
+	svc.controlStreamMu.Unlock()
+
+	var once sync.Once
+	cancel := func() {
+		once.Do(func() {
+			svc.controlStreamMu.Lock()
+			waiters := svc.controlStreamWaiters[request.RunEndpointID]
+			for index, candidate := range waiters {
+				if candidate == updates {
+					waiters = append(waiters[:index], waiters[index+1:]...)
+					break
+				}
+			}
+			if len(waiters) == 0 {
+				delete(svc.controlStreamWaiters, request.RunEndpointID)
+			} else {
+				svc.controlStreamWaiters[request.RunEndpointID] = waiters
+			}
+			svc.controlStreamMu.Unlock()
+			close(updates)
+		})
+	}
+	return RunControlEventSubscription{Initial: initial, Events: updates, Cancel: cancel}, nil
+}
+
+func (svc *CoreService) publishRunControlEvent(runEndpointID string, eventType string) {
+	if runEndpointID == "" || eventType == "" {
+		return
+	}
+	svc.controlStreamMu.Lock()
+	sequence := svc.controlStreamSeq[runEndpointID] + 1
+	svc.controlStreamSeq[runEndpointID] = sequence
+	event := domain.RunControlEvent{RunEndpointID: runEndpointID, Sequence: sequence, Type: eventType, ServerTime: svc.now(), RetrySeconds: defaultRunControlEventRetrySeconds}
+	svc.controlStreamEvents[runEndpointID] = event
+	waiters := append([]chan domain.RunControlEvent(nil), svc.controlStreamWaiters[runEndpointID]...)
+	svc.controlStreamMu.Unlock()
+	for _, waiter := range waiters {
+		select {
+		case waiter <- event:
+		default:
+		}
+	}
+}

platform/service/job_channel.go

@@ -146,6 +146,7 @@ func (svc *CoreService) notifyRunJobWaiters(runEndpointID string) {
 	for _, waiter := range waiters {
 		close(waiter)
 	}
+	svc.publishRunControlEvent(runEndpointID, domain.RunControlEventTypeJobChanged)
 }
 
 func withoutCapability(capabilities []string, forbidden string) []string {

platform/service/job_channel_test.go

@@ -128,6 +128,24 @@ func TestCoreServiceRunJobClaimWithWaitWakesOnCreate(t *testing.T) {
 	}
 }
 
+func TestCoreServiceRunControlSubscriptionWakesOnCreateJob(t *testing.T) {
+	svc, sessionToken := newRegisteredRunJobService(t)
+	subscription, err := svc.SubscribeRunControlEvents(domain.RunControlStreamRequest{RunEndpointID: "run-local", SessionToken: sessionToken})
+	if err != nil {
+		t.Fatalf("subscribe control events: %v", err)
+	}
+	defer subscription.Cancel()
+	createQueuedRunJob(t, svc, "job-control-wake", "idem-control-wake")
+	select {
+	case event := <-subscription.Events:
+		if event.Type != domain.RunControlEventTypeJobChanged || event.RunEndpointID != "run-local" || event.Sequence == 0 {
+			t.Fatalf("unexpected control event: %+v", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for control wake event")
+	}
+}
+
 func TestCoreServiceRunJobClaimSkipsServerFileCapabilityWithoutDeclaration(t *testing.T) {
 	svc := newTestCoreService()
 	plugin, endpoint := createPluginAndRunEndpoint(t, svc)

platform/service/resources.go

@@ -86,6 +86,7 @@ type Core interface {
 	ListRunEndpoints(domain.RunEndpointFilter) ([]domain.RunEndpoint, error)
 	RegisterRunHello(domain.RunControlHello) (domain.RunControlHelloResult, error)
 	AcceptRunHeartbeat(domain.RunControlHeartbeat) (domain.RunControlHeartbeatResult, error)
+	SubscribeRunControlEvents(domain.RunControlStreamRequest) (RunControlEventSubscription, error)
 	AuthorizeRunRequestSignature(domain.RunRequestSignature) error
 	CreateServerInstance(domain.ServerInstance) (domain.ServerInstance, error)
 	CreateServerInstanceForSession(string, domain.ServerInstance) (domain.ServerInstance, error)
@@ -233,6 +234,10 @@ type CoreService struct {
 	controlMu             sync.Mutex
 	runSessions           map[string]domain.RunControlSession
 	runSessionSeq         uint64
+	controlStreamMu       sync.Mutex
+	controlStreamSeq      map[string]uint64
+	controlStreamEvents   map[string]domain.RunControlEvent
+	controlStreamWaiters  map[string][]chan domain.RunControlEvent
 	jobMu                 sync.Mutex
 	jobWaitMu             sync.Mutex
 	jobWaiters            map[string][]chan struct{}
@@ -284,6 +289,9 @@ func newCoreServiceWithLogStore(store repo.Store, logStore LogBodyStore, now fun
 		now:                   now,
 		authSessions:          map[string]string{},
 		runSessions:           map[string]domain.RunControlSession{},
+		controlStreamSeq:      map[string]uint64{},
+		controlStreamEvents:   map[string]domain.RunControlEvent{},
+		controlStreamWaiters:  map[string][]chan domain.RunControlEvent{},
 		jobWaiters:            map[string][]chan struct{}{},
 		logStore:              logStore,
 		logProjectionStates:   map[string]map[string]pluginLogSequenceState{},