From 55a5d6de80e1def6238255c1a210a621f1dabfa9 Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Wed, 26 Aug 2026 23:06:06 +0800 Subject: [PATCH] Add Run control event stream --- platform/api/resource_handlers.go | 106 +++++++++++++++++++++++++++ platform/api/routes.md | 3 +- platform/domain/control.go | 28 +++++++ platform/dto/control.go | 22 ++++++ platform/protocol/run-contracts.md | 7 +- platform/service/control_events.go | 77 +++++++++++++++++++ platform/service/job_channel.go | 1 + platform/service/job_channel_test.go | 18 +++++ platform/service/resources.go | 8 ++ platform/validator/control.go | 7 ++ 10 files changed, 274 insertions(+), 3 deletions(-) create mode 100644 platform/service/control_events.go diff --git a/platform/api/resource_handlers.go b/platform/api/resource_handlers.go index 283b3c1..cb7317c 100644 --- a/platform/api/resource_handlers.go +++ b/platform/api/resource_handlers.go @@ -1,10 +1,13 @@ package api import ( + "encoding/json" + "fmt" "io" "net/http" "strconv" "strings" + "time" "browser.local/platform/domain" "browser.local/platform/dto" @@ -109,6 +112,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/server-instances/{id}", h.serverInstanceDetail) mux.HandleFunc("/api/v1/run/control/hello", h.runControlHello) mux.HandleFunc("/api/v1/run/control/heartbeat", h.requireRunSignature(h.runControlHeartbeat)) + mux.HandleFunc("/api/v1/run/control/events", h.requireRunSignature(h.runControlEvents)) mux.HandleFunc("/api/v1/run/lifecycle/report", h.requireRunSignature(h.runLifecycleReport)) mux.HandleFunc("/api/v1/run/jobs/claim", h.requireRunSignature(h.runJobClaim)) mux.HandleFunc("/api/v1/run/jobs/ack", h.requireRunSignature(h.runJobAck)) @@ -1691,6 +1695,108 @@ func (h *coreHandlers) runControlHeartbeat(w http.ResponseWriter, r *http.Reques writeJSON(w, http.StatusOK, dto.RunControlHeartbeatFromDomain(result)) } +// runControlEvents godoc +// @Summary Stream lightweight Run control events +// @Description Opens a signed Run-only event stream for wakeups such as queued job changes. Events never carry job payloads, logs, artifacts, host paths, or credentials. +// @Tags run +// @Accept json +// @Produce text/event-stream +// @Param body body dto.RunControlStreamRequest true "Run control stream request" +// @Success 200 {object} dto.RunControlEventResponse +// @Failure 400 {object} dto.ErrorResponse +// @Failure 401 {object} dto.ErrorResponse +// @Failure 405 {object} dto.ErrorResponse +// @Router /api/v1/run/control/events [post] +func (h *coreHandlers) runControlEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeMethodNotAllowed(w, http.MethodPost) + return + } + request, err := decodeJSON[dto.RunControlStreamRequest](r) + if err != nil { + writeDecodeError(w, err) + return + } + subscription, err := h.core.SubscribeRunControlEvents(request.ToDomain()) + if err != nil { + writeServiceError(w, err) + return + } + defer subscription.Cancel() + flusher, ok := w.(http.Flusher) + if !ok { + writeServiceError(w, validator.ValidationError{Violations: []string{"streaming response is unavailable"}}) + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + lastSeq := request.LastEventSeq + ready := dto.RunControlEventResponse{RunEndpointID: request.RunEndpointID, Sequence: lastSeq, Type: domain.RunControlEventTypeReady, ServerTime: time.Now().UTC(), RetrySeconds: 5} + if err := writeRunControlSSE(w, flusher, ready); err != nil { + return + } + if subscription.Initial != nil { + event := dto.RunControlEventFromDomain(*subscription.Initial) + if event.Sequence > lastSeq { + lastSeq = event.Sequence + } + if err := writeRunControlSSE(w, flusher, event); err != nil { + return + } + } + heartbeatTicker := time.NewTicker(15 * time.Second) + defer heartbeatTicker.Stop() + for { + select { + case <-r.Context().Done(): + return + case event, ok := <-subscription.Events: + if !ok { + return + } + response := dto.RunControlEventFromDomain(event) + if response.Sequence > lastSeq { + lastSeq = response.Sequence + } + if err := writeRunControlSSE(w, flusher, response); err != nil { + return + } + case <-heartbeatTicker.C: + heartbeat := dto.RunControlEventResponse{RunEndpointID: request.RunEndpointID, Sequence: lastSeq, Type: domain.RunControlEventTypeHeartbeat, ServerTime: time.Now().UTC(), RetrySeconds: 5} + if err := writeRunControlSSE(w, flusher, heartbeat); err != nil { + return + } + } + } +} + +func writeRunControlSSE(w http.ResponseWriter, flusher http.Flusher, event dto.RunControlEventResponse) error { + data, err := json.Marshal(event) + if err != nil { + return err + } + if event.Type != "" { + if _, err := fmt.Fprintf(w, "event: %s\n", event.Type); err != nil { + return err + } + } + if _, err := fmt.Fprintf(w, "id: %d\n", event.Sequence); err != nil { + return err + } + if event.RetrySeconds > 0 { + if _, err := fmt.Fprintf(w, "retry: %d\n", event.RetrySeconds*1000); err != nil { + return err + } + } + if _, err := fmt.Fprintf(w, "data: %s\n\n", data); err != nil { + return err + } + flusher.Flush() + return nil +} + // runLifecycleReport godoc // @Summary Report autonomous run lifecycle result // @Description Lets a registered run endpoint report an observed lifecycle terminal result without a platform-assigned job lease. diff --git a/platform/api/routes.md b/platform/api/routes.md index 8b78dd2..2cbcf9d 100644 --- a/platform/api/routes.md +++ b/platform/api/routes.md @@ -170,13 +170,14 @@ The following signed routes are Run-only and never part of browser/plugin DTOs: - `POST /api/v1/run/control/hello`: accept `RunControlHelloRequest`, create or update run endpoint metadata, and return `RunControlHelloResponse` with a platform-issued session token. - `POST /api/v1/run/control/heartbeat`: accept `RunControlHeartbeatRequest`, require the active session token, update heartbeat metadata, and return `RunControlHeartbeatResponse` with the next heartbeat hint and optional capability refresh request. +- `POST /api/v1/run/control/events`: accept signed `RunControlStreamRequest` and keep a `text/event-stream` response open for lightweight Run wakeups such as `control.ready`, `control.heartbeat`, and `job.changed`. The stream never carries job payloads, logs, artifact bytes, host paths, credentials, or direct sockets. Run control actions carry only lightweight metadata: endpoint ID, display name, version, status, capability fingerprint/list, capacity, session token, and timing hints. They do not carry job bodies, logs, artifact chunks, host paths, raw credentials, or direct sockets. Control is the highest-priority run-facing channel; artifact/file transfer pressure must not delay heartbeat processing or mutate endpoint capacity through heavy payload fields. ## Implemented Run Job Actions -- `POST /api/v1/run/jobs/claim`: accept `RunJobClaimRequest`, validate the active Run session, optionally hold the request for a bounded `waitSeconds` window, sweep expired endpoint work, and durably claim one eligible queued/retrying job with a monotonic per-job attempt, hashed lease credential, ack deadline, and execution lease. +- `POST /api/v1/run/jobs/claim`: accept `RunJobClaimRequest`, validate the active Run session, optionally support bounded legacy `waitSeconds`, sweep expired endpoint work, and durably claim one eligible queued/retrying job with a monotonic per-job attempt, hashed lease credential, ack deadline, and execution lease. Current workers should use the persistent control event stream for wakeups and keep claim as the durable assignment fetch. - `POST /api/v1/run/jobs/ack`: accept `RunJobAckRequest`, fence endpoint/session generation/attempt/lease, reject late acknowledgements, and move the current attempt into running state. - `POST /api/v1/run/jobs/progress`: accept `RunJobProgressRequest`, reject stale sequences and expired/old attempts, persist bounded progress, and renew the current execution lease. - `POST /api/v1/run/jobs/result`: accept `RunJobResultRequest` and write an idempotent terminal result or durable retry-wait transition with capped exponential backoff. diff --git a/platform/domain/control.go b/platform/domain/control.go index 50cec3b..6991848 100644 --- a/platform/domain/control.go +++ b/platform/domain/control.go @@ -2,6 +2,12 @@ package domain import "time" +const ( + RunControlEventTypeReady = "control.ready" + RunControlEventTypeHeartbeat = "control.heartbeat" + RunControlEventTypeJobChanged = "job.changed" +) + type RunCapabilityReport struct { Capabilities []string Fingerprint string @@ -53,6 +59,20 @@ type RunControlHeartbeatResult struct { ServerTime time.Time } +type RunControlStreamRequest struct { + RunEndpointID string + SessionToken string + LastEventSeq uint64 +} + +type RunControlEvent struct { + RunEndpointID string + Sequence uint64 + Type string + ServerTime time.Time + RetrySeconds int +} + type RunLifecycleReport struct { RunEndpointID string SessionToken string @@ -126,6 +146,14 @@ func CopyRunControlHeartbeatResult(result RunControlHeartbeatResult) RunControlH return result } +func CopyRunControlStreamRequest(request RunControlStreamRequest) RunControlStreamRequest { + return request +} + +func CopyRunControlEvent(event RunControlEvent) RunControlEvent { + return event +} + func CopyRunLifecycleReport(report RunLifecycleReport) RunLifecycleReport { report.ExecutionResult.ServerDeploymentEvidence = CopyServerDeploymentEvidence(report.ExecutionResult.ServerDeploymentEvidence) report.ExecutionResult.DeploymentReceipt = CopyServerDeploymentExecutionReceipt(report.ExecutionResult.DeploymentReceipt) diff --git a/platform/dto/control.go b/platform/dto/control.go index 3588a8c..f0a1159 100644 --- a/platform/dto/control.go +++ b/platform/dto/control.go @@ -57,6 +57,20 @@ type RunControlHeartbeatResponse struct { ServerTime time.Time `json:"serverTime"` } +type RunControlStreamRequest struct { + RunEndpointID string `json:"runEndpointId"` + SessionToken string `json:"sessionToken"` + LastEventSeq uint64 `json:"lastEventSeq,omitempty"` +} + +type RunControlEventResponse struct { + RunEndpointID string `json:"runEndpointId"` + Sequence uint64 `json:"sequence"` + Type string `json:"type"` + ServerTime time.Time `json:"serverTime"` + RetrySeconds int `json:"retrySeconds,omitempty"` +} + type RunLifecycleReportRequest struct { RunEndpointID string `json:"runEndpointId"` SessionToken string `json:"sessionToken"` @@ -115,6 +129,10 @@ func (request RunControlHeartbeatRequest) ToDomain() domain.RunControlHeartbeat } } +func (request RunControlStreamRequest) ToDomain() domain.RunControlStreamRequest { + return domain.RunControlStreamRequest{RunEndpointID: request.RunEndpointID, SessionToken: request.SessionToken, LastEventSeq: request.LastEventSeq} +} + func (request RunLifecycleReportRequest) ToDomain() domain.RunLifecycleReport { return domain.RunLifecycleReport{ RunEndpointID: request.RunEndpointID, @@ -155,6 +173,10 @@ func RunControlHeartbeatFromDomain(result domain.RunControlHeartbeatResult) RunC } } +func RunControlEventFromDomain(event domain.RunControlEvent) RunControlEventResponse { + return RunControlEventResponse{RunEndpointID: event.RunEndpointID, Sequence: event.Sequence, Type: event.Type, ServerTime: event.ServerTime, RetrySeconds: event.RetrySeconds} +} + func RunLifecycleReportFromDomain(result domain.RunLifecycleReportResult) RunLifecycleReportResponse { result = domain.CopyRunLifecycleReportResult(result) return RunLifecycleReportResponse{Accepted: result.Accepted, RunEndpointID: result.RunEndpointID, ServerInstanceID: result.ServerInstanceID, ProjectedState: result.ProjectedState, ServerTime: result.ServerTime} diff --git a/platform/protocol/run-contracts.md b/platform/protocol/run-contracts.md index 1462a06..7e4122f 100644 --- a/platform/protocol/run-contracts.md +++ b/platform/protocol/run-contracts.md @@ -8,6 +8,7 @@ Implemented HTTP JSON routes: - `POST /api/v1/run/control/hello` - `POST /api/v1/run/control/heartbeat` +- `POST /api/v1/run/control/events` Named control DTOs: @@ -15,10 +16,12 @@ Named control DTOs: - `RunHelloResponse` - `RunHeartbeatRequest` - `RunHeartbeatResponse` +- `RunControlStreamRequest` +- `RunControlEvent` - `RunCapabilityReport` - `RunCapacityReport` -Control payloads must remain small and must not include logs, artifact chunks, host paths, raw credentials, direct sockets, or long task results. Hello creates or updates run endpoint metadata and issues an in-memory platform session token. A server-scoped generated Run must use the endpoint identity reserved for its server; Platform rejects a valid component key presented for another endpoint. Registration is binding/authentication only for generated Run bootstrap and must not enqueue lifecycle or status jobs merely because Run appeared. Heartbeat requires that active session token and may request capability refresh when the fingerprint changes. +Control payloads must remain small and must not include logs, artifact chunks, host paths, raw credentials, direct sockets, job assignments, execution input, or long task results. Hello creates or updates run endpoint metadata and issues an in-memory platform session token. A server-scoped generated Run must use the endpoint identity reserved for its server; Platform rejects a valid component key presented for another endpoint. Registration is binding/authentication only for generated Run bootstrap and must not enqueue lifecycle or status jobs merely because Run appeared. Heartbeat requires that active session token and may request capability refresh when the fingerprint changes. The control event stream is a signed Run-only `text/event-stream` wake channel; events such as `job.changed` only tell Run to claim durable work through `/run/jobs/claim`. Capacity reports include bounded `maxJobs`, `runningJobs`, `queuedJobs`, `logBacklogBatches`, `artifactBacklogChunks`, and enumerated pressure codes. They report queue and spool counts only, never log bodies, artifact chunks, machine paths, PIDs, sockets, credentials, or transport endpoints. @@ -46,7 +49,7 @@ Named job DTOs: - `RunJobReconcileRequest` - `RunJobReconcileResponse` -Jobs must carry bounded metadata such as `jobId`, `runEndpointId`, `serverInstanceId`, `capability`, `idempotencyKey`, lease token, attempt, progress, terminal state, message, error code, and result reference. Job payloads must not carry logs, artifact chunks, host paths, raw credentials, direct sockets, or large inline result bodies. +Jobs must carry bounded metadata such as `jobId`, `runEndpointId`, `serverInstanceId`, `capability`, `idempotencyKey`, lease token, attempt, progress, terminal state, message, error code, and result reference. Job payloads must not carry logs, artifact chunks, host paths, raw credentials, direct sockets, or large inline result bodies. Current Run workers use the persistent control event stream for wakeups and keep claim/ack/progress/result as the durable lease and execution channel; bounded claim long-poll remains a compatibility fallback for older workers. Plugin lifecycle assignments may add only a validated plugin identifier, enumerated lifecycle operation, target version, and logical workspace scope. Explicit operator-requested install, enable, disable, upgrade, rollback, retire, dependency-check, and bounded lifecycle commands remain Platform-authorized jobs. Generated Run package startup is not dependent on registration-time job assignment; it is driven by the autonomous lifecycle plan embedded by the platform builder. Assignments cannot carry arbitrary shell, provider configuration, raw credentials, host paths, PIDs, sockets, DSNs, or RCON secrets. diff --git a/platform/service/control_events.go b/platform/service/control_events.go new file mode 100644 index 0000000..7d5e416 --- /dev/null +++ b/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 = © + } + 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: + } + } +} diff --git a/platform/service/job_channel.go b/platform/service/job_channel.go index ac56713..3b8789f 100644 --- a/platform/service/job_channel.go +++ b/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 { diff --git a/platform/service/job_channel_test.go b/platform/service/job_channel_test.go index 27a6d81..65cc25e 100644 --- a/platform/service/job_channel_test.go +++ b/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) diff --git a/platform/service/resources.go b/platform/service/resources.go index e014100..9dda60c 100644 --- a/platform/service/resources.go +++ b/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{}, diff --git a/platform/validator/control.go b/platform/validator/control.go index 6db8c51..9b8b298 100644 --- a/platform/validator/control.go +++ b/platform/validator/control.go @@ -59,6 +59,13 @@ func ValidateRunControlHeartbeat(heartbeat domain.RunControlHeartbeat) error { return finish(violations) } +func ValidateRunControlStreamRequest(request domain.RunControlStreamRequest) error { + var violations []string + violations = appendRequired(violations, "runEndpointId", request.RunEndpointID) + violations = appendRequired(violations, "sessionToken", request.SessionToken) + return finish(violations) +} + func appendCapacityViolations(violations []string, capacity domain.RunCapacity) []string { if capacity.MaxJobs < 0 || capacity.RunningJobs < 0 || capacity.QueuedJobs < 0 || capacity.LogBacklogBatches < 0 || capacity.ArtifactBacklogChunks < 0 { violations = append(violations, "capacity counts must not be negative")