diff --git a/AGENTS.md b/AGENTS.md index 550d718..6d7ecfb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -95,6 +95,8 @@ Do not hardcode game-specific deployment behavior in run or platform services. V When a game needs "install if missing, update if present, then start" behavior, implement it as plugin-owned lifecycle actions. Platform may package and dispatch those actions, and run may execute them through generic capabilities, but neither platform nor run should special-case a game by name to perform those steps. +Realtime log collection must be plugin-declared and plugin-configured. For live terminal output, prefer the plugin-declared supervised process channels (`process.stdout` / `process.stderr`) for the process that run started, and treat game-specific file tails only as plugin-declared sources for history, fallback, or explicit backfill. Do not inspect or prioritize a game log file such as `SCUM.log` merely because it exists on disk; if a plugin needs a file tail, window/console capture, startup flag, hidden window behavior, or another live-log source, declare that behavior in the plugin manifest/action/config and keep run/platform generic. + Run-platform communication must remain channelized: - Control is lightweight and high priority. diff --git a/openspec/changes/stream-live-server-logs-sse/design.md b/openspec/changes/stream-live-server-logs-sse/design.md new file mode 100644 index 0000000..c9563fc --- /dev/null +++ b/openspec/changes/stream-live-server-logs-sse/design.md @@ -0,0 +1,15 @@ +## Overview + +Realtime browser log display is a one-way stream, so the platform exposes Server-Sent Events instead of WebSocket for this change. SSE gives the browser one long-lived HTTP response, works with standard `EventSource`, carries same-origin HttpOnly session cookies, and only needs Nginx buffering disabled. + +## Transport Boundary + +Run continues to upload logs through `POST /api/v1/run/logs/batches`. That path remains durable and retryable: Run writes to local spool, sends bounded batches, receives sequence ACKs, and can retry without depending on browser presence. + +Run log capture remains plugin-declared. For live terminal output, Run captures the supervised process channels declared by the plugin as `process.stdout` / `process.stderr`; game file tails such as `file.tail` are explicit plugin-declared history/backfill sources rather than a generic default. A managed process started by Run writes stdout/stderr into Run-owned capture files and Run tails those capture files into durable batch ingest, so restarting Run can resume transmission for an already-running supervised process without inspecting game-specific logs such as `SCUM.log`. + +The browser subscribes to `GET /api/v1/server-instances/{id}/logs/events`. Platform authorizes the user session against the server instance, replays a bounded recent history per stream, then publishes newly ingested log entries from memory fan-out. `POST /api/v1/log-streams/query` stays as an explicit historical cursor API, not a realtime polling loop. + +## Proxy Notes + +SSE does not require `Upgrade` or `Connection: upgrade`. Reverse proxies must avoid buffering the stream and should keep the upstream read timeout long enough for idle log periods. diff --git a/openspec/changes/stream-live-server-logs-sse/proposal.md b/openspec/changes/stream-live-server-logs-sse/proposal.md new file mode 100644 index 0000000..8422cab --- /dev/null +++ b/openspec/changes/stream-live-server-logs-sse/proposal.md @@ -0,0 +1,16 @@ +## Why + +The browser terminal and server log detail view were polling `POST /api/v1/log-streams/query` on a short interval, and terminal polling multiplied that request count by every candidate log stream. This wastes HTTP requests and can overload the platform or reverse proxy while still failing to feel truly realtime. + +## What Changes + +- Add a platform-owned `GET /api/v1/server-instances/{id}/logs/events` Server-Sent Events stream for browser live logs. +- Keep Run-to-Platform log transport as durable signed HTTP batch ingest with local spool, sequence acknowledgement, and cursor query for history/reconnect repair. +- Switch the server detail log view, live log drawer, and management terminal to a single EventSource connection with bounded initial history instead of periodic `/log-streams/query` polling. +- Add Nginx proxy settings for unbuffered SSE forwarding. + +## Impact + +- Affects `platform/` API, DTO, service log ingest fan-out, tests, and protocol docs. +- Affects `platform_web/` API types/client, log UI components, tests, and Nginx config. +- Does not add WebSocket terminal transport, direct browser-to-Run connections, plugin-held platform keys, host paths, credentials, or game-specific log behavior in platform/run. diff --git a/openspec/changes/stream-live-server-logs-sse/specs/browser-live-log-streaming/spec.md b/openspec/changes/stream-live-server-logs-sse/specs/browser-live-log-streaming/spec.md new file mode 100644 index 0000000..2b3be03 --- /dev/null +++ b/openspec/changes/stream-live-server-logs-sse/specs/browser-live-log-streaming/spec.md @@ -0,0 +1,52 @@ +## ADDED Requirements + +### Requirement: Browser live logs use a platform push stream +The Platform SHALL provide a server-scoped browser log event stream that sends safe log stream metadata and log entries over Server-Sent Events. + +#### Scenario: Operator opens live logs +- **WHEN** an authorized operator opens a server log view or management terminal +- **THEN** platform_web opens `GET /api/v1/server-instances/{id}/logs/events` with `EventSource` +- **AND** the view does not start a periodic `/api/v1/log-streams/query` polling loop + +#### Scenario: Initial history is replayed +- **WHEN** the browser opens the log event stream with a bounded `historyLimit` +- **THEN** Platform replays recent stored entries for the server's log streams before sending the ready event +- **AND** each event contains only safe stream metadata and log entry fields + +### Requirement: Durable Run log ingest remains independent +Run-to-Platform log transfer SHALL remain durable HTTP batch ingest with local spool and sequence acknowledgement. Browser streaming MUST fan out only from platform-ingested log data. + +#### Scenario: Run uploads a batch +- **WHEN** Run uploads a valid contiguous log batch +- **THEN** Platform stores it, updates the stream latest sequence, acknowledges the batch, and publishes the new entries to matching browser subscribers +- **AND** duplicate batch acknowledgements do not publish duplicate browser events + +### Requirement: Run live logs follow plugin-declared process channels +Run live terminal output SHALL come from plugin-declared log sources and the supervised process that Run started. Game-specific file logs MUST NOT be used as the default live terminal source unless the plugin declares that file source for explicit history, fallback, or backfill. + +#### Scenario: Run starts a supervised process +- **WHEN** Run executes a plugin lifecycle start action in supervised mode +- **THEN** Run captures the process stdout and stderr into Run-owned durable capture files +- **AND** Run tails those capture files into durable log batch ingest using the plugin-declared process stream keys +- **AND** Run hides the managed Windows process window when the OS supports hidden startup + +#### Scenario: Run restarts while the game process remains alive +- **WHEN** Run restarts and reloads its persisted process journal for an already-running supervised process +- **THEN** Run resumes tailing the Run-owned stdout/stderr capture files from persisted offsets +- **AND** Run does not inspect game-specific logs such as `SCUM.log` to synthesize terminal output + +### Requirement: Browser log streaming uses platform session authorization +The log event stream SHALL be authorized by the current platform user session and server access rules. Plugins and browser code MUST NOT receive Run session tokens, component keys, host paths, raw credentials, or direct Run socket information. + +#### Scenario: Unauthorized user subscribes +- **WHEN** a user without access opens a server log event stream +- **THEN** Platform rejects the request using the existing authorization error behavior +- **AND** no log entries or stream metadata are sent + +### Requirement: Reverse proxies forward log events without buffering +The deployed platform_web reverse proxy SHALL forward the log event route without response buffering and with a long read timeout so idle log periods do not force browser polling. + +#### Scenario: Nginx proxies SSE +- **WHEN** Nginx forwards `/api/v1/server-instances/{id}/logs/events` +- **THEN** buffering is disabled for that location +- **AND** the route does not require WebSocket upgrade headers diff --git a/openspec/changes/stream-live-server-logs-sse/tasks.md b/openspec/changes/stream-live-server-logs-sse/tasks.md new file mode 100644 index 0000000..73ae54d --- /dev/null +++ b/openspec/changes/stream-live-server-logs-sse/tasks.md @@ -0,0 +1,29 @@ +## Prompt Boundaries + +- Positive prompt (正向提示词): Realtime server logs and the management terminal must use one platform-owned push stream per open view, with bounded history replay and no periodic `/log-streams/query` polling. +- Directional prompt (方向提示词): Work inside `platform/`, `platform_web/`, and `run/`, preserve durable Run log ingest, plugin-declared log-source ownership, platform session authorization, existing UI controls, and Nginx proxying; verify with focused Go/frontend tests plus structure checks. +- Boundary prompt (任务边界): Do not add browser-to-Run sockets, raw credentials, host paths, plugin-owned auth keys, billing/cloud workflows, or game-specific SCUM paths/commands to platform/run code. + +## 1. Platform SSE Contract + +- [x] 1.1 Add safe log event DTOs and a server-scoped SSE route. +- [x] 1.2 Publish newly accepted log batch entries to non-blocking server-instance subscribers. +- [x] 1.3 Replay bounded recent history on stream open while keeping cursor query available for explicit history. + +## 2. Frontend Realtime Logs + +- [x] 2.1 Add EventSource client support for server log events. +- [x] 2.2 Replace live log drawer and management terminal `/log-streams/query` intervals with SSE. +- [x] 2.3 Replace server detail log polling with SSE history replay and live append. + +## 3. Proxy And Verification + +- [x] 3.1 Disable Nginx buffering for the log events route. +- [x] 3.2 Add focused backend/frontend tests for SSE behavior and no terminal polling. +- [x] 3.3 Run final verification: platform Go tests, platform_web tests/typecheck, `scripts/check-structure.sh`, and strict OpenSpec validation. + +## 4. Run Process Log Capture + +- [x] 4.1 Capture supervised process stdout/stderr through Run-owned durable capture files instead of game-specific log inspection. +- [x] 4.2 Resume managed process log tailing after Run restart using the persisted process journal and capture offsets. +- [x] 4.3 Keep plugin-declared log source keys optional for runtime readiness; explicit file backfill still uses the requested plugin source. diff --git a/platform/api/log_events_handlers.go b/platform/api/log_events_handlers.go new file mode 100644 index 0000000..890eb1d --- /dev/null +++ b/platform/api/log_events_handlers.go @@ -0,0 +1,187 @@ +package api + +import ( + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "browser.local/platform/domain" + "browser.local/platform/dto" + "browser.local/platform/service" +) + +const ( + defaultLogEventHistoryLimit = 100 + maxLogEventHistoryLimit = 500 + logEventHeartbeatInterval = 15 * time.Second +) + +// serverLogEvents godoc +// @Summary Stream server log events +// @Description Streams safe server log entries over Server-Sent Events. Durable cursor query remains available for history and reconnect repair. +// @Tags logs +// @Produce text/event-stream +// @Param id path string true "Server instance ID" +// @Param historyLimit query int false "Recent entries per stream to replay before live events" +// @Success 200 {string} string "event-stream" +// @Failure 401 {object} dto.ErrorResponse +// @Failure 403 {object} dto.ErrorResponse +// @Failure 404 {object} dto.ErrorResponse +// @Failure 405 {object} dto.ErrorResponse +// @Router /api/v1/server-instances/{id}/logs/events [get] +func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeMethodNotAllowed(w, http.MethodGet) + return + } + instance, streams, subscription, err := h.openLogEventSubscription(r) + if err != nil { + writeServiceError(w, err) + return + } + defer subscription.Close() + flusher, ok := w.(http.Flusher) + if !ok { + writeServiceError(w, fmt.Errorf("streaming response unsupported")) + return + } + + header := w.Header() + header.Set("Content-Type", "text/event-stream") + header.Set("Cache-Control", "no-cache, no-transform") + header.Set("Connection", "keep-alive") + header.Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + + historyLimit := parseLogEventHistoryLimit(r.URL.Query().Get("historyLimit")) + for _, stream := range streams { + if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil { + return + } + if historyLimit > 0 { + if err := h.writeLogEventHistory(w, stream, historyLimit); err != nil { + _ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()}) + flusher.Flush() + return + } + } + } + if err := writeSSEJSON(w, "ready", "", dto.LogStreamEventsReadyResponse{ServerInstanceID: instance.ID, StreamCount: len(streams), ServerTime: time.Now().UTC()}); err != nil { + return + } + flusher.Flush() + + heartbeat := time.NewTicker(logEventHeartbeatInterval) + defer heartbeat.Stop() + for { + select { + case <-r.Context().Done(): + return + case event, ok := <-subscription.Events: + if !ok { + return + } + if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { + return + } + flusher.Flush() + case <-heartbeat.C: + if _, err := fmt.Fprintf(w, ": heartbeat %s\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil { + return + } + flusher.Flush() + } + } +} + +func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, service.LogEventSubscription, error) { + var instance domain.ServerInstance + var streams []domain.LogStream + var subscription service.LogEventSubscription + var err error + if h.enforceAuthorization { + sessionID := bearerToken(r) + instance, err = h.core.GetServerInstanceForSession(sessionID, r.PathValue("id")) + if err != nil { + return domain.ServerInstance{}, nil, subscription, err + } + streams, err = h.core.ListLogStreamsForSession(sessionID, domain.LogStreamFilter{ServerInstanceID: instance.ID}) + if err != nil { + return domain.ServerInstance{}, nil, subscription, err + } + subscription, err = h.core.SubscribeLogEventsForSession(sessionID, instance.ID) + } else { + instance, err = h.core.GetServerInstance(r.PathValue("id")) + if err != nil { + return domain.ServerInstance{}, nil, subscription, err + } + streams, err = h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID}) + if err != nil { + return domain.ServerInstance{}, nil, subscription, err + } + subscription, err = h.core.SubscribeLogEvents(instance.ID) + } + return instance, streams, subscription, err +} + +func (h *coreHandlers) writeLogEventHistory(w http.ResponseWriter, stream domain.LogStream, limit int) error { + afterSeq := uint64(0) + if stream.LatestSeq > uint64(limit) { + afterSeq = stream.LatestSeq - uint64(limit) + } + cursor, err := h.core.QueryLogStream(domain.LogStreamCursorQuery{LogStreamID: stream.ID, AfterSeq: afterSeq, Limit: limit}) + if err != nil { + return err + } + for _, entry := range cursor.Entries { + event := domain.LogStreamEvent{ServerInstanceID: stream.ServerInstanceID, Stream: stream, Entry: entry, LatestSeq: cursor.LatestSeq} + if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { + return err + } + } + return nil +} + +func parseLogEventHistoryLimit(value string) int { + if strings.TrimSpace(value) == "" { + return defaultLogEventHistoryLimit + } + limit, err := strconv.Atoi(value) + if err != nil || limit < 0 { + return defaultLogEventHistoryLimit + } + if limit > maxLogEventHistoryLimit { + return maxLogEventHistoryLimit + } + return limit +} + +func writeSSEJSON(w http.ResponseWriter, eventName string, id string, value any) error { + payload, err := json.Marshal(value) + if err != nil { + return err + } + if id != "" { + if _, err := fmt.Fprintf(w, "id: %s\n", sanitizeSSEField(id)); err != nil { + return err + } + } + if _, err := fmt.Fprintf(w, "event: %s\n", sanitizeSSEField(eventName)); err != nil { + return err + } + _, err = fmt.Fprintf(w, "data: %s\n\n", payload) + return err +} + +func sanitizeSSEField(value string) string { + value = strings.ReplaceAll(value, "\r", "") + value = strings.ReplaceAll(value, "\n", "") + return value +} + +func logEventID(event domain.LogStreamEvent) string { + return fmt.Sprintf("%s:%d", sanitizeSSEField(event.Stream.ID), event.Entry.Seq) +} diff --git a/platform/api/log_ingest_handlers_test.go b/platform/api/log_ingest_handlers_test.go index d63f825..b2ff7a3 100644 --- a/platform/api/log_ingest_handlers_test.go +++ b/platform/api/log_ingest_handlers_test.go @@ -1,7 +1,10 @@ package api import ( + "bufio" "net/http" + "net/http/httptest" + "strings" "testing" "time" @@ -35,6 +38,32 @@ func TestLogIngestAPIWorkflow(t *testing.T) { } } +func TestLogEventsSSEReplaysHistory(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + batch := validLogBatchRequest(t, hello.SessionToken, 1, 2) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch), http.StatusOK) + + server := httptest.NewServer(router) + defer server.Close() + client := server.Client() + client.Timeout = 2 * time.Second + response, err := client.Get(server.URL + "/api/v1/server-instances/server-1/logs/events?historyLimit=2") + if err != nil { + t.Fatalf("open log event stream: %v", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK || !strings.HasPrefix(response.Header.Get("Content-Type"), "text/event-stream") { + t.Fatalf("unexpected event stream response: status=%d content-type=%q", response.StatusCode, response.Header.Get("Content-Type")) + } + body := readSSEUntil(t, response, "event: ready") + for _, fragment := range []string{"event: stream", "event: log", `"streamId":"log-1"`, `"seq":1`, `"seq":2`} { + if !strings.Contains(body, fragment) { + t.Fatalf("expected SSE body to contain %q, got:\n%s", fragment, body) + } + } +} + func TestLogIngestAPIDuplicateAndErrors(t *testing.T) { router := newTestRouter() hello := createLogIngestAPIFixtures(t, router) @@ -77,6 +106,20 @@ func createLogIngestAPIFixtures(t *testing.T, router http.Handler) dto.RunContro return hello } +func readSSEUntil(t *testing.T, response *http.Response, marker string) string { + t.Helper() + reader := bufio.NewReader(response.Body) + var body strings.Builder + for !strings.Contains(body.String(), marker) { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("read event stream: %v\n%s", err, body.String()) + } + body.WriteString(line) + } + return body.String() +} + func validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) dto.LogBatchIngestRequest { t.Helper() entries := make([]dto.LogEntryBody, 0, lastSeq-firstSeq+1) diff --git a/platform/api/resource_handlers.go b/platform/api/resource_handlers.go index 3ed41a3..4dcc42b 100644 --- a/platform/api/resource_handlers.go +++ b/platform/api/resource_handlers.go @@ -108,6 +108,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/server-instances/{id}/dependencies/install", h.serverDependenciesInstall) mux.HandleFunc("/api/v1/server-instances/{id}/dependencies", h.serverDependencies) mux.HandleFunc("/api/v1/server-instances/{id}/logs/live", h.serverLiveLogs) + mux.HandleFunc("/api/v1/server-instances/{id}/logs/events", h.serverLogEvents) mux.HandleFunc("/api/v1/server-instances/{id}/logs/backfill", h.serverLogsBackfill) mux.HandleFunc("/api/v1/server-instances/{id}/config/diff", h.serverInstanceConfigDiff) mux.HandleFunc("/api/v1/server-instances/{id}/config/approve", h.serverInstanceConfigApprove) diff --git a/platform/api/resource_handlers_test.go b/platform/api/resource_handlers_test.go index ee4da41..a24fb44 100644 --- a/platform/api/resource_handlers_test.go +++ b/platform/api/resource_handlers_test.go @@ -114,7 +114,18 @@ func TestCoreAPICreateListDetailWorkflows(t *testing.T) { } getJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams/log-1") streams := getJSON[dto.LogStreamListResponse](t, router, "/api/v1/log-streams?serverInstanceId=server-1&streamKey=stdout") - assertListCount(t, streams.Count, 1) + if streams.Count < 1 { + t.Fatalf("expected stdout streams, got %+v", streams) + } + foundExplicitStream := false + for _, stream := range streams.Items { + if stream.ID == "log-1" { + foundExplicitStream = true + } + } + if !foundExplicitStream { + t.Fatalf("expected explicitly created stream in list, got %+v", streams) + } auditResponse := postJSON[dto.AuditEventResponse](t, router, "/api/v1/audit-events", dto.AuditEventCreateRequest{ ID: "audit-1", @@ -361,7 +372,13 @@ func TestCoreAPIServerRuntimeDistributionAndJobWorkflows(t *testing.T) { t.Fatalf("unexpected log backfill job: %+v", backfill) } liveLogs := getJSONWithAuth[dto.LogStreamListResponse](t, router, "/api/v1/server-instances/"+serverID+"/logs/live", adminSession) - if liveLogs.Count != 1 || liveLogs.Items[0].StreamKey != "stdout" { + foundFileLog := false + for _, stream := range liveLogs.Items { + if stream.Source == domain.LogStreamSourceFile && stream.StreamKey == "latest-log" { + foundFileLog = true + } + } + if !foundFileLog { t.Fatalf("unexpected live logs: %+v", liveLogs) } @@ -1503,7 +1520,7 @@ func TestRuntimeBindingAPIIsAuthorizedValidatedAndRedacted(t *testing.T) { assertErrorResponse(t, undeclared, http.StatusBadRequest, errorCodeValidation) created := postOKJSONWithAuth[dto.ServerLifecycleResponse](t, router, "/api/v1/server-instances/workflows/create", dto.ServerLifecycleCreateRequest{ID: "runtime-create-complete", PluginID: registration.Manifest.ID, RunEndpointID: "run-local", Name: "Runtime Create Complete", IdempotencyKey: "runtime-create-complete", ProfileKey: "local", Bindings: map[string]string{"server-root": "runtime.server-root", "rcon.password": "secret://runtime-create-complete/rcon"}}, adminSession) - if created.Job.TargetKey != "local" { + if created.Job.TargetKey != "actions/install.json" { t.Fatalf("create workflow did not dispatch selected profile: %+v", created.Job) } incompleteCreate := requestJSONWithAuth(t, router, http.MethodPost, "/api/v1/server-instances/workflows/create", dto.ServerLifecycleCreateRequest{ID: "runtime-create-incomplete", PluginID: registration.Manifest.ID, RunEndpointID: "run-local", Name: "Runtime Create Incomplete", IdempotencyKey: "runtime-create-incomplete", ProfileKey: "local", Bindings: map[string]string{"server-root": "runtime.server-root"}}, adminSession) @@ -1858,6 +1875,7 @@ func createRuntimeAPIFixtures(t *testing.T, router http.Handler, adminSession st } pluginRequest.RuntimeProfiles.DependencyProbes = []dto.RuntimeDependencyProbeBody{{Key: "java-runtime", Kind: "command.version", TargetKey: "java", Platforms: []string{"linux"}}} pluginRequest.RuntimeProfiles.InstallPlans = []dto.RuntimeInstallPlanBody{{Key: "java-install", Title: "Install Java", Platforms: []string{"linux"}, Steps: []dto.RuntimeInstallStepBody{{Type: "package", TargetKey: "java", PackageManager: "apt", PackageName: "openjdk-21-jre"}}}} + pluginRequest.RuntimeProfiles.LogSources = []dto.RuntimeLogSourceBody{{Key: "latest", Kind: "file.tail", TargetKey: "logs/latest", StreamKey: "latest-log", CursorKind: "offset", RetentionDays: 30}} pluginRequest.RuntimeProfiles.ClientManagers = []dto.RuntimeClientManagerProfileBody{{Key: "scum-client-manager", DisplayName: "SCUM Client Manager", Version: "1.0.0", Repository: dto.RuntimeRepositoryBody{URL: "https://github.com/F88888/scum_client.git", RevisionPolicy: "branch", Branch: "main"}, SupportedTargets: []dto.RuntimeTargetBody{{OS: "windows", Arch: "amd64"}, {OS: "linux", Arch: "amd64"}}, Build: dto.RuntimeBuildBody{System: "go", EntryRef: "main.go"}, OutputArtifacts: []string{"scum_client.exe"}, Deployment: dto.RuntimeClientManagerDeploymentBody{Mode: "run-supervised", ExecutableRef: "scum_client.exe", RequiredRunCapabilities: []string{domain.JobCapabilityClientManagerDeploy, domain.JobCapabilityClientManagerControl, domain.JobCapabilityClientManagerUpdate, domain.JobCapabilityClientManagerRollback, domain.JobCapabilityClientManagerUninstall}}, Lifecycle: dto.RuntimeClientManagerLifecycleBody{Actions: []string{"start", "stop", "restart", "status", "update", "rollback", "uninstall"}, StartupTimeoutSeconds: 60, StopTimeoutSeconds: 30}, Health: dto.RuntimeClientManagerHealthBody{Mode: "component-heartbeat", IntervalSeconds: 15, DegradedAfterSeconds: 45, OfflineAfterSeconds: 120, RequiredCapabilities: []string{"component.register", "component.heartbeat", "component.health"}}, Compatibility: dto.RuntimeClientManagerCompatibilityBody{MinimumVersion: "1.0.0"}, UpdatePolicy: dto.RuntimeClientManagerUpdatePolicyBody{Strategy: "manual-staged", RequireApproval: true, HealthConfirmationSeconds: 60, RetainPrevious: true}}} postJSON[dto.GamePluginResponse](t, router, "/api/v1/game-plugins", pluginRequest) @@ -1886,7 +1904,7 @@ func createRuntimeAPIFixtures(t *testing.T, router http.Handler, adminSession st Name: "Runtime API Server", State: domain.ServerInstanceStateReady, }, adminSession) - putJSONWithAuth[dto.RuntimeBindingResponse](t, router, "/api/v1/server-instances/"+server.ID+"/runtime-binding", dto.RuntimeBindingUpdateRequest{ProfileKey: "local", Bindings: map[string]string{}}, adminSession) + putJSONWithAuth[dto.RuntimeBindingResponse](t, router, "/api/v1/server-instances/"+server.ID+"/runtime-binding", dto.RuntimeBindingUpdateRequest{ProfileKey: "local", Bindings: map[string]string{"logs/latest": "runtime.logs.latest"}}, adminSession) postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{ ID: "log-runtime-api", ServerInstanceID: server.ID, diff --git a/platform/api/routes.md b/platform/api/routes.md index 1bac050..cb64cc8 100644 --- a/platform/api/routes.md +++ b/platform/api/routes.md @@ -14,7 +14,7 @@ All routes use JSON request and response bodies. Collection routes support `GET` | Plugin marketplace | `GET /api/v1/plugin-marketplace/plugins` | `GET /api/v1/plugin-marketplace/plugins/{id}`, `POST /api/v1/plugin-marketplace/plugins/{id}/state` | `MarketplacePluginResponse`, `MarketplacePluginListResponse`, `MarketplacePluginStateRequest` | | Plugin bridge | `POST /api/v1/plugin-bridge/authorize`, `POST /api/v1/plugin-bridge/execute` | n/a | `PluginBridgeAuthorizeRequest`, `PluginBridgeAuthorizeResponse`, `PluginBridgeExecuteRequest`, `PluginBridgeExecuteResponse` | | Server instances | `GET /api/v1/server-instances`, `POST /api/v1/server-instances` | `GET /api/v1/server-instances/{id}`, `PUT /api/v1/server-instances/{id}`, `DELETE /api/v1/server-instances/{id}` | `ServerInstanceCreateRequest`, `ServerInstanceUpdateRequest`, `ServerInstanceResponse`, `ServerInstanceListResponse` | -| Server runtime distribution | n/a | `GET /api/v1/server-instances/{id}/runtime/actions`, `POST /api/v1/server-instances/{id}/run/generate`, `POST /api/v1/server-instances/{id}/run/download`, `POST /api/v1/server-instances/{id}/run/key/reset`, `POST /api/v1/server-instances/{id}/run/update`, `GET /api/v1/server-instances/{id}/run/update`, `POST /api/v1/server-instances/{id}/client-managers/generate`, `POST /api/v1/server-instances/{id}/client-managers/download`, `POST /api/v1/server-instances/{id}/client-managers/key/reset`, `GET /api/v1/server-instances/{id}/dependencies`, `POST /api/v1/server-instances/{id}/dependencies/check`, `POST /api/v1/server-instances/{id}/dependencies/install`, `GET /api/v1/server-instances/{id}/logs/live`, `POST /api/v1/server-instances/{id}/logs/backfill` | `ServerRuntimeActionsResponse`, `RunDistributionGenerateRequest`, `RunDistributionResponse`, `RunUpdateRequest`, `RunUpdateJobResponse`/`RunUpdateJobListResponse`, `ClientManagerBuildRequest`, `ClientManagerDistributionResponse`, `ClientManagerDownloadRequest`, `ComponentKeyResetRequest`, `ComponentKeyResponse`, `DependencyCatalogResponse`, `DependencyJobRequest`, `LogBackfillRequest` | +| Server runtime distribution | n/a | `GET /api/v1/server-instances/{id}/runtime/actions`, `POST /api/v1/server-instances/{id}/run/generate`, `POST /api/v1/server-instances/{id}/run/download`, `POST /api/v1/server-instances/{id}/run/key/reset`, `POST /api/v1/server-instances/{id}/run/update`, `GET /api/v1/server-instances/{id}/run/update`, `POST /api/v1/server-instances/{id}/client-managers/generate`, `POST /api/v1/server-instances/{id}/client-managers/download`, `POST /api/v1/server-instances/{id}/client-managers/key/reset`, `GET /api/v1/server-instances/{id}/dependencies`, `POST /api/v1/server-instances/{id}/dependencies/check`, `POST /api/v1/server-instances/{id}/dependencies/install`, `GET /api/v1/server-instances/{id}/logs/live`, `GET /api/v1/server-instances/{id}/logs/events`, `POST /api/v1/server-instances/{id}/logs/backfill` | `ServerRuntimeActionsResponse`, `RunDistributionGenerateRequest`, `RunDistributionResponse`, `RunUpdateRequest`, `RunUpdateJobResponse`/`RunUpdateJobListResponse`, `ClientManagerBuildRequest`, `ClientManagerDistributionResponse`, `ClientManagerDownloadRequest`, `ComponentKeyResetRequest`, `ComponentKeyResponse`, `DependencyCatalogResponse`, `DependencyJobRequest`, `LogBackfillRequest`, `LogStreamEventResponse` | | Metrics | `GET /api/v1/metrics/platform`, `GET /api/v1/metrics/server-instances` | n/a | `PlatformResourceUsageResponse`, `ServerMetricsResponse`, `ServerMetricsListResponse` | | Server config | n/a | `GET /api/v1/server-instances/{id}/config`, `POST /api/v1/server-instances/{id}/config/diff`, `POST /api/v1/server-instances/{id}/config/approve` | `ServerConfigResponse`, `ServerConfigDiffPreviewRequest`, `ServerConfigDiffPreviewResponse`, `ServerConfigWriteApprovalRequest`, `ServerConfigWriteDispatchResponse` | | File operations | `POST /api/v1/file-operations/dispatch` | n/a | `FileOperationDispatchRequest`, `FileOperationDispatchResponse` | @@ -146,7 +146,7 @@ Lifecycle workflow responses include accepted status, action, bounded server ins ## Implemented Runtime Distribution And Client Manager Actions - `GET /api/v1/server-instances/{id}/runtime-binding`: returns the visible server's selected profile and redacted logical binding readiness. Values are represented only by configured/secret-backed flags. -- `PUT /api/v1/server-instances/{id}/runtime-binding`: lets the server owner or a platform administrator select a declared profile and patch safe logical refs. Undeclared keys, unsafe paths/sockets/credentials, and changes to an existing active binding are rejected. +- `PUT /api/v1/server-instances/{id}/runtime-binding`: lets the server owner or a platform administrator select a declared profile and patch safe logical refs for non-deleted servers. Undeclared keys, unsafe paths/sockets/credentials, and plaintext secrets are rejected. - `GET /api/v1/server-instances/{id}/runtime/actions`: returns the current user-visible runtime action matrix for the server, including run endpoint status, action availability, and safe unavailable reasons. - `POST /api/v1/server-instances/{id}/run/generate`: accepts `RunDistributionGenerateRequest`, queues a platform-owned Docker build, creates or reuses the server's current encrypted run key, writes that key into the secret-bearing generated package config, publishes an artifact, and returns `RunDistributionResponse` with checksum, key generation, artifact ID, build job ID, and redacted secret ref only. - `POST /api/v1/server-instances/{id}/run/download`: opens the latest available run package through `ArtifactDownloadReferenceResponse` after server-scoped authorization. @@ -160,6 +160,7 @@ Lifecycle workflow responses include accepted status, action, bounded server ins - `POST /api/v1/server-instances/{id}/dependencies/check`: accepts `DependencyJobRequest` and queues a `dependencies.check` run job for a declared logical probe key. - `POST /api/v1/server-instances/{id}/dependencies/install`: accepts `DependencyJobRequest` with an install plan key and the exact catalog `planDigest`; stale/missing digests are denied before job creation. - `GET /api/v1/server-instances/{id}/logs/live`: returns safe live log stream metadata for the selected server using `LogStreamListResponse`. +- `GET /api/v1/server-instances/{id}/logs/events`: streams selected server log metadata and entries as `text/event-stream`; the optional `historyLimit` query replays recent stored entries before live push events. - `POST /api/v1/server-instances/{id}/logs/backfill`: accepts `LogBackfillRequest`, queues a `logs.backfill` job with source key, checkpoint ref, limit, and idempotency metadata, and keeps log bodies out of job results. Runtime distribution and client-manager APIs require the current bearer session, server visibility, plugin-declared permissions, complete runtime bindings where required, and platform-builder readiness. Run-side lifecycle commands separately require run endpoint capability support. Responses and audit summaries expose artifact IDs, job IDs, checksums, key generations, fingerprints, status, and redacted `secret://runtime-keys/.../current` refs only. They do not expose raw run keys, client-manager keys, FTP passwords, database DSNs, RCON passwords, host paths, direct sockets, run endpoint private addresses, build workspace paths, or large inline logs. @@ -196,6 +197,7 @@ Job ack/progress/result/cancel/reconcile calls remain lightweight and independen - `POST /api/v1/run/logs/batches`: accept `LogBatchIngestRequest`, validate run session and stream metadata, store contiguous entries, update `LogStream.LatestSeq`, and return `LogBatchIngestResponse` with the acknowledged range. - `POST /api/v1/log-streams/query`: accept `LogStreamCursorRequest` and return `LogStreamCursorResponse` with bounded ordered entries after a cursor. +- `GET /api/v1/server-instances/{id}/logs/events`: authorize the browser session for the server, replay bounded recent entries, and push newly ingested log entries over SSE without polling log stream queries. Log ingest actions carry durable log metadata and bounded entries only: run endpoint ID, session token, stream identity, source, sequence range, compression metadata, checksum, entries, and cursor limits. They do not carry artifact chunks, host paths, raw credentials, direct sockets, or unbounded inline data. Log ingest is durable and independently retried. Artifact/file transfer backlog must not prevent log acknowledgement, duplicate acknowledgement, cursor state updates, or spool cleanup. diff --git a/platform/domain/job_channel.go b/platform/domain/job_channel.go index ad2594e..d1a0abe 100644 --- a/platform/domain/job_channel.go +++ b/platform/domain/job_channel.go @@ -324,6 +324,8 @@ type RunJobReconcileResult struct { func CopyRunJobAssignment(assignment RunJobAssignment) RunJobAssignment { assignment.ExecutionInput.Inputs = CopyStringMap(assignment.ExecutionInput.Inputs) + assignment.ExecutionInput.LogSource = CopyRuntimeLogSourcePtr(assignment.ExecutionInput.LogSource) + assignment.ExecutionInput.LogSources = CopyRuntimeLogSources(assignment.ExecutionInput.LogSources) assignment.ExecutionInput.DLLExtensions = append([]RuntimeDLLExtensionPlan(nil), assignment.ExecutionInput.DLLExtensions...) assignment.ExecutionInput.SourceRCON = CopyRuntimeSourceRCONPlan(assignment.ExecutionInput.SourceRCON) return assignment diff --git a/platform/domain/log_ingest.go b/platform/domain/log_ingest.go index fbbeff6..660b351 100644 --- a/platform/domain/log_ingest.go +++ b/platform/domain/log_ingest.go @@ -48,6 +48,13 @@ type LogStreamCursorResult struct { LatestSeq uint64 } +type LogStreamEvent struct { + ServerInstanceID string + Stream LogStream + Entry LogEntry + LatestSeq uint64 +} + type LogBatchRecord struct { Checksum string FirstSeq uint64 @@ -87,6 +94,12 @@ func CopyLogStreamCursorResult(result LogStreamCursorResult) LogStreamCursorResu return result } +func CopyLogStreamEvent(event LogStreamEvent) LogStreamEvent { + event.Stream = CopyLogStream(event.Stream) + event.Entry = CopyLogEntry(event.Entry) + return event +} + func CopyLogBatchRecord(record LogBatchRecord) LogBatchRecord { record.Entries = CopyLogEntries(record.Entries) return record diff --git a/platform/domain/resources.go b/platform/domain/resources.go index 88dfcd1..e49b29e 100644 --- a/platform/domain/resources.go +++ b/platform/domain/resources.go @@ -1109,6 +1109,8 @@ type JobExecutionInput struct { LifecycleOperation string TargetVersion string Inputs map[string]string + LogSource *RuntimeLogSource + LogSources []RuntimeLogSource DLLExtensions []RuntimeDLLExtensionPlan SourceRCON *RuntimeSourceRCONPlan Deployment *ServerDeploymentDefinition @@ -1988,6 +1990,8 @@ func CopyRunEndpoint(endpoint RunEndpoint) RunEndpoint { func CopyJob(job Job) Job { job.ExecutionInput.Inputs = CopyStringMap(job.ExecutionInput.Inputs) + job.ExecutionInput.LogSource = CopyRuntimeLogSourcePtr(job.ExecutionInput.LogSource) + job.ExecutionInput.LogSources = CopyRuntimeLogSources(job.ExecutionInput.LogSources) job.ExecutionInput.DLLExtensions = append([]RuntimeDLLExtensionPlan(nil), job.ExecutionInput.DLLExtensions...) job.ExecutionInput.SourceRCON = CopyRuntimeSourceRCONPlan(job.ExecutionInput.SourceRCON) job.ExecutionInput.ServerDeploymentPlan = CopyServerDeploymentPlan(job.ExecutionInput.ServerDeploymentPlan) @@ -2000,6 +2004,21 @@ func CopyJob(job Job) Job { return job } +func CopyRuntimeLogSourcePtr(source *RuntimeLogSource) *RuntimeLogSource { + if source == nil { + return nil + } + copy := *source + return © +} + +func CopyRuntimeLogSources(sources []RuntimeLogSource) []RuntimeLogSource { + if sources == nil { + return nil + } + return append([]RuntimeLogSource(nil), sources...) +} + func CopyRuntimeSourceRCONPlan(plan *RuntimeSourceRCONPlan) *RuntimeSourceRCONPlan { if plan == nil { return nil diff --git a/platform/dto/job_channel.go b/platform/dto/job_channel.go index ede5c88..20f5a77 100644 --- a/platform/dto/job_channel.go +++ b/platform/dto/job_channel.go @@ -105,6 +105,8 @@ type RunJobExecutionInputBody struct { LifecycleOperation string `json:"lifecycleOperation,omitempty"` TargetVersion string `json:"targetVersion,omitempty"` Inputs map[string]string `json:"inputs,omitempty"` + LogSource *RuntimeLogSourceBody `json:"logSource,omitempty"` + LogSources []RuntimeLogSourceBody `json:"logSources,omitempty"` DLLExtensions []RuntimeDLLExtensionPlanBody `json:"dllExtensions,omitempty"` SourceRCON *RuntimeSourceRCONPlanBody `json:"sourceRcon,omitempty"` Deployment *ServerDeploymentExecutionBody `json:"deployment,omitempty"` @@ -661,7 +663,7 @@ func RunJobAssignmentFromDomain(assignment domain.RunJobAssignment) RunJobAssign State: assignment.State, Progress: progressReportFromDomain(assignment.Progress), ResultRef: assignment.ResultRef, - ExecutionInput: RunJobExecutionInputBody{WorkspaceScope: assignment.ExecutionInput.WorkspaceScope, Content: assignment.ExecutionInput.Content, ExpectedVersion: assignment.ExecutionInput.ExpectedVersion, ExpectedChecksum: assignment.ExecutionInput.ExpectedChecksum, MaxReadBytes: assignment.ExecutionInput.MaxReadBytes, RemoteAdapterKey: assignment.ExecutionInput.RemoteAdapterKey, RemoteAdapterKind: assignment.ExecutionInput.RemoteAdapterKind, TimeoutSeconds: assignment.ExecutionInput.TimeoutSeconds, PluginID: assignment.ExecutionInput.PluginID, LifecycleOperation: assignment.ExecutionInput.LifecycleOperation, TargetVersion: assignment.ExecutionInput.TargetVersion, Inputs: domain.CopyStringMap(assignment.ExecutionInput.Inputs), DLLExtensions: dllExtensionPlansFromDomain(assignment.ExecutionInput.DLLExtensions), SourceRCON: runtimeSourceRCONPlanFromDomain(assignment.ExecutionInput.SourceRCON), Deployment: deploymentExecutionFromDomain(assignment.ExecutionInput.Deployment), ServerDeploymentPlan: serverDeploymentPlanFromDomain(assignment.ExecutionInput.ServerDeploymentPlan)}, + ExecutionInput: RunJobExecutionInputBody{WorkspaceScope: assignment.ExecutionInput.WorkspaceScope, Content: assignment.ExecutionInput.Content, ExpectedVersion: assignment.ExecutionInput.ExpectedVersion, ExpectedChecksum: assignment.ExecutionInput.ExpectedChecksum, MaxReadBytes: assignment.ExecutionInput.MaxReadBytes, RemoteAdapterKey: assignment.ExecutionInput.RemoteAdapterKey, RemoteAdapterKind: assignment.ExecutionInput.RemoteAdapterKind, TimeoutSeconds: assignment.ExecutionInput.TimeoutSeconds, PluginID: assignment.ExecutionInput.PluginID, LifecycleOperation: assignment.ExecutionInput.LifecycleOperation, TargetVersion: assignment.ExecutionInput.TargetVersion, Inputs: domain.CopyStringMap(assignment.ExecutionInput.Inputs), LogSource: runtimeLogSourceFromDomain(assignment.ExecutionInput.LogSource), LogSources: runtimeLogSourcesFromDomain(assignment.ExecutionInput.LogSources), DLLExtensions: dllExtensionPlansFromDomain(assignment.ExecutionInput.DLLExtensions), SourceRCON: runtimeSourceRCONPlanFromDomain(assignment.ExecutionInput.SourceRCON), Deployment: deploymentExecutionFromDomain(assignment.ExecutionInput.Deployment), ServerDeploymentPlan: serverDeploymentPlanFromDomain(assignment.ExecutionInput.ServerDeploymentPlan)}, LeaseToken: assignment.LeaseToken, Attempt: assignment.Attempt, FencingToken: assignment.FencingToken, @@ -736,6 +738,24 @@ func runtimeSourceRCONPlanFromDomain(plan *domain.RuntimeSourceRCONPlan) *Runtim return &RuntimeSourceRCONPlanBody{Protocol: plan.Protocol, ExtensionKey: plan.ExtensionKey, ModKey: plan.ModKey, ConfigRef: plan.ConfigRef, DeploymentStateRef: plan.DeploymentStateRef, Port: plan.Port} } +func runtimeLogSourceFromDomain(source *domain.RuntimeLogSource) *RuntimeLogSourceBody { + if source == nil { + return nil + } + return &RuntimeLogSourceBody{Key: source.Key, Kind: source.Kind, TargetKey: source.TargetKey, StreamKey: source.StreamKey, CursorKind: source.CursorKind, RetentionDays: source.RetentionDays} +} + +func runtimeLogSourcesFromDomain(sources []domain.RuntimeLogSource) []RuntimeLogSourceBody { + if len(sources) == 0 { + return nil + } + out := make([]RuntimeLogSourceBody, 0, len(sources)) + for _, source := range sources { + out = append(out, RuntimeLogSourceBody{Key: source.Key, Kind: source.Kind, TargetKey: source.TargetKey, StreamKey: source.StreamKey, CursorKind: source.CursorKind, RetentionDays: source.RetentionDays}) + } + return out +} + func progressReportToDomain(progress JobProgressBody) domain.RunJobProgressReport { return domain.RunJobProgressReport{ Percent: progress.Percent, diff --git a/platform/dto/log_ingest.go b/platform/dto/log_ingest.go index 2712843..87fa908 100644 --- a/platform/dto/log_ingest.go +++ b/platform/dto/log_ingest.go @@ -53,6 +53,21 @@ type LogStreamCursorResponse struct { LatestSeq uint64 `json:"latestSeq"` } +type LogStreamEventResponse struct { + ServerInstanceID string `json:"serverInstanceId"` + StreamID string `json:"streamId"` + Source domain.LogStreamSource `json:"source"` + StreamKey string `json:"streamKey"` + LatestSeq uint64 `json:"latestSeq"` + Entry LogEntryBody `json:"entry"` +} + +type LogStreamEventsReadyResponse struct { + ServerInstanceID string `json:"serverInstanceId"` + StreamCount int `json:"streamCount"` + ServerTime time.Time `json:"serverTime"` +} + func (request LogBatchIngestRequest) ToDomain() domain.LogBatchIngest { return domain.LogBatchIngest{ RunEndpointID: request.RunEndpointID, @@ -99,6 +114,18 @@ func LogStreamCursorFromDomain(result domain.LogStreamCursorResult) LogStreamCur } } +func LogStreamEventFromDomain(event domain.LogStreamEvent) LogStreamEventResponse { + event = domain.CopyLogStreamEvent(event) + return LogStreamEventResponse{ + ServerInstanceID: event.ServerInstanceID, + StreamID: event.Stream.ID, + Source: event.Stream.Source, + StreamKey: event.Stream.StreamKey, + LatestSeq: event.LatestSeq, + Entry: logEntryFromDomain(event.Entry), + } +} + func logEntriesToDomain(entries []LogEntryBody) []domain.LogEntry { if entries == nil { return nil @@ -117,20 +144,24 @@ func logEntriesToDomain(entries []LogEntryBody) []domain.LogEntry { return out } +func logEntryFromDomain(entry domain.LogEntry) LogEntryBody { + return LogEntryBody{ + Seq: entry.Seq, + Timestamp: entry.Timestamp, + Level: entry.Level, + Line: entry.Line, + Fields: copyStringMap(entry.Fields), + Redacted: entry.Redacted, + } +} + func logEntriesFromDomain(entries []domain.LogEntry) []LogEntryBody { if entries == nil { return nil } out := make([]LogEntryBody, len(entries)) for i, entry := range entries { - out[i] = LogEntryBody{ - Seq: entry.Seq, - Timestamp: entry.Timestamp, - Level: entry.Level, - Line: entry.Line, - Fields: copyStringMap(entry.Fields), - Redacted: entry.Redacted, - } + out[i] = logEntryFromDomain(entry) } return out } diff --git a/platform/protocol/run-contracts.md b/platform/protocol/run-contracts.md index f4c3b5a..bcb5d11 100644 --- a/platform/protocol/run-contracts.md +++ b/platform/protocol/run-contracts.md @@ -63,6 +63,7 @@ Implemented HTTP JSON routes: - `POST /api/v1/run/logs/batches` - `POST /api/v1/log-streams/query` +- `GET /api/v1/server-instances/{id}/logs/events` Named log DTOs: @@ -71,8 +72,9 @@ Named log DTOs: - `LogEntry` - `LogStreamCursorRequest` - `LogStreamCursorResponse` +- `LogStreamEventResponse` -Log ingest supports bounded batches, sequence ranges, checksum validation, retry-safe duplicate acknowledgement, latest sequence tracking, and cursor query. Log payloads must not carry artifact chunks, host paths, raw credentials, direct sockets, or unbounded inline data. +Log ingest supports bounded batches, sequence ranges, checksum validation, retry-safe duplicate acknowledgement, latest sequence tracking, cursor query, and browser SSE fan-out from already-ingested platform logs. Log payloads must not carry artifact chunks, host paths, raw credentials, direct sockets, or unbounded inline data. Log ingest is durable and independently retried. Artifact/file transfer backlog must not prevent log batch acknowledgement, duplicate acknowledgement, cursor state updates, or spool cleanup. diff --git a/platform/service/distributions.go b/platform/service/distributions.go index 6b3eac0..86a38eb 100644 --- a/platform/service/distributions.go +++ b/platform/service/distributions.go @@ -745,6 +745,11 @@ func (svc *CoreService) QueueLogBackfillForSession(sessionID string, request dom _ = svc.recordAuditEvent(user.ID, "logs.backfill.denied", "server-instance", instance.ID, domain.AuditResultDenied, "log backfill denied: plugin capability is not declared") return domain.Job{}, ErrForbidden } + source, err := declaredFileLogSource(plugin, request.SourceKey) + if err != nil { + _ = svc.recordAuditEvent(user.ID, "logs.backfill.denied", "server-instance", instance.ID, domain.AuditResultDenied, "log backfill denied: log source is not declared") + return domain.Job{}, err + } if err := svc.requireCompleteRuntimeBindings(user.ID, instance.ID, "logs.backfill.denied"); err != nil { return domain.Job{}, err } @@ -753,10 +758,11 @@ func (svc *CoreService) QueueLogBackfillForSession(sessionID string, request dom ServerInstanceID: instance.ID, RunEndpointID: instance.RunEndpointID, Capability: domain.JobCapabilityLogsBackfill, - TargetKey: "logs/" + request.SourceKey, + TargetKey: "logs/" + source.Key, InputRef: request.CheckpointRef, IdempotencyKey: request.IdempotencyKey, Progress: domain.JobProgress{Percent: 0, Message: "historical log backfill queued"}, + ExecutionInput: domain.JobExecutionInput{LogSource: &source}, }) if err != nil { _ = svc.recordAuditEvent(user.ID, "logs.backfill.denied", "server-instance", instance.ID, domain.AuditResultDenied, "log backfill denied: endpoint unsupported or offline") @@ -768,6 +774,20 @@ func (svc *CoreService) QueueLogBackfillForSession(sessionID string, request dom return domain.CopyJob(job), nil } +func declaredFileLogSource(plugin domain.GamePlugin, sourceKey string) (domain.RuntimeLogSource, error) { + for _, source := range plugin.RuntimeProfiles.LogSources { + if source.Key != sourceKey { + continue + } + if source.Kind != "file.tail" || strings.TrimSpace(source.TargetKey) == "" || strings.TrimSpace(source.StreamKey) == "" { + return domain.RuntimeLogSource{}, validationError("log source must be a file.tail source with target and stream keys") + } + copy := source + return copy, nil + } + return domain.RuntimeLogSource{}, validationError("log source is not declared by plugin") +} + func (svc *CoreService) validateDistributionPluginPermission(actorID string, plugin domain.GamePlugin, serverInstanceID string, permission string, deniedAction string) error { if plugin.Status != domain.GamePluginStatusInstalled { _ = svc.recordAuditEvent(actorID, deniedAction, "server-instance", serverInstanceID, domain.AuditResultDenied, "distribution operation denied: plugin is not installed") diff --git a/platform/service/distributions_test.go b/platform/service/distributions_test.go index bce0a42..a212592 100644 --- a/platform/service/distributions_test.go +++ b/platform/service/distributions_test.go @@ -627,6 +627,7 @@ func newDistributionTestFixture(t *testing.T) (*CoreService, string, domain.Serv ) plugin.RuntimeProfiles.DependencyProbes = []domain.RuntimeDependencyProbe{{Key: "java-runtime", Kind: "command.version", TargetKey: "java", Platforms: []string{"linux"}}} plugin.RuntimeProfiles.InstallPlans = []domain.RuntimeInstallPlan{{Key: "java-install", Title: "Install Java", Platforms: []string{"linux"}, Steps: []domain.RuntimeInstallStep{{Type: "package", TargetKey: "java", PackageManager: "apt", PackageName: "openjdk-21-jre"}}}} + plugin.RuntimeProfiles.LogSources = []domain.RuntimeLogSource{{Key: "latest", Kind: "file.tail", TargetKey: "logs/latest", StreamKey: "latest-log", CursorKind: "offset", RetentionDays: 30}} plugin.RuntimeProfiles.ClientManagers = []domain.RuntimeClientManagerProfile{{Key: "scum-client-manager", DisplayName: "SCUM Client Manager", Version: "1.0.0", RepositoryURL: "https://github.com/F88888/scum_client.git", RevisionPolicy: "branch", Branch: "main", SupportedTargets: []domain.RuntimeTarget{{OS: "windows", Arch: "amd64"}, {OS: "linux", Arch: "amd64"}}, BuildSystem: "go", EntryRef: "main.go", OutputArtifacts: []string{"scum_client.exe"}, Deployment: domain.RuntimeClientManagerDeployment{Mode: "run-supervised", ExecutableRef: "scum_client.exe", RequiredRunCapabilities: []string{domain.JobCapabilityClientManagerDeploy, domain.JobCapabilityClientManagerControl, domain.JobCapabilityClientManagerUpdate, domain.JobCapabilityClientManagerRollback, domain.JobCapabilityClientManagerUninstall}}, Lifecycle: domain.RuntimeClientManagerLifecycle{Actions: []string{"start", "stop", "restart", "status", "update", "rollback", "uninstall"}, StartupTimeoutSeconds: 60, StopTimeoutSeconds: 30}, Health: domain.RuntimeClientManagerHealth{Mode: "component-heartbeat", IntervalSeconds: 15, DegradedAfterSeconds: 45, OfflineAfterSeconds: 120, RequiredCapabilities: []string{"component.register", "component.heartbeat", "component.health"}}, Compatibility: domain.RuntimeClientManagerCompatibility{MinimumVersion: "1.0.0"}, UpdatePolicy: domain.RuntimeClientManagerUpdatePolicy{Strategy: "manual-staged", RequireApproval: true, HealthConfirmationSeconds: 60, RetainPrevious: true}}} if err := svc.store.GamePlugins().Update(plugin); err != nil { t.Fatalf("update plugin fixture: %v", err) @@ -664,10 +665,35 @@ func newDistributionTestFixture(t *testing.T) (*CoreService, string, domain.Serv if err != nil { t.Fatalf("create distribution server: %v", err) } - createCompleteRuntimeBinding(t, svc, instance, "local") + binding, err := svc.buildRuntimeBinding(instance, plugin, domain.RuntimeBindingUpdate{ProfileKey: "local", Bindings: map[string]string{"logs/latest": "runtime.logs.latest"}}, true) + if err != nil { + t.Fatalf("build complete runtime binding: %v", err) + } + if err := svc.store.RuntimeBindings().Create(binding); err != nil { + t.Fatalf("create runtime binding: %v", err) + } return svc, session, instance } +func TestQueueLogBackfillFreezesDeclaredFileLogSource(t *testing.T) { + svc, session, instance := newDistributionTestFixture(t) + + job, err := svc.QueueLogBackfillForSession(session, domain.LogBackfillRequest{ServerInstanceID: instance.ID, SourceKey: "latest", CheckpointRef: "artifact://logs/checkpoint/1", IdempotencyKey: "log-source-freeze"}) + if err != nil { + t.Fatalf("queue log backfill: %v", err) + } + if job.ExecutionInput.LogSource == nil || job.ExecutionInput.LogSource.Key != "latest" || job.ExecutionInput.LogSource.StreamKey != "latest-log" || job.ExecutionInput.LogSource.TargetKey != "logs/latest" { + t.Fatalf("expected frozen declared log source, got %+v", job.ExecutionInput.LogSource) + } + stream, err := svc.GetLogStream(jobLogStreamID(job.ID, "latest-log")) + if err != nil || stream.Source != domain.LogStreamSourceFile || stream.StreamKey != "latest-log" { + t.Fatalf("expected file log stream for backfill job, stream=%+v err=%v", stream, err) + } + if _, err := svc.QueueLogBackfillForSession(session, domain.LogBackfillRequest{ServerInstanceID: instance.ID, SourceKey: "missing", IdempotencyKey: "log-source-missing"}); err == nil || !strings.Contains(err.Error(), "not declared") { + t.Fatalf("expected undeclared source rejection, got %v", err) + } +} + func readGeneratedPackageConfig(t *testing.T, svc *CoreService, session string, artifactID string) generatedPackageConfig { t.Helper() _ = session diff --git a/platform/service/job_channel.go b/platform/service/job_channel.go index 6048d6f..7dc7276 100644 --- a/platform/service/job_channel.go +++ b/platform/service/job_channel.go @@ -639,7 +639,7 @@ func assignmentFromJob(job domain.Job, leaseToken string) domain.RunJobAssignmen State: job.State, Progress: domain.RunJobProgressReport{Percent: job.Progress.Percent, Phase: job.Progress.Phase, Message: job.Progress.Message}, ResultRef: job.ResultRef, - ExecutionInput: domain.JobExecutionInput{WorkspaceScope: job.ExecutionInput.WorkspaceScope, Content: job.ExecutionInput.Content, ExpectedVersion: job.ExecutionInput.ExpectedVersion, ExpectedChecksum: job.ExecutionInput.ExpectedChecksum, MaxReadBytes: job.ExecutionInput.MaxReadBytes, RemoteAdapterKey: job.ExecutionInput.RemoteAdapterKey, RemoteAdapterKind: job.ExecutionInput.RemoteAdapterKind, TimeoutSeconds: job.ExecutionInput.TimeoutSeconds, PluginID: job.ExecutionInput.PluginID, LifecycleOperation: job.ExecutionInput.LifecycleOperation, TargetVersion: job.ExecutionInput.TargetVersion, Inputs: domain.CopyStringMap(job.ExecutionInput.Inputs), DLLExtensions: append([]domain.RuntimeDLLExtensionPlan(nil), job.ExecutionInput.DLLExtensions...), SourceRCON: domain.CopyRuntimeSourceRCONPlan(job.ExecutionInput.SourceRCON), Deployment: deploymentPlanForDispatchValue(job.ExecutionInput.Deployment), ServerDeploymentPlan: domain.CopyServerDeploymentPlan(job.ExecutionInput.ServerDeploymentPlan)}, + ExecutionInput: domain.JobExecutionInput{WorkspaceScope: job.ExecutionInput.WorkspaceScope, Content: job.ExecutionInput.Content, ExpectedVersion: job.ExecutionInput.ExpectedVersion, ExpectedChecksum: job.ExecutionInput.ExpectedChecksum, MaxReadBytes: job.ExecutionInput.MaxReadBytes, RemoteAdapterKey: job.ExecutionInput.RemoteAdapterKey, RemoteAdapterKind: job.ExecutionInput.RemoteAdapterKind, TimeoutSeconds: job.ExecutionInput.TimeoutSeconds, PluginID: job.ExecutionInput.PluginID, LifecycleOperation: job.ExecutionInput.LifecycleOperation, TargetVersion: job.ExecutionInput.TargetVersion, Inputs: domain.CopyStringMap(job.ExecutionInput.Inputs), LogSource: domain.CopyRuntimeLogSourcePtr(job.ExecutionInput.LogSource), LogSources: domain.CopyRuntimeLogSources(job.ExecutionInput.LogSources), DLLExtensions: append([]domain.RuntimeDLLExtensionPlan(nil), job.ExecutionInput.DLLExtensions...), SourceRCON: domain.CopyRuntimeSourceRCONPlan(job.ExecutionInput.SourceRCON), Deployment: deploymentPlanForDispatchValue(job.ExecutionInput.Deployment), ServerDeploymentPlan: domain.CopyServerDeploymentPlan(job.ExecutionInput.ServerDeploymentPlan)}, LeaseToken: leaseToken, Attempt: job.Attempt, FencingToken: fencingToken, diff --git a/platform/service/log_events.go b/platform/service/log_events.go new file mode 100644 index 0000000..21f98d2 --- /dev/null +++ b/platform/service/log_events.go @@ -0,0 +1,87 @@ +package service + +import ( + "strings" + + "browser.local/platform/domain" +) + +const logEventSubscriberBuffer = 512 + +type LogEventSubscription struct { + Events <-chan domain.LogStreamEvent + Close func() +} + +type logEventSubscriber struct { + serverInstanceID string + events chan domain.LogStreamEvent +} + +func (svc *CoreService) SubscribeLogEvents(serverInstanceID string) (LogEventSubscription, error) { + serverInstanceID = strings.TrimSpace(serverInstanceID) + if serverInstanceID == "" { + return LogEventSubscription{}, validationError("serverInstanceId is required") + } + if _, err := svc.store.ServerInstances().Get(serverInstanceID); err != nil { + return LogEventSubscription{}, err + } + events := make(chan domain.LogStreamEvent, logEventSubscriberBuffer) + svc.logEventMu.Lock() + svc.logEventSubscriberSeq++ + id := svc.logEventSubscriberSeq + svc.logEventSubscribers[id] = logEventSubscriber{serverInstanceID: serverInstanceID, events: events} + svc.logEventMu.Unlock() + closeOnce := func() { + svc.logEventMu.Lock() + if subscriber, ok := svc.logEventSubscribers[id]; ok { + delete(svc.logEventSubscribers, id) + close(subscriber.events) + } + svc.logEventMu.Unlock() + } + return LogEventSubscription{Events: events, Close: closeOnce}, nil +} + +func (svc *CoreService) SubscribeLogEventsForSession(sessionID string, serverInstanceID string) (LogEventSubscription, error) { + instance, err := svc.GetServerInstanceForSession(sessionID, serverInstanceID) + if err != nil { + return LogEventSubscription{}, err + } + return svc.SubscribeLogEvents(instance.ID) +} + +func (svc *CoreService) publishLogEvents(stream domain.LogStream, entries []domain.LogEntry) { + if len(entries) == 0 { + return + } + events := make([]domain.LogStreamEvent, len(entries)) + for index, entry := range entries { + events[index] = domain.CopyLogStreamEvent(domain.LogStreamEvent{ + ServerInstanceID: stream.ServerInstanceID, + Stream: stream, + Entry: entry, + LatestSeq: stream.LatestSeq, + }) + } + svc.logEventMu.Lock() + for id, subscriber := range svc.logEventSubscribers { + if subscriber.serverInstanceID != stream.ServerInstanceID { + continue + } + dropped := false + for _, event := range events { + select { + case subscriber.events <- event: + default: + delete(svc.logEventSubscribers, id) + close(subscriber.events) + dropped = true + } + if dropped { + break + } + } + } + svc.logEventMu.Unlock() +} diff --git a/platform/service/log_ingest.go b/platform/service/log_ingest.go index 5cafdbf..c62218f 100644 --- a/platform/service/log_ingest.go +++ b/platform/service/log_ingest.go @@ -1,7 +1,12 @@ package service import ( + "errors" + "strings" + "time" + "browser.local/platform/domain" + "browser.local/platform/repo" "browser.local/platform/validator" ) @@ -19,6 +24,11 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB stamp := svc.now() stream, err := svc.store.LogStreams().Get(batch.LogStreamID) + if errors.Is(err, repo.ErrNotFound) { + if repairErr := svc.ensureJobLogStreamForBatch(batch, stamp); repairErr == nil { + stream, err = svc.store.LogStreams().Get(batch.LogStreamID) + } + } if err != nil { return domain.LogBatchIngestResult{}, err } @@ -76,6 +86,7 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB if err := svc.projectGameMapTrajectoryEvents(projectionBatch); err != nil { return domain.LogBatchIngestResult{}, err } + svc.publishLogEvents(stream, storedBatch.Entries) return domain.LogBatchIngestResult{ Accepted: true, LogStreamID: batch.LogStreamID, @@ -86,6 +97,35 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB }, nil } +func (svc *CoreService) ensureJobLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error { + jobID, ok := jobIDFromLogBatch(batch) + if !ok { + return repo.ErrNotFound + } + job, err := svc.store.Jobs().Get(jobID) + if err != nil { + return err + } + if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID { + return validationError("log batch job scope does not match stream") + } + return svc.ensureJobLogStreams(job, stamp) +} + +func jobIDFromLogBatch(batch domain.LogBatchIngest) (string, bool) { + streamKey := strings.TrimSpace(batch.StreamKey) + if streamKey == "" || !strings.HasPrefix(batch.LogStreamID, "job.") { + return "", false + } + suffix := "." + streamKey + body := strings.TrimPrefix(batch.LogStreamID, "job.") + if !strings.HasSuffix(body, suffix) { + return "", false + } + jobID := strings.TrimSuffix(body, suffix) + return jobID, strings.TrimSpace(jobID) != "" +} + func logBatchRecordMatches(record domain.LogBatchRecord, batch domain.LogBatchIngest) bool { if record.Checksum == batch.Checksum { return true diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index 734f6e1..d93f2f0 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -39,6 +39,38 @@ func TestCoreServiceIngestsLogBatchAndQueriesCursor(t *testing.T) { } } +func TestCoreServicePublishesLogEventsForAcceptedBatch(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + createLogStreamFixture(t, svc) + subscription, err := svc.SubscribeLogEvents("server-1") + if err != nil { + t.Fatalf("subscribe log events: %v", err) + } + defer subscription.Close() + + batch := validLogBatch(t, sessionToken, 1, 1) + if _, err := svc.IngestLogBatch(batch); err != nil { + t.Fatalf("ingest log batch: %v", err) + } + select { + case event := <-subscription.Events: + if event.Stream.ID != "log-1" || event.Entry.Seq != 1 || event.LatestSeq != 1 { + t.Fatalf("unexpected log event: %+v", event) + } + case <-time.After(time.Second): + t.Fatal("expected log event after accepted batch") + } + + if _, err := svc.IngestLogBatch(batch); err != nil { + t.Fatalf("ingest duplicate batch: %v", err) + } + select { + case event := <-subscription.Events: + t.Fatalf("duplicate batch should not publish a second event: %+v", event) + default: + } +} + func TestCoreServiceLogBatchDuplicateAck(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) createLogStreamFixture(t, svc) @@ -114,6 +146,93 @@ func TestCoreServiceAcceptsAutoCreatedRunJobLogStreams(t *testing.T) { } } +func TestCoreServiceAcceptsPluginDeclaredProcessLogStreams(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + job, err := svc.CreateJob(domain.Job{ + ID: "job-declared-process-logs", + ServerInstanceID: "server-1", + RunEndpointID: "run-local", + Capability: domain.LifecycleCapabilityStart, + IdempotencyKey: "job-declared-process-logs", + ExecutionInput: domain.JobExecutionInput{WorkspaceScope: "run-local", LifecycleOperation: "start", LogSources: []domain.RuntimeLogSource{ + {Key: "console-out", Kind: "process.stdout", StreamKey: "scum.console.stdout", CursorKind: "sequence", RetentionDays: 30}, + {Key: "console-err", Kind: "process.stderr", StreamKey: "scum.console.stderr", CursorKind: "sequence", RetentionDays: 30}, + }}, + }) + if err != nil { + t.Fatalf("create job: %v", err) + } + streamID := jobLogStreamID(job.ID, "scum.console.stdout") + stream, err := svc.GetLogStream(streamID) + if err != nil { + t.Fatalf("get declared process log stream: %v", err) + } + if stream.StreamKey != "scum.console.stdout" || stream.Source != domain.LogStreamSourceProcess { + t.Fatalf("unexpected declared stream metadata: %+v", stream) + } + entry := domain.LogEntry{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Level: "info", Line: "LogStreaming: Display: server ready"} + ack, err := svc.IngestLogBatch(domain.LogBatchIngest{ + RunEndpointID: "run-local", + SessionToken: sessionToken, + LogStreamID: streamID, + ServerInstanceID: "server-1", + StreamKey: "scum.console.stdout", + Source: domain.LogStreamSourceProcess, + FirstSeq: entry.Seq, + LastSeq: entry.Seq, + Compression: "none", + Checksum: validator.LogLineChecksum(entry.Line), + Entries: []domain.LogEntry{entry}, + }) + if err != nil { + t.Fatalf("ingest declared process log batch: %v", err) + } + if !ack.Accepted || ack.LatestSeq != entry.Seq { + t.Fatalf("unexpected declared stream ack: %+v", ack) + } +} + +func TestCoreServiceRepairsMissingDeclaredProcessLogStreamOnIngest(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + job := domain.Job{ + ID: "job-repaired-process-logs", + ServerInstanceID: "server-1", + RunEndpointID: "run-local", + Capability: domain.LifecycleCapabilityStart, + IdempotencyKey: "job-repaired-process-logs", + ExecutionInput: domain.JobExecutionInput{WorkspaceScope: "run-local", LifecycleOperation: "start", LogSources: []domain.RuntimeLogSource{ + {Key: "console-out", Kind: "process.stdout", StreamKey: "scum.console.stdout", CursorKind: "sequence", RetentionDays: 30}, + }}, + } + if err := svc.store.Jobs().Create(job); err != nil { + t.Fatalf("seed legacy job without streams: %v", err) + } + streamID := jobLogStreamID(job.ID, "scum.console.stdout") + if _, err := svc.GetLogStream(streamID); err == nil { + t.Fatal("expected declared stream to be missing before ingest repair") + } + entry := domain.LogEntry{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Level: "info", Line: "LogStreaming: Display: recovered from spool"} + ack, err := svc.IngestLogBatch(domain.LogBatchIngest{ + RunEndpointID: "run-local", + SessionToken: sessionToken, + LogStreamID: streamID, + ServerInstanceID: "server-1", + StreamKey: "scum.console.stdout", + Source: domain.LogStreamSourceProcess, + FirstSeq: entry.Seq, + LastSeq: entry.Seq, + Compression: "none", + Checksum: validator.LogLineChecksum(entry.Line), + Entries: []domain.LogEntry{entry}, + }) + if err != nil { + t.Fatalf("ingest repaired declared process log batch: %v", err) + } + if !ack.Accepted || ack.LatestSeq != entry.Seq { + t.Fatalf("unexpected repaired stream ack: %+v", ack) + } +} + func TestCoreServiceRejectsOutOfOrderAndConflictingLogBatches(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) createLogStreamFixture(t, svc) diff --git a/platform/service/protected_requests.go b/platform/service/protected_requests.go index 3ffd597..7ddf100 100644 --- a/platform/service/protected_requests.go +++ b/platform/service/protected_requests.go @@ -100,6 +100,19 @@ func (svc *CoreService) dispatchProtectedRequest(command domain.GameClientBridge if jobID == "" { return validationError("protected request job binding is missing") } + executionInput := domain.JobExecutionInput{ + WorkspaceScope: command.ProfileKey, + RemoteAdapterKey: declaration.ProtectedRequest.TransportKey, + RemoteAdapterKind: adapterKind, + TimeoutSeconds: declaration.TimeoutSeconds, + PluginID: command.PluginID, + } + if declaration.ProtectedRequest.Kind == "rcon" { + if resolution, resolveErr := svc.resolveProtectedSourceRCONDispatch(command.ServerInstanceID, declaration.ProtectedRequest); resolveErr == nil { + executionInput.WorkspaceScope = resolution.binding.ProfileKey + executionInput.SourceRCON = resolution.plan + } + } if err := svc.protectedRequests.Put(jobID, protectedRequestPayload{commandID: command.ID, kind: declaration.ProtectedRequest.Kind, transportKey: declaration.ProtectedRequest.TransportKey, targetKey: declaration.ProtectedRequest.TargetKey, requestText: requestText, expiresAt: command.ExpiresAt}); err != nil { return err } @@ -113,13 +126,7 @@ func (svc *CoreService) dispatchProtectedRequest(command domain.GameClientBridge IdempotencyKey: "protected-request:" + command.ID, Progress: domain.JobProgress{Percent: 0, Message: "protected request queued"}, RetryPolicy: domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 1, MaxBackoffSeconds: 1}, - ExecutionInput: domain.JobExecutionInput{ - WorkspaceScope: command.ProfileKey, - RemoteAdapterKey: declaration.ProtectedRequest.TransportKey, - RemoteAdapterKind: adapterKind, - TimeoutSeconds: declaration.TimeoutSeconds, - PluginID: command.PluginID, - }, + ExecutionInput: executionInput, } if job.RunEndpointID == "" { svc.protectedRequests.Delete(jobID) diff --git a/platform/service/resources.go b/platform/service/resources.go index dd28f5b..3f33225 100644 --- a/platform/service/resources.go +++ b/platform/service/resources.go @@ -202,6 +202,8 @@ type Core interface { GetLogStreamForSession(string, string) (domain.LogStream, error) ListLogStreamsForSession(string, domain.LogStreamFilter) ([]domain.LogStream, error) QueryLogStreamForSession(string, domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) + SubscribeLogEvents(string) (LogEventSubscription, error) + SubscribeLogEventsForSession(string, string) (LogEventSubscription, error) IngestLogBatch(domain.LogBatchIngest) (domain.LogBatchIngestResult, error) QueryLogStream(domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) ListGamePlayersForSession(string, domain.GamePlayerFilter) ([]domain.GamePlayer, error) @@ -236,6 +238,9 @@ type CoreService struct { bridgeMu sync.Mutex bridgeSeq uint64 logStore LogBodyStore + logEventMu sync.Mutex + logEventSubscribers map[uint64]logEventSubscriber + logEventSubscriberSeq uint64 artifactStore ArtifactBodyStore artifactMu sync.Mutex artifactTransfers map[string]domain.ArtifactTransferSession @@ -279,6 +284,7 @@ func newCoreServiceWithLogStore(store repo.Store, logStore LogBodyStore, now fun authSessions: map[string]string{}, runSessions: map[string]domain.RunControlSession{}, logStore: logStore, + logEventSubscribers: map[uint64]logEventSubscriber{}, artifactStore: artifactStore, artifactTransfers: map[string]domain.ArtifactTransferSession{}, artifactPayloads: map[string][]byte{}, @@ -2329,21 +2335,36 @@ func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) err streams := []struct { key string source domain.LogStreamSource - }{ - {key: "stdout", source: domain.LogStreamSourceProcess}, - {key: "stderr", source: domain.LogStreamSourceProcess}, + }{} + addStream := func(key string, source domain.LogStreamSource) { + key = strings.TrimSpace(key) + if key == "" { + return + } + for _, stream := range streams { + if stream.key == key { + return + } + } + streams = append(streams, struct { + key string + source domain.LogStreamSource + }{key: key, source: source}) + } + addStream("stdout", domain.LogStreamSourceProcess) + addStream("stderr", domain.LogStreamSourceProcess) + for _, source := range job.ExecutionInput.LogSources { + if source.Kind != "process.stdout" && source.Kind != "process.stderr" { + continue + } + addStream(source.StreamKey, domain.LogStreamSourceProcess) } if job.Capability == domain.JobCapabilityRemoteRunProgram { - streams = append(streams, - struct { - key string - source domain.LogStreamSource - }{key: "management-program.stdout", source: domain.LogStreamSourceManagementProgram}, - struct { - key string - source domain.LogStreamSource - }{key: "management-program.stderr", source: domain.LogStreamSourceManagementProgram}, - ) + addStream("management-program.stdout", domain.LogStreamSourceManagementProgram) + addStream("management-program.stderr", domain.LogStreamSourceManagementProgram) + } + if job.Capability == domain.JobCapabilityLogsBackfill && job.ExecutionInput.LogSource != nil && strings.TrimSpace(job.ExecutionInput.LogSource.StreamKey) != "" { + addStream(job.ExecutionInput.LogSource.StreamKey, domain.LogStreamSourceFile) } for _, item := range streams { stream := domain.LogStream{ diff --git a/platform/service/resources_test.go b/platform/service/resources_test.go index ff8d84c..f5aa7f6 100644 --- a/platform/service/resources_test.go +++ b/platform/service/resources_test.go @@ -1695,7 +1695,20 @@ func createCompleteRuntimeBinding(t *testing.T, svc *CoreService, instance domai if err != nil { t.Fatalf("get plugin for runtime binding: %v", err) } - binding, err := svc.buildRuntimeBinding(instance, plugin, domain.RuntimeBindingUpdate{ProfileKey: profileKey, Bindings: map[string]string{}}, true) + profile, ok := runtimeLifecycleProfile(plugin.RuntimeProfiles, profileKey) + if !ok { + t.Fatalf("runtime profile %s missing", profileKey) + } + required, _ := runtimeBindingKeys(plugin.RuntimeProfiles, profile) + bindings := map[string]string{} + for _, key := range required { + if runtimeBindingTestKeyIsSensitive(key) { + bindings[key] = "secret://" + instance.ID + "/" + strings.ReplaceAll(key, "/", "-") + } else { + bindings[key] = "runtime." + strings.ReplaceAll(key, "/", ".") + } + } + binding, err := svc.buildRuntimeBinding(instance, plugin, domain.RuntimeBindingUpdate{ProfileKey: profileKey, Bindings: bindings}, true) if err != nil { t.Fatalf("build runtime binding: %v", err) } @@ -1705,6 +1718,11 @@ func createCompleteRuntimeBinding(t *testing.T, svc *CoreService, instance domai return binding } +func runtimeBindingTestKeyIsSensitive(key string) bool { + normalized := strings.ToLower(key) + return strings.Contains(normalized, "password") || strings.Contains(normalized, "credential") || strings.Contains(normalized, "secret") || strings.Contains(normalized, "token") || strings.Contains(normalized, "dsn") +} + func validPluginManifestRegistration() domain.GamePluginManifestRegistration { return domain.GamePluginManifestRegistration{ ManifestRef: "artifact://manifests/game.example/0.1.0", diff --git a/platform/service/runtime_bindings.go b/platform/service/runtime_bindings.go index 9d86984..0ebd585 100644 --- a/platform/service/runtime_bindings.go +++ b/platform/service/runtime_bindings.go @@ -39,8 +39,8 @@ func (svc *CoreService) UpdateServerRuntimeBindingForSession(sessionID, serverIn if existingErr != nil && !errors.Is(existingErr, repo.ErrNotFound) { return domain.RuntimeBindingView{}, existingErr } - if (instance.State == domain.ServerInstanceStateInstalling || instance.State == domain.ServerInstanceStateRunning) && existingErr == nil || instance.State == domain.ServerInstanceStateDeleted { - return domain.RuntimeBindingView{}, validationError("runtime binding cannot be changed while the server is active") + if instance.State == domain.ServerInstanceStateDeleted { + return domain.RuntimeBindingView{}, validationError("runtime binding cannot be changed after the server is deleted") } plugin, err := svc.store.GamePlugins().Get(instance.PluginID) if err != nil { @@ -124,7 +124,7 @@ func runtimeBindingKeys(profiles domain.GamePluginRuntimeProfiles, profile domai add(probe.TargetKey, probe.Required) } for _, source := range profiles.LogSources { - add(source.TargetKey, source.TargetKey != "") + add(source.TargetKey, false) } for _, plan := range profiles.InstallPlans { for _, step := range plan.Steps { diff --git a/platform/service/runtime_bindings_test.go b/platform/service/runtime_bindings_test.go index 342ec60..db3ab06 100644 --- a/platform/service/runtime_bindings_test.go +++ b/platform/service/runtime_bindings_test.go @@ -98,6 +98,52 @@ func TestRuntimeBindingValidationAndLifecycleGating(t *testing.T) { if err != nil || result.Job.TargetKey != "actions/start.json" || result.Job.ExecutionInput.WorkspaceScope != "local" { t.Fatalf("expected complete binding to permit start, result=%+v err=%v", result, err) } + view, err = svc.UpdateServerRuntimeBindingForSession(ownerSession, instance.ID, domain.RuntimeBindingUpdate{ProfileKey: "local", Bindings: map[string]string{"server-root": "runtime.server-root-updated"}}) + if err != nil || view.Status != domain.RuntimeBindingStatusComplete { + t.Fatalf("expected active server runtime binding edits to remain available, view=%+v err=%v", view, err) + } + instance.State = domain.ServerInstanceStateDeleted + if err := svc.store.ServerInstances().Update(instance); err != nil { + t.Fatalf("mark deleted: %v", err) + } + if _, err := svc.UpdateServerRuntimeBindingForSession(ownerSession, instance.ID, domain.RuntimeBindingUpdate{ProfileKey: "local", Bindings: map[string]string{"server-root": "runtime.server-root"}}); err == nil || !strings.Contains(err.Error(), "deleted") { + t.Fatalf("expected deleted server runtime binding edit rejection, got %v", err) + } +} + +func TestRuntimeBindingLogSourcesAreConfigurableButNotRequired(t *testing.T) { + svc := newTestCoreService() + plugin, endpoint := createPluginAndRunEndpoint(t, svc) + plugin.RuntimeProfiles = requiredRuntimeProfilesFixture() + plugin.RuntimeProfiles.LogSources = []domain.RuntimeLogSource{ + {Key: "console-stdout", Kind: "process.stdout", TargetKey: "process/server", StreamKey: "console.stdout", CursorKind: "sequence", RetentionDays: 30}, + {Key: "latest", Kind: "file.tail", TargetKey: "logs/latest", StreamKey: "latest-log", CursorKind: "offset", RetentionDays: 30}, + } + if err := svc.store.GamePlugins().Update(plugin); err != nil { + t.Fatalf("update plugin profiles: %v", err) + } + ownerSession := createServiceUserAndLogin(t, svc, domain.User{ID: "runtime-logs-owner", DisplayName: "Runtime Logs Owner", Email: "runtime-logs-owner@example.test", Roles: []string{"server-owner"}, PasswordHash: "secret-password"}) + instance, err := svc.CreateServerInstanceForSession(ownerSession, domain.ServerInstance{ID: "runtime-log-sources", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "Runtime Log Sources", State: domain.ServerInstanceStateRunning}) + if err != nil { + t.Fatalf("create server: %v", err) + } + + view, err := svc.UpdateServerRuntimeBindingForSession(ownerSession, instance.ID, domain.RuntimeBindingUpdate{ProfileKey: "local", Bindings: map[string]string{"server-root": "runtime.server-root", "rcon.password": "secret://runtime-log-sources/rcon"}}) + if err != nil || view.Status != domain.RuntimeBindingStatusComplete || len(view.MissingKeys) != 0 { + t.Fatalf("log sources should not block runtime readiness: view=%+v err=%v", view, err) + } + seenLogs := map[string]domain.RuntimeBindingKeyView{} + for _, key := range view.Keys { + if strings.HasPrefix(key.Key, "logs/") || strings.HasPrefix(key.Key, "process/") { + seenLogs[key.Key] = key + } + } + for _, key := range []string{"logs/latest", "process/server"} { + item, ok := seenLogs[key] + if !ok || item.Required || item.Configured { + t.Fatalf("expected optional unconfigured log key %s, seen=%+v view=%+v", key, seenLogs, view) + } + } } func requiredRuntimeProfilesFixture() domain.GamePluginRuntimeProfiles { diff --git a/platform/service/server_lifecycle.go b/platform/service/server_lifecycle.go index b1c8d4a..20e50ca 100644 --- a/platform/service/server_lifecycle.go +++ b/platform/service/server_lifecycle.go @@ -310,6 +310,7 @@ func (svc *CoreService) dispatchLifecycleJob(instance domain.ServerInstance, act return domain.Job{}, validationError(fmt.Sprintf("plugin %s lifecycle action is required", action)) } var dllExtensions []domain.RuntimeDLLExtensionPlan + var logSources []domain.RuntimeLogSource if action == domain.ServerLifecycleActionStart && hasProfile { endpoint, err := svc.store.RunEndpoints().Get(instance.RunEndpointID) if err != nil { @@ -319,6 +320,7 @@ func (svc *CoreService) dispatchLifecycleJob(instance domain.ServerInstance, act if err != nil { return domain.Job{}, err } + logSources = lifecycleProcessLogSources(plugin.RuntimeProfiles) } job, err := svc.CreateJob(domain.Job{ ID: lifecycleJobID(instance.ID, action, idempotencyKey), @@ -332,6 +334,7 @@ func (svc *CoreService) dispatchLifecycleJob(instance domain.ServerInstance, act WorkspaceScope: profileKey, PluginID: plugin.ID, LifecycleOperation: lifecycleExecutionOperation(action), + LogSources: logSources, DLLExtensions: dllExtensions, Deployment: deploymentPlanForDispatch(instance.Deployment), }, @@ -345,6 +348,16 @@ func (svc *CoreService) dispatchLifecycleJob(instance domain.ServerInstance, act return job, nil } +func lifecycleProcessLogSources(profiles domain.GamePluginRuntimeProfiles) []domain.RuntimeLogSource { + sources := []domain.RuntimeLogSource{} + for _, source := range profiles.LogSources { + if source.Kind == "process.stdout" || source.Kind == "process.stderr" { + sources = append(sources, source) + } + } + return sources +} + func lifecycleJobProgress(deployment domain.ServerDeploymentDefinition) domain.JobProgress { if deployment.Mode != "" { return domain.JobProgress{Percent: 0, Phase: "queued", Message: "deployment queued; awaiting Run claim"} diff --git a/platform/service/server_lifecycle_test.go b/platform/service/server_lifecycle_test.go index f427a11..7b34391 100644 --- a/platform/service/server_lifecycle_test.go +++ b/platform/service/server_lifecycle_test.go @@ -52,6 +52,9 @@ func TestCoreServiceServerLifecycleWorkflows(t *testing.T) { if started.Job.ExecutionInput.PluginID != "server.scum" || started.Job.ExecutionInput.WorkspaceScope != "local" || started.Job.ExecutionInput.LifecycleOperation != "start" { t.Fatalf("expected start job to carry plugin/profile metadata, got %+v", started.Job.ExecutionInput) } + if len(started.Job.ExecutionInput.LogSources) != 2 || started.Job.ExecutionInput.LogSources[0].StreamKey != "scum.console.stdout" || started.Job.ExecutionInput.LogSources[1].StreamKey != "scum.console.stderr" { + t.Fatalf("expected start job to carry plugin-declared process log sources, got %+v", started.Job.ExecutionInput.LogSources) + } claimAndCompleteLifecycleJob(t, svc, sessionToken, domain.LifecycleCapabilityStart, domain.JobStateSucceeded) running, err := svc.GetServerInstance("server-1") if err != nil { @@ -362,8 +365,11 @@ func createLifecyclePlugin(t *testing.T, svc *CoreService) domain.GamePlugin { Start: "actions/start.json", Stop: "actions/stop.json", }, - Permissions: domain.PluginPermissions{Jobs: true, Logs: true}, - RuntimeProfiles: domain.GamePluginRuntimeProfiles{LifecycleProfiles: []domain.RuntimeLifecycleProfile{{Key: "local", Mode: "local-process", Capabilities: []string{domain.LifecycleCapabilityInstall, domain.LifecycleCapabilityStart, domain.LifecycleCapabilityStop}}}}, + Permissions: domain.PluginPermissions{Jobs: true, Logs: true}, + RuntimeProfiles: domain.GamePluginRuntimeProfiles{ + LifecycleProfiles: []domain.RuntimeLifecycleProfile{{Key: "local", Mode: "local-process", Capabilities: []string{domain.LifecycleCapabilityInstall, domain.LifecycleCapabilityStart, domain.LifecycleCapabilityStop}}}, + LogSources: []domain.RuntimeLogSource{{Key: "scum-console-stdout", Kind: "process.stdout", TargetKey: "scum/server-process", StreamKey: "scum.console.stdout", CursorKind: "sequence", RetentionDays: 30}, {Key: "scum-console-stderr", Kind: "process.stderr", TargetKey: "scum/server-process", StreamKey: "scum.console.stderr", CursorKind: "sequence", RetentionDays: 30}}, + }, }) if err != nil { t.Fatalf("create lifecycle plugin: %v", err) diff --git a/platform/service/source_rcon.go b/platform/service/source_rcon.go index c067db8..01bd3b8 100644 --- a/platform/service/source_rcon.go +++ b/platform/service/source_rcon.go @@ -212,9 +212,16 @@ func (svc *CoreService) resolveSourceRCONDispatch(instance domain.ServerInstance } func sourceRCONTransport(profiles domain.GamePluginRuntimeProfiles, profile domain.RuntimeLifecycleProfile) (domain.RuntimeTransportProfile, error) { + return sourceRCONTransportForCapability(profiles, profile, "", domain.JobCapabilityRemoteRunRCONCommand) +} + +func sourceRCONTransportForCapability(profiles domain.GamePluginRuntimeProfiles, profile domain.RuntimeLifecycleProfile, requiredKey string, capability string) (domain.RuntimeTransportProfile, error) { var selected domain.RuntimeTransportProfile for _, candidate := range profiles.TransportProfiles { - if !containsString(profile.TransportKeys, candidate.Key) || candidate.Kind != "rcon" || !containsString(candidate.Capabilities, domain.JobCapabilityRemoteRunRCONCommand) { + if requiredKey != "" && candidate.Key != requiredKey { + continue + } + if !containsString(profile.TransportKeys, candidate.Key) || candidate.Kind != "rcon" || !containsString(candidate.Capabilities, capability) { continue } if selected.Key != "" { @@ -228,6 +235,62 @@ func sourceRCONTransport(profiles domain.GamePluginRuntimeProfiles, profile doma return selected, nil } +func (svc *CoreService) resolveProtectedSourceRCONDispatch(serverInstanceID string, request *domain.GameClientBridgeProtectedRequestDeclaration) (sourceRCONDispatchResolution, error) { + if request == nil || request.Kind != "rcon" { + return sourceRCONDispatchResolution{}, validationError("protected request is not RCON") + } + instance, err := svc.store.ServerInstances().Get(serverInstanceID) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + plugin, err := svc.store.GamePlugins().Get(instance.PluginID) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + capability := domain.JobCapabilityRemoteRunProtectedRCON + if plugin.Status != domain.GamePluginStatusInstalled || plugin.Version != instance.PluginVersion || !plugin.Permissions.RemoteAccess || !plugin.RemoteAccess.RCON || !containsString(plugin.RequiredRunCapabilities, capability) || !containsString(plugin.RemoteAccess.RunCapabilities, capability) { + return sourceRCONDispatchResolution{}, forbiddenError("plugin does not declare protected SCUM RCON access") + } + endpoint, err := svc.store.RunEndpoints().Get(instance.RunEndpointID) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + if err := svc.validateRunnableEndpoint(endpoint, capability); err != nil { + return sourceRCONDispatchResolution{}, err + } + if !strings.EqualFold(endpoint.Platform, "windows") || !strings.EqualFold(endpoint.Architecture, "amd64") { + return sourceRCONDispatchResolution{}, validationError("unsupported_extension_platform: SCUM Source RCON requires windows/amd64") + } + binding, err := svc.runtimeBindingForServer(instance.ID) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + binding, err = normalizeRuntimeBinding(plugin, binding) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + if binding.Status != domain.RuntimeBindingStatusComplete || binding.PluginVersion != plugin.Version { + return sourceRCONDispatchResolution{}, validationError("runtime binding is incomplete or stale") + } + profile, exists := runtimeLifecycleProfileForKey(plugin.RuntimeProfiles, binding.ProfileKey) + if !exists || !containsString(profile.Capabilities, capability) || !runtimePlatformsContain(profile.Platforms, "windows") { + return sourceRCONDispatchResolution{}, validationError("selected runtime profile does not support protected SCUM RCON") + } + transport, err := sourceRCONTransportForCapability(plugin.RuntimeProfiles, profile, request.TransportKey, capability) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + if transport.TargetKey != request.TargetKey { + return sourceRCONDispatchResolution{}, validationError("protected RCON transport target is invalid") + } + extension, err := sourceRCONExtension(plugin.RuntimeProfiles, profile, endpoint) + if err != nil { + return sourceRCONDispatchResolution{}, err + } + plan := &domain.RuntimeSourceRCONPlan{Protocol: "source-rcon", ExtensionKey: extension.Key, ModKey: extension.ModKey, ConfigRef: "ue4ss/Mods/" + extension.ModKey + "/config.ini", DeploymentStateRef: "runtime/ue4ss-dll/" + extension.TargetKey + "/release.json", Port: extension.RCONPort} + return sourceRCONDispatchResolution{plugin: plugin, binding: binding, transport: transport, plan: plan}, nil +} + func sourceRCONExtension(profiles domain.GamePluginRuntimeProfiles, profile domain.RuntimeLifecycleProfile, endpoint domain.RunEndpoint) (domain.RuntimeDLLExtensionProfile, error) { byKey := make(map[string]domain.RuntimeDLLExtensionProfile, len(profiles.DLLExtensions)) for _, extension := range profiles.DLLExtensions { diff --git a/platform/service/source_rcon_test.go b/platform/service/source_rcon_test.go index 2bb74b1..acdf175 100644 --- a/platform/service/source_rcon_test.go +++ b/platform/service/source_rcon_test.go @@ -85,6 +85,65 @@ func TestSourceRCONDispatchUsesOneTimeRedactedInput(t *testing.T) { } } +func TestProtectedRCONBridgeDispatchCarriesSourceRCONPlan(t *testing.T) { + svc, _, _, instance := newSourceRCONFixture(t) + plugin, err := svc.store.GamePlugins().Get(instance.PluginID) + if err != nil { + t.Fatal(err) + } + protectedCapability := domain.JobCapabilityRemoteRunProtectedRCON + plugin.RequiredRunCapabilities = append(plugin.RequiredRunCapabilities, protectedCapability) + plugin.RemoteAccess.RunCapabilities = append(plugin.RemoteAccess.RunCapabilities, protectedCapability) + plugin.RuntimeProfiles.ClientManagers = []domain.RuntimeClientManagerProfile{{Key: "scum-client-manager", Health: domain.RuntimeClientManagerHealth{RequiredCapabilities: []string{gameClientBridgeCapability}}}} + plugin.RuntimeProfiles.LifecycleProfiles[0].Capabilities = append(plugin.RuntimeProfiles.LifecycleProfiles[0].Capabilities, protectedCapability) + plugin.RuntimeProfiles.LifecycleProfiles[0].TransportKeys = append(plugin.RuntimeProfiles.LifecycleProfiles[0].TransportKeys, "scum-management") + plugin.RuntimeProfiles.TransportProfiles = append(plugin.RuntimeProfiles.TransportProfiles, domain.RuntimeTransportProfile{Key: "scum-management", Kind: "rcon", TargetKey: "scum-management", Capabilities: []string{protectedCapability}}) + plugin.GameClientBridge.Commands = append(plugin.GameClientBridge.Commands, domain.GameClientBridgeCommandDeclaration{Type: "management.rcon.request", ApprovalLevel: domain.GameClientBridgeApprovalLevelOperator, TimeoutSeconds: 120, MaxPayloadBytes: 8192, ProtectedRequest: &domain.GameClientBridgeProtectedRequestDeclaration{Kind: "rcon", TransportKey: "scum-management", TargetKey: "scum-management", TextField: "requestText", MaxTextBytes: 8192}}) + if err := svc.store.GamePlugins().Update(plugin); err != nil { + t.Fatal(err) + } + binding, err := svc.buildRuntimeBinding(instance, plugin, domain.RuntimeBindingUpdate{ProfileKey: "local", Bindings: map[string]string{"rcon": "runtime-rcon", "scum-management": "runtime-rcon"}}, true) + if err != nil { + t.Fatalf("refresh protected RCON binding: %v", err) + } + if err := svc.store.RuntimeBindings().Update(binding); err != nil { + t.Fatalf("store protected RCON binding: %v", err) + } + endpoint, err := svc.store.RunEndpoints().Get(instance.RunEndpointID) + if err != nil { + t.Fatal(err) + } + endpoint.Capabilities = append(endpoint.Capabilities, protectedCapability) + if err := svc.store.RunEndpoints().Update(endpoint); err != nil { + t.Fatal(err) + } + if _, err := svc.resolveProtectedSourceRCONDispatch(instance.ID, plugin.GameClientBridge.Commands[len(plugin.GameClientBridge.Commands)-1].ProtectedRequest); err != nil { + t.Fatalf("resolve protected Source RCON plan: %v", err) + } + + command, err := svc.queueGameClientBridgeCommand("user-rcon-owner", domain.GameClientBridgeQueueRequest{ServerInstanceID: instance.ID, PluginID: plugin.ID, ProfileKey: "scum-client-manager", CommandType: "management.rcon.request", Payload: map[string]any{"requestText": "#ListPlayers"}, IdempotencyKey: "protected-rcon-1", ExpiresAt: fixedTime.Add(time.Minute)}) + if err != nil { + t.Fatalf("queue protected RCON: %v", err) + } + job, err := svc.store.Jobs().Get(command.RunJobID) + if err != nil { + t.Fatalf("get protected RCON job: %v", err) + } + if job.Capability != protectedCapability || job.InputRef == "" || !strings.HasPrefix(job.InputRef, "input://protected-request/") || job.ExecutionInput.SourceRCON == nil { + t.Fatalf("expected protected RCON job with frozen Source RCON plan, got %+v", job) + } + if job.ExecutionInput.WorkspaceScope != "local" || job.ExecutionInput.RemoteAdapterKey != "scum-management" || job.ExecutionInput.RemoteAdapterKind != "protected-rcon" || job.ExecutionInput.SourceRCON.Port != 27015 { + t.Fatalf("protected RCON plan did not preserve logical runtime binding: %+v", job.ExecutionInput) + } + serialized, err := json.Marshal(job) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(serialized), "#ListPlayers") || strings.Contains(string(serialized), "password=") { + t.Fatalf("protected RCON job leaked transient input: %s", serialized) + } +} + func TestSourceRCONDispatchRejectsUnsafeOrIncompatibleState(t *testing.T) { svc, session, _, instance := newSourceRCONFixture(t) unsafe := domain.SourceRCONCommandRequest{ServerInstanceID: instance.ID, Kind: domain.SourceRCONCommandKindCommand, Command: "SetTime 12\nSpawnItem", IdempotencyKey: "rcon-unsafe"} diff --git a/platform/validator/resources.go b/platform/validator/resources.go index a385be4..2ee813b 100644 --- a/platform/validator/resources.go +++ b/platform/validator/resources.go @@ -1392,9 +1392,35 @@ func ValidateJob(job domain.Job) error { for i, plan := range job.ExecutionInput.DLLExtensions { violations = append(violations, validateRuntimeDLLExtensionPlan(fmt.Sprintf("executionInput.dllExtensions[%d]", i), plan)...) } + if job.ExecutionInput.LogSource != nil { + violations = append(violations, validateRuntimeLogSourcePlanForJob("executionInput.logSource", job.ExecutionInput.LogSource)...) + if job.Capability != domain.JobCapabilityLogsBackfill || job.ServerInstanceID == "" { + violations = append(violations, "executionInput.logSource is allowed only for logs.backfill jobs") + } + } + if len(job.ExecutionInput.LogSources) > 0 { + if job.Capability != domain.LifecycleCapabilityStart || job.ExecutionInput.LifecycleOperation != "start" || job.ServerInstanceID == "" { + violations = append(violations, "executionInput.logSources are allowed only for scoped process.start jobs") + } + seenKinds := map[string]struct{}{} + for i, source := range job.ExecutionInput.LogSources { + prefix := fmt.Sprintf("executionInput.logSources[%d]", i) + violations = append(violations, validateRuntimeProcessLogSourcePlanForJob(prefix, source)...) + if _, exists := seenKinds[source.Kind]; exists { + violations = append(violations, "executionInput.logSources kind is duplicated") + } + seenKinds[source.Kind] = struct{}{} + } + } if job.ExecutionInput.SourceRCON != nil { violations = append(violations, validateRuntimeSourceRCONPlan("executionInput.sourceRcon", job.ExecutionInput.SourceRCON)...) - if job.Capability != domain.JobCapabilityRemoteRunRCONCommand || job.ExecutionInput.RemoteAdapterKind != "rcon" { + isSourceCommand := job.Capability == domain.JobCapabilityRemoteRunRCONCommand + isProtectedRCON := job.Capability == domain.JobCapabilityRemoteRunProtectedRCON + wantAdapterKind := "rcon" + if isProtectedRCON { + wantAdapterKind = "protected-rcon" + } + if (!isSourceCommand && !isProtectedRCON) || job.ExecutionInput.RemoteAdapterKind != wantAdapterKind { violations = append(violations, "executionInput.sourceRcon is allowed only for rcon jobs") } if job.RetryPolicy.MaxAttempts != 1 { @@ -1441,6 +1467,45 @@ func ValidateJob(job domain.Job) error { return finish(violations) } +func validateRuntimeLogSourcePlanForJob(prefix string, source *domain.RuntimeLogSource) []string { + if source == nil { + return nil + } + var violations []string + violations = append(violations, validateProfileKey(prefix+".key", source.Key)...) + violations = append(violations, validateProfileKey(prefix+".targetKey", source.TargetKey)...) + violations = append(violations, validateProfileKey(prefix+".streamKey", source.StreamKey)...) + if source.Kind != "file.tail" { + violations = append(violations, prefix+".kind must be file.tail") + } + if source.CursorKind != "" && !oneOf(source.CursorKind, "offset", "fingerprint") { + violations = append(violations, prefix+".cursorKind is invalid") + } + if source.RetentionDays < 0 || source.RetentionDays > 365 { + violations = append(violations, prefix+".retentionDays is invalid") + } + return violations +} + +func validateRuntimeProcessLogSourcePlanForJob(prefix string, source domain.RuntimeLogSource) []string { + var violations []string + violations = append(violations, validateProfileKey(prefix+".key", source.Key)...) + if source.TargetKey != "" { + violations = append(violations, validateProfileKey(prefix+".targetKey", source.TargetKey)...) + } + violations = append(violations, validateProfileKey(prefix+".streamKey", source.StreamKey)...) + if source.Kind != "process.stdout" && source.Kind != "process.stderr" { + violations = append(violations, prefix+".kind must be process.stdout or process.stderr") + } + if source.CursorKind != "" && source.CursorKind != "sequence" { + violations = append(violations, prefix+".cursorKind is invalid") + } + if source.RetentionDays < 0 || source.RetentionDays > 365 { + violations = append(violations, prefix+".retentionDays is invalid") + } + return violations +} + func ValidateArtifact(artifact domain.Artifact) error { var violations []string violations = appendRequired(violations, "id", artifact.ID) diff --git a/platform_web/api/client.ts b/platform_web/api/client.ts index 83a1620..1892687 100644 --- a/platform_web/api/client.ts +++ b/platform_web/api/client.ts @@ -55,6 +55,7 @@ import type { LlmConfigSuggestionRequest, LlmConfigSuggestionResponse, LogBackfillRequest, + LogStreamEventOptions, LogStreamCursorRequest, LogStreamCursorResponse, LogStreamListResponse, @@ -446,6 +447,17 @@ export class PlatformApiClient { return this.request(`/server-instances/${encodeURIComponent(id)}/logs/live`); } + openServerLogEvents(id: string, options: LogStreamEventOptions = {}): EventSource { + return new EventSource(this.serverLogEventsUrl(id, options), { withCredentials: true }); + } + + serverLogEventsUrl(id: string, options: LogStreamEventOptions = {}): string { + const params = new URLSearchParams(); + if (options.historyLimit !== undefined) params.set("historyLimit", String(options.historyLimit)); + const query = params.toString(); + return `${this.baseUrl}/server-instances/${encodeURIComponent(id)}/logs/events${query ? `?${query}` : ""}`; + } + async requestLogBackfill(id: string, request: LogBackfillRequest): Promise { return this.request(`/server-instances/${encodeURIComponent(id)}/logs/backfill`, { method: "POST", diff --git a/platform_web/api/contracts.md b/platform_web/api/contracts.md index b24d041..b4dba99 100644 --- a/platform_web/api/contracts.md +++ b/platform_web/api/contracts.md @@ -60,7 +60,7 @@ Existing platform APIs already cover server lifecycle, jobs, log stream metadata - Operation/job traceability reuses `GET /api/v1/jobs`, `GET /api/v1/jobs/{id}`, `POST /api/v1/jobs/{id}/cancel`, and `GET /api/v1/audit-events`; the frontend wraps these in one visible operation lifecycle per user intent. Browser Job contracts explicitly exclude raw or hashed lease tokens, Run session tokens/generations, secret refs, host paths, sockets, and credentials. The safe schema rejects those keys, and existing API client 401/403 behavior remains authoritative for expired sessions and cross-owner access. -- Log filtering by level/keyword/time/source is applied client-side over `POST /api/v1/log-streams/query` (`LogStreamCursorRequest`) results until the platform exposes server-side filters. +- Live log and management-terminal output uses `GET /api/v1/server-instances/{id}/logs/events` as a single `EventSource`/SSE stream with bounded initial history. `POST /api/v1/log-streams/query` remains available for explicit historical cursor reads and reconnect repair, not periodic browser polling. # Client Manager API projection `PlatformApiClient` exposes list/detail and typed deploy, control, update, retry, revoke-session, and uninstall methods. `schemas/clientManagerLifecycle.ts` validates status/action/job/health fields and rejects forbidden machine or credential fields before rendering. Lifecycle commands carry profile, distribution, expected deployment generation, approval/confirmation, and idempotency only. Artifact bytes remain in the platform-owned artifact transfer client. diff --git a/platform_web/api/types.ts b/platform_web/api/types.ts index 7b4fd6a..fa57d8e 100644 --- a/platform_web/api/types.ts +++ b/platform_web/api/types.ts @@ -13,10 +13,10 @@ export type DependencyState = "unknown" | "present" | "missing" | "installing" | export type RunUpdatePhase = "queued" | "downloading" | "staged" | "restart-requested" | "activating" | "succeeded" | "rolled-back" | "failed"; export type ServerLifecycleAction = "create" | "start" | "stop" | "status"; -export type GameClientBridgeCommandState = "pending" | "claimed" | "succeeded" | "failed" | "cancelled" | "expired"; +export type GameClientBridgeCommandState = "pending" | "claimed" | "succeeded" | "failed" | "cancelled" | "expired" | "unknown"; export type GameClientBridgeApprovalState = "not_required" | "pending" | "approved" | "rejected"; export type GameClientBridgeApprovalLevel = "none" | "operator" | "platform-admin"; -export type GameClientBridgeResultStatus = "succeeded" | "failed" | "cancelled"; +export type GameClientBridgeResultStatus = "succeeded" | "failed" | "cancelled" | "unknown"; export type GameClientBridgeJsonValue = string | number | boolean | null | GameClientBridgeJsonValue[] | GameClientBridgeJsonObject; export interface GameClientBridgeJsonObject { @@ -1494,6 +1494,25 @@ export interface LogStreamCursorResponse { latestSeq: number; } +export interface LogStreamEventResponse { + serverInstanceId: string; + streamId: string; + source: string; + streamKey: string; + latestSeq: number; + entry: LogEntryBody; +} + +export interface LogStreamEventsReadyResponse { + serverInstanceId: string; + streamCount: number; + serverTime: string; +} + +export interface LogStreamEventOptions { + historyLimit?: number; +} + export interface AuditEventResponse { id: string; actorId: string; diff --git a/platform_web/components/ServerDeploymentWorkflow.tsx b/platform_web/components/ServerDeploymentWorkflow.tsx index 015f1d7..f5d4c6a 100644 --- a/platform_web/components/ServerDeploymentWorkflow.tsx +++ b/platform_web/components/ServerDeploymentWorkflow.tsx @@ -112,7 +112,7 @@ export function ServerDeploymentWorkflow({ open, kind, plugins, endpoints, initi const protectedState = (nextValue: string, configured: boolean) => nextValue.trim() ? "将替换" : configured ? "保持已配置" : "未配置"; const actionLabel = kind === "create" ? "创建服务器" : "保存部署设置"; - return + return
void submit(event)} aria-label={kind === "create" ? "创建服务器部署向导" : "编辑服务器部署向导"}>
    {workflowSteps.map((item, index) => { const Icon = item.icon; return
  1. {index < step ? : }{index + 1}. {item.label}
    2. ; })}
{step === pluginStep &&
diff --git a/platform_web/components/ServerLiveOperations.tsx b/platform_web/components/ServerLiveOperations.tsx index b4da7d8..81d31be 100644 --- a/platform_web/components/ServerLiveOperations.tsx +++ b/platform_web/components/ServerLiveOperations.tsx @@ -1,30 +1,29 @@ import { ListChecks, Pause, Play, RotateCw, Send, Sparkles, Terminal, Trash2, X } from "lucide-react"; -import { type FormEvent, type KeyboardEvent as ReactKeyboardEvent, type ReactNode, useCallback, useEffect, useMemo, useState } from "react"; +import { type FormEvent, type KeyboardEvent as ReactKeyboardEvent, type ReactNode, useCallback, useEffect, useMemo, useRef, useState } from "react"; import { platformApiClient } from "../api/client"; -import type { LogEntryBody, LogStreamResponse } from "../api/types"; -import { sourceRCONRawCommandRequest } from "../schemas/sourceRcon"; +import type { GameClientBridgeCommandResponse, LogEntryBody, LogStreamResponse } from "../api/types"; +import { scumManagementRCONCommandRequest } from "../schemas/scumManagementRcon"; import { cx } from "../utils/classes"; +import { appendLiveLogEntries, entryFromServerLogEvent, mergeLogStreams, parseLogStreamEvent, parseServerLogEvent, streamFromServerLogEvent, type LiveLogEntry } from "../utils/logEvents"; import { EmptyState, ErrorState, LoadingState, ResultBadge } from "./StateViews"; type LoadState = { status: "loading" } | { status: "error"; reason: string } | { status: "ready"; data: T }; -type LiveLogEntry = LogEntryBody & { source: string; streamId: string; streamKey: string }; type TerminalLine = { id: string; tone: "input" | "info" | "success" | "warn" | "error"; text: string; at: string; sortKey: number; streamKey?: string; level?: string; seq?: number }; type TerminalQuickCommand = { label: string; command: string; hint: string }; -const liveLogPollMs = 1000; -const terminalLogPollMs = 1000; -const logStreamPollMs = 5000; +const terminalBridgeResultPollMs = 1000; +const terminalBridgeResultPollAttempts = 30; const liveLogHistoryWindow = 100; -const terminalLogQueryLimit = 150; const terminalHistoryWindow = 150; const maxLogEntries = 500; const maxTerminalLines = 600; const terminalQuickCommandCatalog: Record = { "game.scum": [ - { label: "查询玩家", command: "ListPlayers", hint: "SCUM RCON" }, - { label: "服务器信息", command: "ServerInfo", hint: "SCUM RCON" }, - { label: "设为中午", command: "SetTime 12", hint: "SCUM RCON" } + { label: "查询玩家", command: "#ListPlayers", hint: "在线玩家列表" }, + { label: "查询队伍", command: "#ListSquads 1", hint: "服务器队伍列表" }, + { label: "查询车辆", command: "#ListSpawnedVehicles true", hint: "已生成车辆列表" }, + { label: "设为中午", command: "#SetTime 12", hint: "设置游戏时间" } ] }; @@ -85,10 +84,15 @@ export function ServerLiveLogDrawer({ open, serverId, serverName, onClose }: Ser const [streams, setStreams] = useState>({ status: "loading" }); const [selectedStreamId, setSelectedStreamId] = useState(""); const [entries, setEntries] = useState([]); - const [cursorByStream, setCursorByStream] = useState>({}); const [paused, setPaused] = useState(false); const [keyword, setKeyword] = useState(""); const [lastRefreshAt, setLastRefreshAt] = useState(""); + const [eventSourceKey, setEventSourceKey] = useState(0); + const pausedRef = useRef(paused); + + useEffect(() => { + pausedRef.current = paused; + }, [paused]); const loadStreams = useCallback(async (showLoading = true) => { if (!open) return; @@ -105,40 +109,45 @@ export function ServerLiveLogDrawer({ open, serverId, serverName, onClose }: Ser useEffect(() => { if (!open) return; setEntries([]); - setCursorByStream({}); setPaused(false); + setLastRefreshAt(""); void loadStreams(); }, [loadStreams, open]); useEffect(() => { if (!open) return undefined; - const timer = window.setInterval(() => void loadStreams(false).catch(() => undefined), logStreamPollMs); - return () => window.clearInterval(timer); - }, [loadStreams, open]); + let ready = false; + const events = platformApiClient.openServerLogEvents(serverId, { historyLimit: liveLogHistoryWindow }); + events.addEventListener("open", () => setLastRefreshAt(new Date().toLocaleTimeString())); + events.addEventListener("stream", (event) => { + const stream = parseLogStreamEvent(event); + if (!stream) return; + ready = true; + setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); + setSelectedStreamId((current) => current || stream.id); + }); + events.addEventListener("ready", () => { + ready = true; + setStreams((current) => current.status === "ready" ? current : { status: "ready", data: [] }); + }); + events.addEventListener("log", (event) => { + const payload = parseServerLogEvent(event); + if (!payload) return; + ready = true; + setLastRefreshAt(new Date().toLocaleTimeString()); + setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], streamFromServerLogEvent(payload)) })); + setSelectedStreamId((current) => current || payload.streamId); + if (pausedRef.current) return; + setEntries((current) => appendLiveLogEntries(current, [entryFromServerLogEvent(payload)], maxLogEntries)); + }); + events.onerror = () => { + if (!ready) setStreams({ status: "error", reason: "实时日志推送连接失败" }); + }; + return () => events.close(); + }, [eventSourceKey, open, serverId]); const selectedStream = streams.status === "ready" ? streams.data.find((stream) => stream.id === selectedStreamId) : undefined; - const tailSelectedStream = useCallback(async () => { - if (!open || !selectedStream) return; - const afterSeq = cursorByStream[selectedStream.id] ?? initialLogCursor(selectedStream, liveLogHistoryWindow); - const cursor = await platformApiClient.queryLogStream({ logStreamId: selectedStream.id, afterSeq, limit: 100 }); - const nextSeq = nextCursorSeq(afterSeq, cursor.entries, cursor.nextSeq); - setCursorByStream((current) => updateCursor(current, selectedStream.id, nextSeq)); - setLastRefreshAt(new Date().toLocaleTimeString()); - if (cursor.entries.length === 0) return; - setEntries((current) => [ - ...current, - ...cursor.entries.map((entry) => ({ ...entry, source: selectedStream.source || selectedStream.streamKey, streamId: selectedStream.id, streamKey: selectedStream.streamKey })) - ].slice(-maxLogEntries)); - }, [cursorByStream, open, selectedStream]); - - useEffect(() => { - if (!open || paused || !selectedStream) return undefined; - void tailSelectedStream().catch(() => undefined); - const timer = window.setInterval(() => void tailSelectedStream().catch(() => undefined), liveLogPollMs); - return () => window.clearInterval(timer); - }, [open, paused, selectedStream, tailSelectedStream]); - const visibleEntries = useMemo(() => { const query = keyword.trim().toLowerCase(); return entries.filter((entry) => entry.streamId === selectedStreamId && (!query || entry.line.toLowerCase().includes(query) || (entry.level ?? "info").toLowerCase().includes(query))); @@ -146,35 +155,28 @@ export function ServerLiveLogDrawer({ open, serverId, serverName, onClose }: Ser function clearVisibleBuffer() { setEntries((current) => current.filter((entry) => entry.streamId !== selectedStreamId)); - if (selectedStream) setCursorByStream((current) => ({ ...current, [selectedStream.id]: selectedStream.latestSeq })); } function selectLogStream(nextStreamId: string) { setSelectedStreamId(nextStreamId); - setEntries([]); - setCursorByStream((current) => { - const next = { ...current }; - delete next[nextStreamId]; - return next; - }); } return ( - +
setKeyword(event.target.value)} /> - +
- 状态：{paused ? "已暂停" : "自动刷新"} · 最新刷新 {lastRefreshAt || "等待"} · 游标 {selectedStream ? cursorByStream[selectedStream.id] ?? 0 : "--"} + 状态：{paused ? "已暂停" : "实时推送"} · 最新事件 {lastRefreshAt || "等待"} · 游标 {selectedStream ? selectedStream.latestSeq : "--"} {streams.status === "loading" && } {streams.status === "error" && void loadStreams()} compact />} {streams.status === "ready" && streams.data.length === 0 && } - {streams.status === "ready" && streams.data.length > 0 && visibleEntries.length === 0 && } + {streams.status === "ready" && streams.data.length > 0 && visibleEntries.length === 0 && } {visibleEntries.length > 0 && (
{visibleEntries.map((entry) => ( @@ -204,7 +206,6 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu const [pending, setPending] = useState(false); const [lines, setLines] = useState([]); const [streams, setStreams] = useState>({ status: "loading" }); - const [cursorByStream, setCursorByStream] = useState>({}); const [commandHistory, setCommandHistory] = useState([]); const [historyIndex, setHistoryIndex] = useState(null); const [result, setResult] = useState<{ status: "pending" | "succeeded" | "failed"; label: string } | null>(null); @@ -228,57 +229,43 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu } }, [open, serverId]); - const tailTerminalLogs = useCallback(async (targetStreams = terminalStreams) => { - if (!open || targetStreams.length === 0) return; - const cursorUpdates: Record = {}; - const batches = await Promise.all(targetStreams.map(async (stream) => { - const afterSeq = cursorByStream[stream.id] ?? initialLogCursor(stream, terminalHistoryWindow); - const cursor = await platformApiClient.queryLogStream({ logStreamId: stream.id, afterSeq, limit: terminalLogQueryLimit }); - cursorUpdates[stream.id] = nextCursorSeq(afterSeq, cursor.entries, cursor.nextSeq); - return cursor.entries.map((entry) => terminalLineFromLog(stream, entry)); - })); - setCursorByStream((current) => { - let changed = false; - const next = { ...current }; - for (const [streamId, cursor] of Object.entries(cursorUpdates)) { - const nextCursor = Math.max(next[streamId] ?? 0, cursor); - if (nextCursor !== next[streamId]) { - next[streamId] = nextCursor; - changed = true; - } - } - return changed ? next : current; - }); - appendLines(batches.flat().sort(compareTerminalLines)); - }, [appendLines, cursorByStream, open, terminalStreams]); - useEffect(() => { if (!open) return; setCommand(""); setPending(false); setResult(null); - setCursorByStream({}); setHistoryIndex(null); - setLines([terminalSystemLine("info", supportsCommands ? "读取平台历史日志，后续按游标实时追加。" : "该插件暂未声明可用的管理终端命令通道。", "SYSTEM")]); + setLines([terminalSystemLine("info", supportsCommands ? "连接平台日志推送，先补最近历史再实时追加。" : "该插件暂未声明可用的管理终端命令通道。", "SYSTEM")]); void loadStreams(); }, [loadStreams, open, supportsCommands]); useEffect(() => { if (!open) return undefined; - const timer = window.setInterval(() => void loadStreams(false).catch(() => undefined), logStreamPollMs); - return () => window.clearInterval(timer); - }, [loadStreams, open]); - - useEffect(() => { - if (!open || streams.status !== "ready") return; - void tailTerminalLogs(terminalRelevantStreams(streams.data)).catch((error) => appendLines([terminalSystemLine("error", error instanceof Error ? error.message : "历史日志读取失败", "LOGS")])); - }, [appendLines, open, streams]); - - useEffect(() => { - if (!open || terminalStreams.length === 0) return undefined; - const timer = window.setInterval(() => void tailTerminalLogs().catch(() => undefined), terminalLogPollMs); - return () => window.clearInterval(timer); - }, [open, tailTerminalLogs, terminalStreams]); + let ready = false; + const events = platformApiClient.openServerLogEvents(serverId, { historyLimit: terminalHistoryWindow }); + events.addEventListener("stream", (event) => { + const stream = parseLogStreamEvent(event); + if (!stream) return; + ready = true; + setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); + }); + events.addEventListener("ready", () => { + ready = true; + setStreams((current) => current.status === "ready" ? current : { status: "ready", data: [] }); + }); + events.addEventListener("log", (event) => { + const payload = parseServerLogEvent(event); + if (!payload) return; + ready = true; + const stream = streamFromServerLogEvent(payload); + setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); + appendLines([terminalLineFromLog(stream, payload.entry)]); + }); + events.onerror = () => { + if (!ready) setStreams({ status: "error", reason: "实时日志推送连接失败" }); + }; + return () => events.close(); + }, [appendLines, open, serverId]); function selectQuickCommand(item: TerminalQuickCommand) { setCommand(item.command); @@ -322,11 +309,20 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu setResult({ status: "pending", label: "正在提交命令" }); appendLines([terminalSystemLine("input", `> ${submitted}`, "COMMAND")]); try { - const response = await platformApiClient.sendSourceRCONCommand(serverId, sourceRCONRawCommandRequest(serverId, submitted)); - const label = `已排队 · 任务 ${response.jobId}`; - setResult({ status: "succeeded", label }); - appendLines([terminalSystemLine("success", `${label} · ${response.message || response.status}`, "PLATFORM", `ok-${response.jobId}`)]); - void tailTerminalLogs().catch(() => undefined); + const response = await platformApiClient.queueGameClientBridgeCommand(serverId, scumManagementRCONCommandRequest(serverId, submitted)); + const label = bridgeCommandDispatchLabel(response.state, response.id); + setResult({ status: "pending", label: `${label} · 等待 Run 返回结果` }); + appendLines([terminalSystemLine("success", `${label} · protected RCON`, "PLATFORM", `ok-${response.id}`)]); + const finalCommand = await waitForBridgeCommandTerminal(response.id); + if (finalCommand) { + const outcome = terminalLineFromBridgeCommand(finalCommand); + setResult({ status: outcome.tone === "success" ? "succeeded" : "failed", label: outcome.text }); + appendLines([outcome]); + } else { + const timeoutLine = terminalSystemLine("warn", `桥接命令 ${response.id} 已排队，但尚未返回终态；继续观察实时日志。`, "PLATFORM", `pending-${response.id}`); + setResult({ status: "pending", label: "等待 Run 返回结果" }); + appendLines([timeoutLine]); + } } catch (error) { const label = error instanceof Error ? error.message : "命令提交失败"; setResult({ status: "failed", label }); @@ -336,13 +332,22 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu } } + async function waitForBridgeCommandTerminal(commandId: string): Promise { + for (let attempt = 0; attempt < terminalBridgeResultPollAttempts; attempt += 1) { + const current = await platformApiClient.getGameClientBridgeCommand(serverId, commandId); + if (isTerminalBridgeCommandState(current.state)) return current; + await delay(terminalBridgeResultPollMs); + } + return null; + } + return (
{serverName} - 最近历史 + {terminalLogPollMs / 1000}s 实时刷新 · 日志源 {logStreamPollMs / 1000}s 探测 · {streams.status === "ready" ? `${terminalStreams.length} 个日志源` : streams.status === "loading" ? "读取日志源" : "日志源异常"} + 最近历史 + SSE 实时推送 · {streams.status === "ready" ? `${terminalStreams.length} 个日志源` : streams.status === "loading" ? "读取日志源" : "日志源异常"}
@@ -386,39 +391,35 @@ function levelClass(level?: string): string { return "log-level-info"; } -function initialLogCursor(stream: LogStreamResponse, historyWindow: number): number { - return Math.max(0, stream.latestSeq - historyWindow); -} - -function nextCursorSeq(afterSeq: number, entries: LogEntryBody[], nextSeq: number): number { - if (entries.length === 0) return Math.max(afterSeq, nextSeq); - return Math.max(afterSeq, nextSeq, entries[entries.length - 1]?.seq ?? afterSeq); -} - -function updateCursor(current: Record, streamId: string, cursor: number): Record { - const nextCursor = Math.max(current[streamId] ?? 0, cursor); - if (nextCursor === current[streamId]) return current; - return { ...current, [streamId]: nextCursor }; -} - function terminalQuickCommandsForPlugin(pluginId: string): TerminalQuickCommand[] { return terminalQuickCommandCatalog[pluginId] ?? []; } +function bridgeCommandDispatchLabel(state: string, commandId: string): string { + return `已${state === "pending" ? "排队" : "提交"} · 桥接命令 ${commandId}`; +} + function terminalRelevantStreams(streams: LogStreamResponse[]): LogStreamResponse[] { - return [...streams].sort(compareTerminalStreams).slice(0, 8); + const active = streams.filter((stream) => stream.latestSeq > 0); + const candidates = active.length > 0 ? active : streams; + return [...candidates].sort(compareTerminalStreams).slice(0, 12); } function compareTerminalStreams(a: LogStreamResponse, b: LogStreamResponse): number { - return terminalStreamRank(a) - terminalStreamRank(b) || a.streamKey.localeCompare(b.streamKey) || a.id.localeCompare(b.id); + const rank = terminalStreamRank(a) - terminalStreamRank(b); + if (rank !== 0) return rank; + const updated = (Date.parse(b.updatedAt) || 0) - (Date.parse(a.updatedAt) || 0); + if (updated !== 0) return updated; + return b.latestSeq - a.latestSeq || a.streamKey.localeCompare(b.streamKey) || a.id.localeCompare(b.id); } function terminalStreamRank(stream: LogStreamResponse): number { const key = `${stream.source}:${stream.streamKey}`.toLowerCase(); - if (key.includes("management-program")) return 0; - if (key.includes("stderr")) return 1; - if (key.includes("stdout")) return 2; - return 3; + if (stream.source === "file" || key.includes("scum.")) return 0; + if (key.includes("management-program")) return 1; + if (key.includes("stderr")) return 2; + if (key.includes("stdout")) return 3; + return 4; } function terminalLineFromLog(stream: LogStreamResponse, entry: LogEntryBody): TerminalLine { @@ -439,6 +440,34 @@ function terminalSystemLine(tone: TerminalLine["tone"], text: string, streamKey: return { id, tone, text, at: new Date(now).toLocaleTimeString(), sortKey: now, streamKey }; } +function isTerminalBridgeCommandState(state: GameClientBridgeCommandResponse["state"]): boolean { + return state === "succeeded" || state === "failed" || state === "cancelled" || state === "expired" || state === "unknown"; +} + +function terminalLineFromBridgeCommand(command: GameClientBridgeCommandResponse): TerminalLine { + const summary = command.result?.summary || command.resultSummary || command.cancellation?.reason || bridgeCommandStateLabel(command.state); + const completed = command.completedAt || command.result?.completedAt || command.cancellation?.cancelledAt || command.updatedAt; + const sortKey = Date.parse(completed) || Date.now(); + const tone: TerminalLine["tone"] = command.state === "succeeded" ? "success" : command.state === "failed" ? "error" : "warn"; + return { id: `bridge-${command.id}-${command.state}`, tone, text: `桥接命令 ${command.id} · ${bridgeCommandStateLabel(command.state)} · ${summary}`, at: new Date(sortKey).toLocaleTimeString(), sortKey, streamKey: "BRIDGE" }; +} + +function bridgeCommandStateLabel(state: GameClientBridgeCommandResponse["state"]): string { + switch (state) { + case "succeeded": return "已成功"; + case "failed": return "已失败"; + case "cancelled": return "已取消"; + case "expired": return "已过期"; + case "unknown": return "状态未知"; + case "claimed": return "Run 已领取"; + case "pending": return "已排队"; + } +} + +function delay(ms: number): Promise { + return new Promise((resolve) => window.setTimeout(resolve, ms)); +} + function terminalTone(entry: LogEntryBody): TerminalLine["tone"] { const value = `${entry.level ?? ""} ${entry.line}`.toLowerCase(); if (/\b(error|fatal|panic|exception|failed|failure)\b/.test(value)) return "error"; diff --git a/platform_web/components/SourceRCONCommandPanel.test.tsx b/platform_web/components/SourceRCONCommandPanel.test.tsx index cfc8cbf..73e362c 100644 --- a/platform_web/components/SourceRCONCommandPanel.test.tsx +++ b/platform_web/components/SourceRCONCommandPanel.test.tsx @@ -3,10 +3,11 @@ import { describe, expect, it } from "vitest"; import sourceRCONCommandPanelSource from "./SourceRCONCommandPanel.tsx?raw"; describe("SourceRCONCommandPanel", () => { - it("uses the typed dispatch API without confirmation, transcript, or connection fields", () => { - expect(sourceRCONCommandPanelSource).toContain("sendSourceRCONCommand"); - expect(sourceRCONCommandPanelSource).toContain("sourceRCONChatRequest"); - expect(sourceRCONCommandPanelSource).toContain("sourceRCONRawCommandRequest"); + it("uses protected bridge dispatch without confirmation, transcript, or connection fields", () => { + expect(sourceRCONCommandPanelSource).toContain("queueGameClientBridgeCommand"); + expect(sourceRCONCommandPanelSource).toContain("scumManagementRCONCommandRequest"); + expect(sourceRCONCommandPanelSource).toContain("scumAnnouncementCommand"); + expect(sourceRCONCommandPanelSource).not.toContain("sendSourceRCONCommand"); expect(sourceRCONCommandPanelSource).not.toContain("ConfirmDialog"); expect(sourceRCONCommandPanelSource).not.toContain("operations."); expect(sourceRCONCommandPanelSource).not.toContain("transcript"); diff --git a/platform_web/components/SourceRCONCommandPanel.tsx b/platform_web/components/SourceRCONCommandPanel.tsx index 047711c..26b9056 100644 --- a/platform_web/components/SourceRCONCommandPanel.tsx +++ b/platform_web/components/SourceRCONCommandPanel.tsx @@ -1,7 +1,7 @@ import { type FormEvent, useState } from "react"; import { platformApiClient } from "../api/client"; -import { sourceRCONChatRequest, sourceRCONRawCommandRequest } from "../schemas/sourceRcon"; +import { scumAnnouncementCommand, scumManagementRCONCommandRequest } from "../schemas/scumManagementRcon"; import { ResultBadge } from "./StateViews"; interface SourceRCONCommandPanelProps { @@ -12,28 +12,25 @@ interface SourceRCONCommandPanelProps { type DispatchState = { status: "pending" | "succeeded" | "failed"; label: string } | null; export function SourceRCONCommandPanel({ serverId, pluginId }: SourceRCONCommandPanelProps) { - const [chatType, setChatType] = useState(4); - const [chatMessage, setChatMessage] = useState(""); - const [targetSteamId, setTargetSteamId] = useState(""); + const [announcement, setAnnouncement] = useState(""); const [rawCommand, setRawCommand] = useState(""); - const [pending, setPending] = useState<"chat" | "command" | null>(null); + const [pending, setPending] = useState<"announcement" | "command" | null>(null); const [dispatch, setDispatch] = useState(null); if (pluginId !== "game.scum") { return null; } - async function sendChat(event: FormEvent) { + async function sendAnnouncement(event: FormEvent) { event.preventDefault(); - setPending("chat"); - setDispatch({ status: "pending", label: "正在提交聊天消息" }); + setPending("announcement"); + setDispatch({ status: "pending", label: "正在提交服务器公告" }); try { - const submitted = await platformApiClient.sendSourceRCONCommand(serverId, sourceRCONChatRequest(serverId, { chatType, message: chatMessage, targetSteamId })); - setChatMessage(""); - setTargetSteamId(""); - setDispatch({ status: "succeeded", label: sourceRCONDispatchLabel(submitted.jobId, submitted.status) }); + const submitted = await platformApiClient.queueGameClientBridgeCommand(serverId, scumManagementRCONCommandRequest(serverId, scumAnnouncementCommand(announcement))); + setAnnouncement(""); + setDispatch({ status: "succeeded", label: protectedRCONDispatchLabel(submitted.id, submitted.state) }); } catch (error) { - setDispatch({ status: "failed", label: error instanceof Error ? error.message : "聊天消息提交失败" }); + setDispatch({ status: "failed", label: error instanceof Error ? error.message : "服务器公告提交失败" }); } finally { setPending(null); } @@ -44,9 +41,9 @@ export function SourceRCONCommandPanel({ serverId, pluginId }: SourceRCONCommand setPending("command"); setDispatch({ status: "pending", label: "正在提交原始管理员指令" }); try { - const submitted = await platformApiClient.sendSourceRCONCommand(serverId, sourceRCONRawCommandRequest(serverId, rawCommand)); + const submitted = await platformApiClient.queueGameClientBridgeCommand(serverId, scumManagementRCONCommandRequest(serverId, rawCommand)); setRawCommand(""); - setDispatch({ status: "succeeded", label: sourceRCONDispatchLabel(submitted.jobId, submitted.status) }); + setDispatch({ status: "succeeded", label: protectedRCONDispatchLabel(submitted.id, submitted.state) }); } catch (error) { setDispatch({ status: "failed", label: error instanceof Error ? error.message : "原始管理员指令提交失败" }); } finally { @@ -55,35 +52,24 @@ export function SourceRCONCommandPanel({ serverId, pluginId }: SourceRCONCommand } return ( -
+
-

SCUM 聊天与管理员指令

-

立即派发一次性任务；只显示安全状态，不保留聊天或指令记录。

+

SCUM 公告与管理员指令

+

通过插件声明的 protected RCON 通道派发；只显示安全状态，不保留指令原文。

{dispatch && }
-
-

发送聊天

- void sendChat(event)}> -
- - -
+
+

发送公告

+ void sendAnnouncement(event)}>