From 7d9c1b7e9ea077ea606b2bd47ffa38a4fbc4a387 Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Tue, 1 Sep 2026 14:26:00 +0800 Subject: [PATCH] Make SCUM logs live relay only --- .../game_client_bridge_companion_handlers.go | 42 ++++++ .../api/game_client_bridge_handlers_test.go | 47 +++++++ platform/api/log_events_handlers.go | 17 ++- platform/api/log_ingest_handlers_test.go | 55 ++++++++ platform/api/resource_handlers.go | 2 + platform/api/routes.md | 15 ++- platform/api/run_log_relay_handlers.go | 38 ++++++ platform/domain/game_client_bridge.go | 4 + platform/dto/game_client_bridge.go | 8 ++ platform/protocol/run-contracts.md | 27 ++-- .../service/game_client_bridge_sessions.go | 22 +++ .../game_client_bridge_sessions_test.go | 30 ++++- platform/service/game_client_bridge_test.go | 2 +- platform/service/log_events.go | 10 ++ platform/service/log_ingest.go | 59 ++++++++ platform/service/log_ingest_test.go | 39 ++++++ platform/service/resources.go | 2 + platform/validator/game_client_bridge.go | 6 + .../ServerManagementTerminalDrawer.test.tsx | 44 ++---- .../ServerManagementTerminalDrawer.tsx | 123 +---------------- .../companion/cmd/scum-companion/main.go | 19 ++- .../companion/console_collector.go | 100 ++++++++++++++ .../companion/console_collector_test.go | 62 +++++++++ .../companion/console_storage.go | 127 ++++++++++++++++++ .../companion/log_stream.go | 125 +++++++++++++++++ .../companion/log_stream_test.go | 66 +++++++++ .../scum-server-plugin/companion/storage.go | 27 ++++ .../companion/storage_test.go | 68 ++++++++++ 28 files changed, 1011 insertions(+), 175 deletions(-) create mode 100644 platform/api/run_log_relay_handlers.go create mode 100644 plugins/examples/scum-server-plugin/companion/console_collector.go create mode 100644 plugins/examples/scum-server-plugin/companion/console_collector_test.go create mode 100644 plugins/examples/scum-server-plugin/companion/console_storage.go create mode 100644 plugins/examples/scum-server-plugin/companion/log_stream.go create mode 100644 plugins/examples/scum-server-plugin/companion/log_stream_test.go diff --git a/platform/api/game_client_bridge_companion_handlers.go b/platform/api/game_client_bridge_companion_handlers.go index 8fcef0c..035b555 100644 --- a/platform/api/game_client_bridge_companion_handlers.go +++ b/platform/api/game_client_bridge_companion_handlers.go @@ -6,6 +6,48 @@ import ( "browser.local/platform/dto" ) +// gameClientBridgeCompanionLogEvents godoc +// @Summary Stream live Run logs to a game companion +// @Description Authorizes a component session and relays only the current supervised process log stream to the companion. The platform does not persist log bodies on this route. +// @Tags game-client-bridge +// @Accept json +// @Produce text/event-stream +// @Param body body dto.GameClientBridgeLogStreamRequest true "Component log stream request" +// @Success 200 {object} dto.LogStreamEventResponse +// @Failure 400 {object} dto.ErrorResponse +// @Failure 401 {object} dto.ErrorResponse +// @Failure 403 {object} dto.ErrorResponse +// @Failure 405 {object} dto.ErrorResponse +// @Router /api/v1/game-client-bridge/companion/logs/events [post] +func (h *coreHandlers) gameClientBridgeCompanionLogEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeMethodNotAllowed(w, http.MethodPost) + return + } + request, err := decodeJSON[dto.GameClientBridgeLogStreamRequest](r) + if err != nil { + writeDecodeError(w, err) + return + } + instance, err := h.core.AuthorizeGameClientBridgeLogStream(request.ToDomain()) + if err != nil { + writeServiceError(w, err) + return + } + subscription, err := h.core.SubscribeLogEvents(instance.ID) + if err != nil { + writeServiceError(w, err) + return + } + defer subscription.Close() + streams, liveEligible, err := h.loadLiveLogSnapshot(instance.ID) + if err != nil { + writeServiceError(w, err) + return + } + h.streamCurrentLogEvents(w, r, instance, streams, liveEligible, subscription) +} + func (h *coreHandlers) gameClientBridgeCompanionClaim(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeMethodNotAllowed(w, http.MethodPost) diff --git a/platform/api/game_client_bridge_handlers_test.go b/platform/api/game_client_bridge_handlers_test.go index 0c29eab..8e09a43 100644 --- a/platform/api/game_client_bridge_handlers_test.go +++ b/platform/api/game_client_bridge_handlers_test.go @@ -1,7 +1,10 @@ package api import ( + "bufio" + "context" "net/http" + "net/http/httptest" "strings" "testing" "time" @@ -38,6 +41,13 @@ func (core *gameClientBridgeCompanionCore) UploadGameClientBridgeSnapshot(domain return domain.CopyGameClientBridgeSnapshot(core.snapshot), nil } +func (core *gameClientBridgeCompanionCore) AuthorizeGameClientBridgeLogStream(request domain.GameClientBridgeLogStreamRequest) (domain.ServerInstance, error) { + if strings.TrimSpace(request.SessionToken) != "component-token" { + return domain.ServerInstance{}, service.ErrUnauthorized + } + return core.Core.GetServerInstance("server-1") +} + func TestGameClientBridgeOperatorRoutes(t *testing.T) { store := repo.NewMemoryStore() coreService := service.NewCoreService(store) @@ -149,3 +159,40 @@ func TestGameClientBridgeCompanionRoutesAreComponentSessionMediated(t *testing.T diagnostic := performJSON(t, router, http.MethodPost, "/api/v1/game-client-bridge/companion/diagnostics", snapshotRequest) assertStatus(t, diagnostic, http.StatusAccepted) } + +func TestGameClientBridgeCompanionLogEventsRelaysCurrentRunOutput(t *testing.T) { + coreService := service.NewCoreService(repo.NewMemoryStore()) + if err := coreService.SeedLocalPlatformAdmin(); err != nil { + t.Fatal(err) + } + core := &gameClientBridgeCompanionCore{Core: coreService} + router := NewTestRouterWithCore(core) + hello := createLogIngestAPIFixtures(t, router) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + request := httptest.NewRequest(http.MethodPost, "/api/v1/game-client-bridge/companion/logs/events", strings.NewReader("{\"sessionToken\":\"component-token\"}")).WithContext(ctx) + request.Header.Set("Content-Type", "application/json") + streamWriter, streamReader := newSSEPipeResponseWriter() + done := make(chan struct{}) + go func() { + router.ServeHTTP(streamWriter, request) + _ = streamWriter.Close() + close(done) + }() + t.Cleanup(func() { + cancel() + _ = streamReader.Close() + <-done + }) + if status := <-streamWriter.status; status != http.StatusOK { + t.Fatalf("unexpected companion SSE status: %d", status) + } + reader := bufio.NewReader(streamReader) + assertSSEEvent(t, reader, "session", "\"logSessionId\":\"session-current\"") + assertSSEEvent(t, reader, "stream", "\"id\":\"log-1\"") + assertSSEEvent(t, reader, "ready", "\"streamCount\":1") + + live := validLogBatchRequest(t, hello.SessionToken, 5, 5) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", live), http.StatusOK) + assertSSEEvent(t, reader, "log", "\"seq\":5") +} diff --git a/platform/api/log_events_handlers.go b/platform/api/log_events_handlers.go index a33bd0e..201a99c 100644 --- a/platform/api/log_events_handlers.go +++ b/platform/api/log_events_handlers.go @@ -24,13 +24,17 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { writeMethodNotAllowed(w, http.MethodGet) return } - liveBoundary := time.Now().UTC().Add(-liveLogSourceClockSkew) instance, streams, liveEligible, subscription, err := h.openLogEventSubscription(r) if err != nil { writeServiceError(w, err) return } defer subscription.Close() + h.streamCurrentLogEvents(w, r, instance, streams, liveEligible, subscription) +} + +func (h *coreHandlers) streamCurrentLogEvents(w http.ResponseWriter, r *http.Request, instance domain.ServerInstance, streams []domain.LogStream, liveEligible bool, subscription service.LogEventSubscription) { + liveBoundary := time.Now().UTC().Add(-liveLogSourceClockSkew) flusher, ok := w.(http.Flusher) if !ok { writeServiceError(w, fmt.Errorf("streaming response unsupported")) @@ -131,7 +135,7 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { if !active.contains(event.Stream) { continue } - if !sourceLogEntryIsLive(event.Entry, liveBoundary) { + if !subscriptionEvent.Live && !sourceLogEntryIsLive(event.Entry, liveBoundary) { if event.Entry.Seq > emittedThrough[event.Stream.ID] { emittedThrough[event.Stream.ID] = event.Entry.Seq } @@ -143,7 +147,10 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { return } } - if event.Entry.Seq <= emittedThrough[event.Stream.ID] { + // Live relay traffic is already a current best-effort observation. + // Do not sequence-gate it: a restarted Run intentionally starts with + // fresh in-memory sequence state and must still reach this subscriber. + if !subscriptionEvent.Live && event.Entry.Seq <= emittedThrough[event.Stream.ID] { continue } if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { @@ -228,7 +235,9 @@ func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInsta return nil, err } for _, stream := range active.streams { - emittedThrough[stream.ID] = stream.LatestSeq + // Stream metadata identifies the current channel only. The platform + // never replays its old body when a live subscriber connects. + emittedThrough[stream.ID] = 0 if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil { return nil, err } diff --git a/platform/api/log_ingest_handlers_test.go b/platform/api/log_ingest_handlers_test.go index ba7390a..8dfc552 100644 --- a/platform/api/log_ingest_handlers_test.go +++ b/platform/api/log_ingest_handlers_test.go @@ -81,6 +81,26 @@ func TestLogIngestAPIWorkflow(t *testing.T) { } } +func TestLiveLogRelayAPIForwardsWithoutStoredOutput(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + batch := validLogBatchRequest(t, hello.SessionToken, 1, 1) + + relay := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", batch) + assertStatus(t, relay, http.StatusOK) + ack := decodeBody[dto.LogBatchIngestResponse](t, relay) + if !ack.Accepted || ack.AcceptedFrom != 1 || ack.AcceptedTo != 1 { + t.Fatalf("unexpected live relay ack: %+v", ack) + } + + query := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: batch.LogStreamID, AfterSeq: 0, Limit: 10}) + assertStatus(t, query, http.StatusOK) + body := decodeBody[dto.LogStreamCursorResponse](t, query) + if len(body.Entries) != 0 || body.LatestSeq != 1 { + t.Fatalf("live relay stored platform log output: %+v", body) + } +} + func TestLogEventsSSEDoesNotReplayHistory(t *testing.T) { router := newTestRouter() hello := createLogIngestAPIFixtures(t, router) @@ -153,6 +173,41 @@ func TestLogEventsSSELiveOnlyStartsAfterSnapshotTail(t *testing.T) { assertSSEEvent(t, reader, "log", `"seq":2`) } +func TestLogEventsSSERelaysLiveBatchesWithoutSequenceGate(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events", nil).WithContext(ctx) + streamWriter, streamReader := newSSEPipeResponseWriter() + done := make(chan struct{}) + go func() { + router.ServeHTTP(streamWriter, request) + _ = streamWriter.Close() + close(done) + }() + t.Cleanup(func() { + cancel() + _ = streamReader.Close() + <-done + }) + if status := <-streamWriter.status; status != http.StatusOK { + t.Fatalf("unexpected SSE status: %d", status) + } + reader := bufio.NewReader(streamReader) + assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`) + assertSSEEvent(t, reader, "stream", `"id":"log-1"`) + assertSSEEvent(t, reader, "ready", `"streamCount":1`) + + first := validLogBatchRequest(t, hello.SessionToken, 4, 4) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", first), http.StatusOK) + assertSSEEvent(t, reader, "log", `"seq":4`) + + second := validLogBatchRequest(t, hello.SessionToken, 1, 1) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", second), http.StatusOK) + assertSSEEvent(t, reader, "log", `"seq":1`) +} + func TestLogEventsSSESkipsBufferedBackfillAfterOpen(t *testing.T) { router := newTestRouter() hello := createLogIngestAPIFixtures(t, router) diff --git a/platform/api/resource_handlers.go b/platform/api/resource_handlers.go index cb7317c..be58726 100644 --- a/platform/api/resource_handlers.go +++ b/platform/api/resource_handlers.go @@ -130,6 +130,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/run/jobs/cancel", h.requireRunSignature(h.runJobCancelPoll)) mux.HandleFunc("/api/v1/run/jobs/reconcile", h.requireRunSignature(h.runJobReconcile)) mux.HandleFunc("/api/v1/run/logs/batches", h.requireRunSignature(h.runLogBatchIngest)) + mux.HandleFunc("/api/v1/run/logs/relay", h.requireRunSignature(h.runLiveLogRelay)) mux.HandleFunc("/api/v1/run/logs/progress", h.requireRunSignature(h.runLogStreamProgress)) mux.HandleFunc("/api/v1/run/artifacts/open", h.requireRunSignature(h.runArtifactOpen)) mux.HandleFunc("/api/v1/run/artifacts/chunks", h.requireRunSignature(h.runArtifactChunkUpload)) @@ -153,6 +154,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/game-client-bridge/companion/commands/claim", h.gameClientBridgeCompanionClaim) mux.HandleFunc("/api/v1/game-client-bridge/companion/commands/{commandId}/ack", h.gameClientBridgeCompanionAck) mux.HandleFunc("/api/v1/game-client-bridge/companion/commands/{commandId}/result", h.gameClientBridgeCompanionResult) + mux.HandleFunc("/api/v1/game-client-bridge/companion/logs/events", h.gameClientBridgeCompanionLogEvents) mux.HandleFunc("/api/v1/game-client-bridge/companion/snapshots", h.gameClientBridgeCompanionSnapshot) mux.HandleFunc("/api/v1/game-client-bridge/companion/diagnostics", h.gameClientBridgeCompanionDiagnostics) } diff --git a/platform/api/routes.md b/platform/api/routes.md index 2cbcf9d..d236c23 100644 --- a/platform/api/routes.md +++ b/platform/api/routes.md @@ -154,7 +154,7 @@ Lifecycle workflow responses include accepted status, action, bounded server ins - `GET /api/v1/server-instances/{id}/dependencies`: returns the target-matched plugin/profile dependency catalog, current safe probe status/evidence, typed plan summaries, and deterministic immutable plan digests. - `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. -Server-scoped terminal log streaming (`GET /api/v1/server-instances/{id}/logs/events`) is registered for the server detail terminal drawer and emits platform-accepted live log SSE events only. It does not replay retained log entries, and it drops accepted log batches whose source entry timestamps are older than the current SSE connection after a small clock-skew allowance; the raw log list/backfill routes (`logs/live` and `logs/backfill`) remain unavailable as product APIs, and internal log ingest and cursor query remain available for run/platform maintenance flows. +Server-scoped terminal log streaming (`GET /api/v1/server-instances/{id}/logs/events`) is registered for the server detail terminal drawer and emits current live Run log relay events only. It does not replay retained platform log entries. SCUM companion streaming (`POST /api/v1/game-client-bridge/companion/logs/events`) uses the component session and receives the same current live stream so the plugin can own game-log storage, analysis, and console fan-out. Runtime distribution and client-manager APIs require the current bearer session, server visibility, plugin-declared permissions, complete runtime bindings only for actions that truly depend on external logical bindings, and platform-builder readiness. Run-side lifecycle commands separately require run endpoint capability support and use plugin-declared lifecycle actions without making manual runtime-profile binding a user prerequisite. Responses and 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. @@ -194,16 +194,17 @@ Job ack/progress/result/cancel/reconcile calls remain lightweight and independen Run file input chunks are used only for browser-staged file uploads that produce `artifact://` job inputs. The route never returns storage backend paths, browser bearer credentials, machine paths, direct sockets, or unrestricted artifact bodies. This channel is lower priority than control, job lifecycle calls, and log ingest. -## Implemented Log Ingest Actions +## Implemented Live Log Relay Actions -- `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/run/logs/relay`: accept `LogBatchIngestRequest`, validate run session and stream metadata, update current stream/session metadata, and immediately fan out entries to live subscribers without storing log bodies. +- `GET /api/v1/server-instances/{id}/logs/events`: stream the current supervised process session to the server terminal drawer without replaying old retained log bodies. +- `POST /api/v1/game-client-bridge/companion/logs/events`: stream the current supervised process session to a component-authenticated game companion so the plugin can store/analyze game logs. +- `POST /api/v1/run/logs/batches`: compatibility/internal durable ingest for older workers; current Run output should use `/run/logs/relay` instead of a local spool/cache/resend loop. - `POST /api/v1/log-streams/query`: accept `LogStreamCursorRequest` and return `LogStreamCursorResponse` with bounded ordered entries after a cursor. -Server-scoped SSE log streaming is removed from product routes. `POST /api/v1/log-streams/query` remains the bounded cursor contract for internal maintenance/debug reads. -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. +Live log relay actions carry current 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. Run must not block lifecycle/control/job progress on whether Platform or a plugin subscriber received live logs. -Platform storage is configured by `PLATFORM_STORAGE_BACKEND`. The default `file` backend writes metadata snapshots to `PLATFORM_METADATA_PATH` and log bodies to segmented files in `PLATFORM_LOG_DIR`; `memory` remains available for tests and ephemeral local runs. Relational stores such as MySQL/Postgres are reserved for metadata, stream cursors, indexes, retention state, and operational records. High-volume log bodies for hundreds or thousands of servers should use a log-optimized backend behind `LogBodyStore`, such as ClickHouse, Loki, OpenSearch/Elasticsearch, or object-storage segments. +Platform storage is configured by `PLATFORM_STORAGE_BACKEND` for platform metadata and compatibility durable-ingest bodies only. Current live relay does not persist log bodies in Platform. SCUM durable console logs and semantic events are stored by the SCUM plugin companion in plugin-owned SQL tables; raw trajectory samples keep world coordinates and do not perform projection or coordinate conversion. ## Implemented Run Artifact Actions diff --git a/platform/api/run_log_relay_handlers.go b/platform/api/run_log_relay_handlers.go new file mode 100644 index 0000000..2bf43e8 --- /dev/null +++ b/platform/api/run_log_relay_handlers.go @@ -0,0 +1,38 @@ +package api + +import ( + "net/http" + + "browser.local/platform/domain" + "browser.local/platform/dto" + "browser.local/platform/repo" +) + +type liveLogRelayCore interface { + RelayLiveLogBatch(domain.LogBatchIngest) (domain.LogBatchIngestResult, error) +} + +// runLiveLogRelay accepts current Run output and immediately fans it out to +// subscribers. It has no durable body or delivery acknowledgement contract. +func (h *coreHandlers) runLiveLogRelay(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeMethodNotAllowed(w, http.MethodPost) + return + } + core, ok := h.core.(liveLogRelayCore) + if !ok { + writeServiceError(w, repo.ErrNotFound) + return + } + request, err := decodeJSON[dto.LogBatchIngestRequest](r) + if err != nil { + writeDecodeError(w, err) + return + } + result, err := core.RelayLiveLogBatch(request.ToDomain()) + if err != nil { + writeServiceError(w, err) + return + } + writeJSON(w, http.StatusOK, dto.LogBatchIngestFromDomain(result)) +} diff --git a/platform/domain/game_client_bridge.go b/platform/domain/game_client_bridge.go index 1168536..434e680 100644 --- a/platform/domain/game_client_bridge.go +++ b/platform/domain/game_client_bridge.go @@ -330,6 +330,10 @@ type GameClientBridgeSnapshotIngestRequest struct { Retention GameClientBridgeRetention } +type GameClientBridgeLogStreamRequest struct { + SessionToken string +} + type GameClientBridgeSnapshotQuery struct { ServerInstanceID string PluginID string diff --git a/platform/dto/game_client_bridge.go b/platform/dto/game_client_bridge.go index c08c52c..9eefec6 100644 --- a/platform/dto/game_client_bridge.go +++ b/platform/dto/game_client_bridge.go @@ -45,6 +45,10 @@ type GameClientBridgeSnapshotIngestRequest struct { MaxRecords int `json:"maxRecords,omitempty"` } +type GameClientBridgeLogStreamRequest struct { + SessionToken string `json:"sessionToken"` +} + type GameClientBridgeCancelRequest struct { Reason string `json:"reason,omitempty"` } @@ -228,6 +232,10 @@ func (request GameClientBridgeSnapshotIngestRequest) ToDomain() domain.GameClien return domain.GameClientBridgeSnapshotIngestRequest{SessionToken: request.SessionToken, Type: request.Type, SchemaVersion: request.SchemaVersion, StreamKey: request.StreamKey, Sequence: request.Sequence, ObservedAt: request.ObservedAt, Payload: domain.CopyGameClientBridgePayload(request.Payload), Retention: domain.GameClientBridgeRetention{KeepForSeconds: request.KeepForSeconds, MaxRecords: request.MaxRecords}} } +func (request GameClientBridgeLogStreamRequest) ToDomain() domain.GameClientBridgeLogStreamRequest { + return domain.GameClientBridgeLogStreamRequest{SessionToken: request.SessionToken} +} + func (query GameClientBridgeSnapshotQuery) ToDomain(serverID, pluginID string) domain.GameClientBridgeSnapshotQuery { return domain.GameClientBridgeSnapshotQuery{ServerInstanceID: serverID, PluginID: pluginID, ProfileKey: query.ProfileKey, Type: query.Type, StreamKey: query.StreamKey, ObservedAfter: query.ObservedAfter, Limit: query.Limit} } diff --git a/platform/protocol/run-contracts.md b/platform/protocol/run-contracts.md index 7e4122f..71fa1a8 100644 --- a/platform/protocol/run-contracts.md +++ b/platform/protocol/run-contracts.md @@ -23,7 +23,7 @@ Named control DTOs: Control payloads must remain small and must not include logs, artifact chunks, host paths, raw credentials, direct sockets, job assignments, execution input, or long task results. Hello creates or updates run endpoint metadata and issues an in-memory platform session token. A server-scoped generated Run must use the endpoint identity reserved for its server; Platform rejects a valid component key presented for another endpoint. Registration is binding/authentication only for generated Run bootstrap and must not enqueue lifecycle or status jobs merely because Run appeared. Heartbeat requires that active session token and may request capability refresh when the fingerprint changes. The control event stream is a signed Run-only `text/event-stream` wake channel; events such as `job.changed` only tell Run to claim durable work through `/run/jobs/claim`. -Capacity reports include bounded `maxJobs`, `runningJobs`, `queuedJobs`, `logBacklogBatches`, `artifactBacklogChunks`, and enumerated pressure codes. They report queue and spool counts only, never log bodies, artifact chunks, machine paths, PIDs, sockets, credentials, or transport endpoints. +Capacity reports include bounded `maxJobs`, `runningJobs`, `queuedJobs`, compatibility `logBacklogBatches`, `artifactBacklogChunks`, and enumerated pressure codes. Current Run implementations must not use `logBacklogBatches` as a durable live-log spool; capacity reports never include log bodies, artifact chunks, machine paths, PIDs, sockets, credentials, or transport endpoints. Control is the highest-priority run/platform path. Artifact/file transfer load must not delay heartbeat acceptance or mutate heartbeat capacity state through heavy payload fields. @@ -65,31 +65,34 @@ The plan is build input for the generated package, not a machine-side job-channe Autonomous lifecycle reports use `POST /api/v1/run/lifecycle/report` with the active Run session and signed envelope when required. The route accepts only bounded terminal lifecycle facts for `process.install`, `process.start`, `process.stop`, or `process.status`; it validates the server/run binding, records bounded evidence, and projects server state from Run-reported process facts without creating or completing a Platform job. A managed-process report includes an opaque `managedProcessId`, monotonic `observationSeq`, and `observedAt`; retries are idempotent and a lower sequence cannot regress a newer fact for that process. -## Log Ingest +## Live Log Relay And Compatibility Log Ingest Implemented HTTP JSON routes: +- `POST /api/v1/run/logs/relay` - `POST /api/v1/run/logs/batches` +- `GET /api/v1/server-instances/{id}/logs/events` +- `POST /api/v1/game-client-bridge/companion/logs/events` - `POST /api/v1/log-streams/query` Named log DTOs: - `LogBatchIngestRequest` - `LogBatchIngestResponse` -- `RunLogStreamProgressRequest` / `RunLogStreamProgressResponse`: signed Run-only sequence recovery for a server-bound `run...*` stream. The response contains only the latest acknowledged sequence. +- `RunLogStreamProgressRequest` / `RunLogStreamProgressResponse`: compatibility signed Run-only cursor metadata for older durable-ingest streams. Current live relay does not depend on this route for delivery or progress. - `LogEntry` - `LogStreamCursorRequest` - `LogStreamCursorResponse` - `LogStreamEventResponse` -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. +Current Run output uses `POST /api/v1/run/logs/relay`. The route validates the active Run session and stream binding, updates only bounded stream metadata such as latest sequence/session identity, and immediately fans entries out to live subscribers. It does not persist log bodies, does not create a resend backlog, does not require platform sequence acknowledgements for progress, and does not block lifecycle/control/job work on delivery success. -Run-assigned Platform jobs use `job..` log stream IDs. Autonomous lifecycle bootstrap is Run-owned machine execution rather than a Platform job, so its durable process logs use `run...`. Platform may auto-create those streams only after validating the active Run session and the server-to-Run binding. For retry compatibility, legacy spooled `job.autonomous-*.` batches are accepted as Run-owned autonomous streams without creating or completing a Platform job. +Server terminal streaming uses `GET /api/v1/server-instances/{id}/logs/events`; SCUM companion streaming uses `POST /api/v1/game-client-bridge/companion/logs/events` with the component session token in the typed JSON body. Both emit only the current supervised process session and do not replay retained platform log bodies. Platform is the live relay only. The game plugin/companion owns game-log storage, semantic analysis, and console-page fan-out behavior. -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. +`POST /api/v1/run/logs/batches` and `POST /api/v1/log-streams/query` remain compatibility/internal maintenance contracts for older durable-ingest flows. New Run workers must not use durable local log spool/cache/resend semantics for current supervised stdout/stderr. Log payloads must not carry artifact chunks, host paths, raw credentials, direct sockets, or unbounded inline data. -The platform stores log stream metadata through `repo.Store` and stores log bodies through the configured `LogBodyStore`. The default `file` backend persists platform metadata to `PLATFORM_METADATA_PATH` and appends log entries to segmented JSONL files under `PLATFORM_LOG_DIR`; the `memory` backend is only for tests and disposable local development. MySQL/Postgres are appropriate for platform metadata, stream state, retention policy, indexes, and operational records, but should not be the primary row-per-log-line store for hundreds or thousands of servers. Production log bodies should move behind the same boundary to append/query backends such as ClickHouse, Loki, OpenSearch/Elasticsearch, or object-storage segments with compact indexes. +The platform stores log stream metadata through `repo.Store`. Game-specific durable log bodies and derived semantic records belong to the owning game plugin; for SCUM that storage uses plugin-owned SQL tables and stores raw world coordinates without map projection or coordinate conversion. ## Artifact @@ -112,9 +115,9 @@ Named artifact DTOs: - `ArtifactTransferCompleteResponse` - `ArtifactResponse` -Artifact upload supports active run session validation, job/server-instance owner scoping, bounded JSON chunk payloads, per-chunk checksum validation, duplicate chunk acknowledgement, resume status, and final checksum verification before an artifact becomes available. Artifact transport is separate from control, job result, log ingest, plugin bridge, and browser file APIs. +Artifact upload supports active run session validation, job/server-instance owner scoping, bounded JSON chunk payloads, per-chunk checksum validation, duplicate chunk acknowledgement, resume status, and final checksum verification before an artifact becomes available. Artifact transport is separate from control, job result, live log relay, plugin bridge, and browser file APIs. -Artifact/file transfer is the lower-priority heavy channel. Chunk upload and completion must not block control heartbeat, job ack/result delivery, cancellation/reconcile calls, or log ingest acknowledgement. Lightweight routes must reject heavy transfer payloads instead of accepting or storing them. +Artifact/file transfer is the lower-priority heavy channel. Chunk upload and completion must not block control heartbeat, job ack/result delivery, cancellation/reconcile calls, or current live log relay. Lightweight routes must reject heavy transfer payloads instead of accepting or storing them. ## Server File Manager Transfer @@ -128,7 +131,7 @@ Browser-facing file management uses server-instance scoped routes on Platform fo Browser uploads are first staged as server-instance artifacts. Platform then queues a `files.write` job whose `inputRef` is `artifact://` and whose execution input names the dedicated `run-file-transfer` channel. Run pulls those bytes through `POST /api/v1/run/files/input-chunk` while proving the active endpoint session plus job attempt and lease. The chunk route is fenced to the active file-write job, validates artifact ownership/checksum, and returns bounded byte ranges only. -File-manager transfer is a separate, low-priority heavy path. Slow uploads, downloads, retries, or file input chunk pulls must not block control heartbeat, job claim/ack/progress/result/cancel/reconcile, durable log batch ingest, or artifact upload acknowledgements. Control, jobs, logs, artifacts, file transfer, and optional game-client bridge remain independently backpressured channels. +File-manager transfer is a separate, low-priority heavy path. Slow uploads, downloads, retries, or file input chunk pulls must not block control heartbeat, job claim/ack/progress/result/cancel/reconcile, current live log relay, or artifact upload acknowledgements. Control, jobs, logs, artifacts, file transfer, and optional game-client bridge remain independently backpressured channels. ## Client Manager lifecycle channel @@ -136,8 +139,8 @@ Client Manager lifecycle jobs use the independent capabilities `client-manager.d Run materializes the declared output such as `config.yaml` from the fenced values and its own configured Platform control URL. Source template values are not credentials and must not override the generated component identity or policy. The lifecycle input never contains the component proof itself, a component session, a browser credential, a host path, or a direct socket; proof remains inside the component package and is supplied to the supervised process only through the declared environment-variable name. -Run persists staging/active/previous slots and a local journal. It rejects stale lease/attempt/generation/target fences, traversal/link/device-file archives, checksum mismatches, undeclared executables, and arbitrary shell. Terminal results use `client-manager.deployed`, `client-manager.control`, `client-manager.updated`, `client-manager.rolled-back`, `client-manager.rollback.restored`, or `client-manager.uninstalled` with logical process/health state only. A stalled Client Manager download must not delay Run heartbeat, job ack/result/cancel, or log spool acknowledgement. +Run persists staging/active/previous slots and a local journal. It rejects stale lease/attempt/generation/target fences, traversal/link/device-file archives, checksum mismatches, undeclared executables, and arbitrary shell. Terminal results use `client-manager.deployed`, `client-manager.control`, `client-manager.updated`, `client-manager.rolled-back`, `client-manager.rollback.restored`, or `client-manager.uninstalled` with logical process/health state only. A stalled Client Manager download must not delay Run heartbeat, job ack/result/cancel, or current live log relay. ## Game Client Bridge -The optional game client bridge is separate from run lifecycle, control registration, job handling, log ingest, and artifact transport. +The optional game client bridge is separate from run lifecycle, control registration, job handling, compatibility log ingest, and artifact transport. A plugin companion may subscribe to the current live log relay through its component session so it can own game-log storage and analysis without requiring Platform to persist or interpret game log bodies. diff --git a/platform/service/game_client_bridge_sessions.go b/platform/service/game_client_bridge_sessions.go index 01b1b9a..a75b6f8 100644 --- a/platform/service/game_client_bridge_sessions.go +++ b/platform/service/game_client_bridge_sessions.go @@ -13,6 +13,7 @@ import ( ) const gameClientBridgeCapability = "game-client.bridge" +const gameClientBridgeLogStreamCapability = "logs.stream" func (svc *CoreService) ClaimGameClientBridgeCommands(request domain.GameClientBridgeClaimRequest) ([]domain.GameClientBridgeCommand, error) { if err := validator.ValidateGameClientBridgeClaimRequest(request); err != nil { @@ -127,6 +128,27 @@ func (svc *CoreService) UploadGameClientBridgeSnapshot(request domain.GameClient return domain.CopyGameClientBridgeSnapshot(snapshot), nil } +func (svc *CoreService) AuthorizeGameClientBridgeLogStream(request domain.GameClientBridgeLogStreamRequest) (domain.ServerInstance, error) { + if err := validator.ValidateGameClientBridgeLogStreamRequest(request); err != nil { + return domain.ServerInstance{}, err + } + component, err := svc.authorizeGameClientBridgeSession(request.SessionToken) + if err != nil { + return domain.ServerInstance{}, err + } + if !containsString(component.Session.Capabilities, gameClientBridgeLogStreamCapability) { + return domain.ServerInstance{}, ErrForbidden + } + instance, err := svc.store.ServerInstances().Get(component.Session.ServerInstanceID) + if err != nil { + return domain.ServerInstance{}, err + } + if instance.PluginID != component.Installation.PluginID { + return domain.ServerInstance{}, ErrUnauthorized + } + return domain.CopyServerInstance(instance), nil +} + func gameClientBridgeSnapshotDeclaration(plugin domain.GamePlugin, snapshotType, schemaVersion string) (domain.GameClientBridgeSnapshotDeclaration, bool) { for _, declaration := range plugin.GameClientBridge.Snapshots { if declaration.Type == snapshotType && declaration.SchemaVersion == schemaVersion { diff --git a/platform/service/game_client_bridge_sessions_test.go b/platform/service/game_client_bridge_sessions_test.go index d23ca98..d777677 100644 --- a/platform/service/game_client_bridge_sessions_test.go +++ b/platform/service/game_client_bridge_sessions_test.go @@ -11,7 +11,7 @@ import ( func seedGameClientBridgeComponentSession(t *testing.T, svc *CoreService, now time.Time, token string) (domain.ClientManagerInstallation, domain.ClientManagerSession) { t.Helper() installation := domain.ClientManagerInstallation{ID: "installation-1", ServerInstanceID: "server-1", PluginID: "game.scum", ProfileKey: "scum-client", RunEndpointID: "run-1", Status: domain.ClientManagerLifecycleOnline, ActiveArtifactID: "artifact-1", KeyGeneration: 2, DeploymentGeneration: 3} - session := domain.ClientManagerSession{ID: "component-session-1", InstallationID: installation.ID, ServerInstanceID: installation.ServerInstanceID, ProfileKey: installation.ProfileKey, RunEndpointID: installation.RunEndpointID, ArtifactID: installation.ActiveArtifactID, KeyGeneration: installation.KeyGeneration, DeploymentGeneration: installation.DeploymentGeneration, TokenHash: tokenHash(token), Capabilities: []string{"component.heartbeat", gameClientBridgeCapability}, Status: domain.ClientManagerSessionActive, ExpiresAt: now.Add(time.Hour)} + session := domain.ClientManagerSession{ID: "component-session-1", InstallationID: installation.ID, ServerInstanceID: installation.ServerInstanceID, ProfileKey: installation.ProfileKey, RunEndpointID: installation.RunEndpointID, ArtifactID: installation.ActiveArtifactID, KeyGeneration: installation.KeyGeneration, DeploymentGeneration: installation.DeploymentGeneration, TokenHash: tokenHash(token), Capabilities: []string{"component.heartbeat", gameClientBridgeCapability, gameClientBridgeLogStreamCapability}, Status: domain.ClientManagerSessionActive, ExpiresAt: now.Add(time.Hour)} key := domain.EncryptedComponentKey{ID: "key-1", ServerInstanceID: installation.ServerInstanceID, ComponentKind: domain.DistributionComponentClientManager, ComponentKey: installation.ProfileKey, Generation: installation.KeyGeneration, Status: domain.ComponentKeyStatusActive} if err := svc.store.ClientManagerInstallations().Create(installation); err != nil { t.Fatal(err) @@ -80,6 +80,34 @@ func TestGameClientBridgeComponentSessionAuthorizesCommandsAndSnapshots(t *testi } } +func TestGameClientBridgeComponentSessionAuthorizesLiveLogStream(t *testing.T) { + svc, clock := newGameClientBridgeService(t) + const token = "component-session-token" + _, session := seedGameClientBridgeComponentSession(t, svc, *clock, token) + if err := svc.store.RunEndpoints().Create(domain.RunEndpoint{ID: session.RunEndpointID, DisplayName: "Component Run", Status: domain.RunEndpointStatusOnline, Capabilities: []string{"process.start", "logs.read"}, Capacity: domain.RunCapacity{MaxJobs: 4}, LastHeartbeatAt: *clock}); err != nil { + t.Fatalf("seed run endpoint: %v", err) + } + server := domain.ServerInstance{ID: session.ServerInstanceID, PluginID: "game.scum", PluginVersion: "1.0.0", RunEndpointID: session.RunEndpointID, Name: "SCUM"} + if err := svc.store.ServerInstances().Create(server); err != nil { + t.Fatalf("create server instance: %v", err) + } + instance, err := svc.AuthorizeGameClientBridgeLogStream(domain.GameClientBridgeLogStreamRequest{SessionToken: token}) + if err != nil || instance.ID != server.ID || instance.PluginID != server.PluginID { + t.Fatalf("authorize companion log stream: instance=%#v err=%v", instance, err) + } + if instance.RunEndpointID != session.RunEndpointID { + t.Fatalf("authorized stream was not bound to component server: %#v", instance) + } + + session.Capabilities = []string{"component.heartbeat", gameClientBridgeCapability} + if err := svc.store.ClientManagerSessions().Update(session); err != nil { + t.Fatal(err) + } + if _, err := svc.AuthorizeGameClientBridgeLogStream(domain.GameClientBridgeLogStreamRequest{SessionToken: token}); !errors.Is(err, ErrForbidden) { + t.Fatalf("expected missing logs.stream rejection, got %v", err) + } +} + func TestGameClientBridgeClaimCannotBeCompletedByAnotherCurrentSession(t *testing.T) { svc, clock := newGameClientBridgeService(t) const firstToken = "component-session-token-one" diff --git a/platform/service/game_client_bridge_test.go b/platform/service/game_client_bridge_test.go index c7a369a..59e5ebd 100644 --- a/platform/service/game_client_bridge_test.go +++ b/platform/service/game_client_bridge_test.go @@ -12,7 +12,7 @@ func newGameClientBridgeService(t *testing.T) (*CoreService, *time.Time) { t.Helper() now := time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC) store := repo.NewMemoryStore() - plugin := domain.GamePlugin{ID: "game.scum", RuntimeProfiles: domain.GamePluginRuntimeProfiles{ClientManagers: []domain.RuntimeClientManagerProfile{{Key: "scum-client", Health: domain.RuntimeClientManagerHealth{RequiredCapabilities: []string{gameClientBridgeCapability}}}}}, GameClientBridge: domain.GameClientBridgeManifest{Commands: []domain.GameClientBridgeCommandDeclaration{{Type: "diagnostic.ping", TimeoutSeconds: 600, MaxPayloadBytes: 4096}}, Snapshots: []domain.GameClientBridgeSnapshotDeclaration{{Type: "players", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 3600, MaxRecords: 100}}, {Type: "health", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 60}}, {Type: "companion.health", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 3600, MaxRecords: 100}}}, Retention: domain.GameClientBridgeRetention{KeepForSeconds: 86400, MaxRecords: 1000}}} + plugin := domain.GamePlugin{ID: "game.scum", Name: "SCUM", Version: "1.0.0", ServerType: "scum", RuntimeProfiles: domain.GamePluginRuntimeProfiles{ClientManagers: []domain.RuntimeClientManagerProfile{{Key: "scum-client", Health: domain.RuntimeClientManagerHealth{RequiredCapabilities: []string{gameClientBridgeCapability, gameClientBridgeLogStreamCapability}}}}}, GameClientBridge: domain.GameClientBridgeManifest{Commands: []domain.GameClientBridgeCommandDeclaration{{Type: "diagnostic.ping", TimeoutSeconds: 600, MaxPayloadBytes: 4096}}, Snapshots: []domain.GameClientBridgeSnapshotDeclaration{{Type: "players", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 3600, MaxRecords: 100}}, {Type: "health", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 60}}, {Type: "companion.health", SchemaVersion: "1", Retention: domain.GameClientBridgeRetention{KeepForSeconds: 3600, MaxRecords: 100}}}, Retention: domain.GameClientBridgeRetention{KeepForSeconds: 86400, MaxRecords: 1000}}} if err := store.GamePlugins().Create(plugin); err != nil { t.Fatalf("seed bridge plugin: %v", err) } diff --git a/platform/service/log_events.go b/platform/service/log_events.go index 992862d..1ad01b2 100644 --- a/platform/service/log_events.go +++ b/platform/service/log_events.go @@ -18,6 +18,7 @@ const ( type LogEventSubscriptionEvent struct { Kind LogEventSubscriptionEventKind LogEvent domain.LogStreamEvent + Live bool ServerInstanceID string ProcessState domain.ServerInstanceState } @@ -66,6 +67,14 @@ func (svc *CoreService) SubscribeLogEventsForSession(sessionID string, serverIns } func (svc *CoreService) publishLogEvents(stream domain.LogStream, entries []domain.LogEntry) { + svc.publishLogEventsWithMode(stream, entries, false) +} + +func (svc *CoreService) publishLiveLogEvents(stream domain.LogStream, entries []domain.LogEntry) { + svc.publishLogEventsWithMode(stream, entries, true) +} + +func (svc *CoreService) publishLogEventsWithMode(stream domain.LogStream, entries []domain.LogEntry, live bool) { if len(entries) == 0 { return } @@ -73,6 +82,7 @@ func (svc *CoreService) publishLogEvents(stream domain.LogStream, entries []doma for index, entry := range entries { events[index] = LogEventSubscriptionEvent{ Kind: LogEventSubscriptionEventLog, + Live: live, LogEvent: domain.CopyLogStreamEvent(domain.LogStreamEvent{ ServerInstanceID: stream.ServerInstanceID, Stream: stream, diff --git a/platform/service/log_ingest.go b/platform/service/log_ingest.go index 3532bdc..dce6416 100644 --- a/platform/service/log_ingest.go +++ b/platform/service/log_ingest.go @@ -12,6 +12,65 @@ import ( const defaultLogQueryLimit = 100 +// RelayLiveLogBatch forwards output observed by Run to live subscribers. It +// intentionally updates only stream metadata; the log body is not written to +// the platform log store. Game plugins own durable log storage and analysis. +func (svc *CoreService) RelayLiveLogBatch(batch domain.LogBatchIngest) (domain.LogBatchIngestResult, error) { + batch = domain.CopyLogBatchIngest(batch) + if err := validator.ValidateLogBatchIngest(batch); err != nil { + return domain.LogBatchIngestResult{}, err + } + if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil { + return domain.LogBatchIngestResult{}, err + } + + lock := svc.logIngestLock(batch.ServerInstanceID) + lock.Lock() + stamp := svc.now() + stream, err := svc.store.LogStreams().Get(batch.LogStreamID) + if errors.Is(err, repo.ErrNotFound) { + if repairErr := svc.ensureLogStreamForBatch(batch, stamp); repairErr != nil { + lock.Unlock() + return domain.LogBatchIngestResult{}, repairErr + } + stream, err = svc.store.LogStreams().Get(batch.LogStreamID) + } + if err != nil { + lock.Unlock() + return domain.LogBatchIngestResult{}, err + } + if err := validateLogBatchStream(batch, stream); err != nil { + lock.Unlock() + return domain.LogBatchIngestResult{}, err + } + if batch.LastSeq > stream.LatestSeq { + stream.LatestSeq = batch.LastSeq + } + stream.UpdatedAt = stamp + if err := svc.store.LogStreams().Update(stream); err != nil { + lock.Unlock() + return domain.LogBatchIngestResult{}, err + } + lock.Unlock() + + entries := domain.CopyLogEntries(batch.Entries) + // Run may be on a machine whose wall clock is skewed. Relay time is the + // authoritative observation time for this best-effort live event; using it + // keeps the SSE live boundary from treating current output as old history. + for index := range entries { + entries[index].Timestamp = stamp.Add(time.Duration(index) * time.Nanosecond) + } + svc.publishLiveLogEvents(stream, entries) + return domain.LogBatchIngestResult{ + Accepted: true, + LogStreamID: batch.LogStreamID, + AcceptedFrom: batch.FirstSeq, + AcceptedTo: batch.LastSeq, + LatestSeq: stream.LatestSeq, + ServerTime: stamp, + }, nil +} + func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogBatchIngestResult, error) { batch = domain.CopyLogBatchIngest(batch) if err := validator.ValidateLogBatchIngest(batch); err != nil { diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index c43441a..1c57a3f 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -92,6 +92,45 @@ func TestCoreServicePublishesLogEventsForAcceptedBatch(t *testing.T) { } } +func TestCoreServiceRelaysLiveBatchWithoutPersistingBody(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) + batch.Entries[0].Timestamp = time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + batch.Checksum, err = validator.LogEntriesChecksum(batch.Entries) + if err != nil { + t.Fatalf("checksum live batch: %v", err) + } + ack, err := svc.RelayLiveLogBatch(batch) + if err != nil || !ack.Accepted || ack.LatestSeq != 1 { + t.Fatalf("relay live batch: ack=%+v err=%v", ack, err) + } + query, err := svc.QueryLogStream(domain.LogStreamCursorQuery{LogStreamID: batch.LogStreamID, AfterSeq: 0, Limit: 10}) + if err != nil { + t.Fatalf("query relayed log stream: %v", err) + } + if len(query.Entries) != 0 || query.LatestSeq != 1 { + t.Fatalf("live relay wrote a platform log body: %+v", query) + } + select { + case event := <-subscription.Events: + if !event.Live { + t.Fatalf("expected live relay event marker: %+v", event) + } + if event.LogEvent.Entry.Timestamp.Before(fixedTime) { + t.Fatalf("live relay kept stale source timestamp: %+v", event.LogEvent.Entry) + } + case <-time.After(time.Second): + t.Fatal("expected relayed live log event") + } +} + func TestCoreServicePersistsAndEnforcesImmutableProcessLogSessionMetadata(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) diff --git a/platform/service/resources.go b/platform/service/resources.go index 2a43e85..4001c65 100644 --- a/platform/service/resources.go +++ b/platform/service/resources.go @@ -194,6 +194,7 @@ type Core interface { CompleteGameClientBridgeCommand(domain.GameClientBridgeResultRequest) (domain.GameClientBridgeCommand, error) CancelGameClientBridgeCommandForSession(string, domain.GameClientBridgeCancelRequest) (domain.GameClientBridgeCommand, error) UploadGameClientBridgeSnapshot(domain.GameClientBridgeSnapshotIngestRequest) (domain.GameClientBridgeSnapshot, error) + AuthorizeGameClientBridgeLogStream(domain.GameClientBridgeLogStreamRequest) (domain.ServerInstance, error) ReconcileGameClientBridgeCommands() error GetGameClientBridgeStatusForSession(string, string) (domain.GameClientBridgeStatus, error) ListGameClientBridgeCommandsForSession(string, domain.GameClientBridgeCommandFilter) ([]domain.GameClientBridgeCommand, error) @@ -221,6 +222,7 @@ type Core interface { SubscribeLogEvents(string) (LogEventSubscription, error) SubscribeLogEventsForSession(string, string) (LogEventSubscription, error) IngestLogBatch(domain.LogBatchIngest) (domain.LogBatchIngestResult, error) + RelayLiveLogBatch(domain.LogBatchIngest) (domain.LogBatchIngestResult, error) GetRunLogStreamProgress(domain.RunLogStreamProgress) (domain.RunLogStreamProgressResult, error) QueryLogStream(domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) SeedPlatformAdmin(string, string) error diff --git a/platform/validator/game_client_bridge.go b/platform/validator/game_client_bridge.go index 3af4704..fb8107f 100644 --- a/platform/validator/game_client_bridge.go +++ b/platform/validator/game_client_bridge.go @@ -112,6 +112,12 @@ func ValidateGameClientBridgeSnapshotIngestRequest(request domain.GameClientBrid return finish(violations) } +func ValidateGameClientBridgeLogStreamRequest(request domain.GameClientBridgeLogStreamRequest) error { + var violations []string + violations = appendGameClientBridgeSession(violations, request.SessionToken) + return finish(violations) +} + func ValidateGameClientBridgeSnapshotQuery(query domain.GameClientBridgeSnapshotQuery) error { var violations []string violations = appendGameClientBridgeIdentifier(violations, "serverInstanceId", query.ServerInstanceID, true) diff --git a/platform_web/components/ServerManagementTerminalDrawer.test.tsx b/platform_web/components/ServerManagementTerminalDrawer.test.tsx index 8a4a266..2d86a1d 100644 --- a/platform_web/components/ServerManagementTerminalDrawer.test.tsx +++ b/platform_web/components/ServerManagementTerminalDrawer.test.tsx @@ -86,7 +86,9 @@ describe("ServerManagementTerminalDrawer", () => { expect(Array.from(container?.querySelectorAll(".terminal-text") ?? []).filter((node) => node.textContent === "generation A live output")).toHaveLength(1); expect(apiMocks.openServerLogEvents).toHaveBeenCalledTimes(1); expect(apiMocks.listLogStreams).not.toHaveBeenCalled(); + expect(apiMocks.queryLogStream).not.toHaveBeenCalled(); expect(apiMocks.dispatchSourceRCONCommand).not.toHaveBeenCalled(); + expect(container?.textContent).not.toContain("查看历史"); }); it("uses the server-provided clock for terminal system lines", async () => { @@ -99,7 +101,7 @@ describe("ServerManagementTerminalDrawer", () => { expect(systemLine?.querySelector("time")?.textContent).toBe(formatTerminalServerTime(serverTime)); }); - it("loads the current session tail when stream metadata arrives", async () => { + it("does not load stored output when current stream metadata arrives", async () => { const stream = logStream("stdout-current", "session-current", "process.stdout"); stream.latestSeq = 42; apiMocks.queryLogStream.mockResolvedValue({ logStreamId: stream.id, entries: [logEntry(41, "tail before drawer opened"), logEntry(42, "latest stored output")], nextSeq: 42, latestSeq: 42 }); @@ -109,9 +111,9 @@ describe("ServerManagementTerminalDrawer", () => { await emitStream(stream); await flushPromises(); - expect(container?.textContent).toContain("tail before drawer opened"); - expect(container?.textContent).toContain("latest stored output"); - expect(apiMocks.queryLogStream).toHaveBeenCalledWith({ logStreamId: stream.id, afterSeq: 0, limit: 500 }); + expect(container?.textContent).not.toContain("tail before drawer opened"); + expect(container?.textContent).not.toContain("latest stored output"); + expect(apiMocks.queryLogStream).not.toHaveBeenCalled(); }); it("renders an empty current session without accepting unrelated or sessionless logs", async () => { @@ -164,36 +166,17 @@ describe("ServerManagementTerminalDrawer", () => { expect(apiMocks.openServerLogEvents).toHaveBeenCalledTimes(1); }); - it("keeps selected historical output separate while live output continues in the background", async () => { - const oldStream = logStream("stdout-old", "session-old", "process.stdout", "2026-08-01T00:00:00Z"); - oldStream.latestSeq = 900; - apiMocks.listLogStreams.mockResolvedValue({ items: [oldStream], count: 1 }); - apiMocks.queryLogStream.mockImplementation((request: { logStreamId: string }) => Promise.resolve(request.logStreamId === oldStream.id - ? { logStreamId: oldStream.id, entries: [logEntry(1, "selected historical output", "2026-08-01T00:00:01Z")], nextSeq: 1, latestSeq: 1 } - : { logStreamId: request.logStreamId, entries: [], nextSeq: 0, latestSeq: 0 })); + it("keeps the terminal as a pure current live stream and never opens platform history", async () => { await renderDrawer(); await emitSession("session-current"); await emitStream(logStream("stdout-current", "session-current", "process.stdout")); await emitLog("stdout-current", "session-current", "process.stdout", logEntry(1, "current live output")); - await clickButton("查看历史"); - await flushPromises(); - const select = container?.querySelector('select[aria-label="选择历史日志流"]'); - if (!select) throw new Error("history stream selector not found"); - await act(async () => setSelectValue(select, oldStream.id)); - await flushPromises(); - - expect(container?.textContent).toContain("selected historical output"); - expect(container?.textContent).not.toContain("current live output"); - await emitLog("stdout-current", "session-current", "process.stdout", logEntry(2, "new live output while viewing history")); - expect(container?.textContent).not.toContain("new live output while viewing history"); - - await clickButton("实时输出"); expect(container?.textContent).toContain("current live output"); - expect(container?.textContent).toContain("new live output while viewing history"); - expect(container?.textContent).not.toContain("selected historical output"); - expect(apiMocks.listLogStreams).toHaveBeenCalledWith("server-1"); - expect(apiMocks.queryLogStream).toHaveBeenCalledWith({ logStreamId: oldStream.id, afterSeq: 400, limit: 500 }); + expect(container?.textContent).not.toContain("查看历史"); + expect(container?.querySelector('select[aria-label="选择历史日志流"]')).toBeNull(); + expect(apiMocks.listLogStreams).not.toHaveBeenCalled(); + expect(apiMocks.queryLogStream).not.toHaveBeenCalled(); expect(apiMocks.dispatchSourceRCONCommand).not.toHaveBeenCalled(); }); @@ -262,11 +245,6 @@ function setInputValue(input: HTMLInputElement, value: string) { input.dispatchEvent(new Event("input", { bubbles: true })); } -function setSelectValue(select: HTMLSelectElement, value: string) { - Object.getOwnPropertyDescriptor(HTMLSelectElement.prototype, "value")?.set?.call(select, value); - select.dispatchEvent(new Event("change", { bubbles: true })); -} - function logStream(id: string, logSessionId: string, streamKey: string, updatedAt = "2026-08-14T00:00:00Z"): LogStreamResponse { return { id, serverInstanceId: "server-1", source: "process", streamKey, logSessionId, sessionStartedAt: updatedAt, latestSeq: 1, storageBackend: "database", retentionPolicy: "default", createdAt: updatedAt, updatedAt }; } diff --git a/platform_web/components/ServerManagementTerminalDrawer.tsx b/platform_web/components/ServerManagementTerminalDrawer.tsx index c433225..f98d6a9 100644 --- a/platform_web/components/ServerManagementTerminalDrawer.tsx +++ b/platform_web/components/ServerManagementTerminalDrawer.tsx @@ -1,4 +1,4 @@ -import { History, ListChecks, Send, Sparkles, Terminal, Trash2, X } from "lucide-react"; +import { ListChecks, Send, Sparkles, Terminal, Trash2, X } from "lucide-react"; import { type FormEvent, type KeyboardEvent as ReactKeyboardEvent, type ReactNode, useCallback, useEffect, useMemo, useRef, useState } from "react"; import { platformApiClient } from "../api/client"; @@ -10,13 +10,11 @@ import { formatTerminalLogTime, formatTerminalServerTime } from "../utils/logTim import { EmptyState, ResultBadge } from "./StateViews"; type LoadState = { status: "loading" } | { status: "error"; reason: string } | { status: "ready"; data: T }; -type HistoryLineState = { status: "idle" } | LoadState; 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 terminalJobResultPollMs = 1000; const terminalJobResultPollAttempts = 30; -const terminalHistoryWindow = 500; const maxTerminalLines = 10000; const terminalQuickCommandCatalog: Record = { "game.scum": [ @@ -92,17 +90,11 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu const [result, setResult] = useState<{ status: "pending" | "succeeded" | "failed"; label: string } | null>(null); const [followLatest, setFollowLatest] = useState(true); const [liveSessionId, setLiveSessionId] = useState(null); - const [historyOpen, setHistoryOpen] = useState(false); - const [historyStreams, setHistoryStreams] = useState>({ status: "loading" }); - const [historyLines, setHistoryLines] = useState({ status: "idle" }); - const [selectedHistoryStreamId, setSelectedHistoryStreamId] = useState(""); const outputRef = useRef(null); const followLatestRef = useRef(true); const serverTimeRef = useRef(undefined); const initialHistoryPendingRef = useRef(false); const liveSessionRef = useRef(undefined); - const historyRequestRef = useRef(0); - const liveTailLoadedRef = useRef(new Map()); const quickCommands = useMemo(() => terminalQuickCommandsForPlugin(pluginId), [pluginId]); const supportsCommands = quickCommands.length > 0; @@ -125,22 +117,6 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu }); }, []); - const loadLiveStreamTail = useCallback((stream: LogStreamResponse) => { - const latestSeq = Number(stream.latestSeq); - if (!Number.isFinite(latestSeq) || latestSeq <= 0) return; - const loadedThrough = liveTailLoadedRef.current.get(stream.id) ?? 0; - if (loadedThrough >= latestSeq) return; - liveTailLoadedRef.current.set(stream.id, latestSeq); - void platformApiClient.queryLogStream({ logStreamId: stream.id, afterSeq: Math.max(0, latestSeq - terminalHistoryWindow), limit: terminalHistoryWindow }).then((response) => { - if (!eventBelongsToLiveSession(stream.logSessionId, liveSessionRef.current)) return; - appendLines(response.entries.map((entry) => terminalLineFromLog({ ...stream, latestSeq: response.latestSeq }, entry))); - lockTerminalFollow(); - }).catch((error) => { - if (!eventBelongsToLiveSession(stream.logSessionId, liveSessionRef.current)) return; - appendLines([terminalSystemLine("warn", `当前会话尾部日志读取失败，等待后续实时输出：${error instanceof Error ? error.message : "未知错误"}`, "SYSTEM", `tail-error-${stream.id}`, serverTimeRef.current)]); - }); - }, [appendLines, lockTerminalFollow]); - useEffect(() => { if (!open) return; setStreams({ status: "loading" }); @@ -149,14 +125,8 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu setResult(null); setHistoryIndex(null); setLiveSessionId(null); - setHistoryOpen(false); - setHistoryStreams({ status: "loading" }); - setHistoryLines({ status: "idle" }); - setSelectedHistoryStreamId(""); liveSessionRef.current = undefined; - liveTailLoadedRef.current.clear(); serverTimeRef.current = undefined; - historyRequestRef.current += 1; initialHistoryPendingRef.current = true; followLatestRef.current = true; setFollowLatest(true); @@ -189,7 +159,7 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu setStreams({ status: "ready", data: [] }); setLines((current) => mergeTerminalLines(current, [nextSessionId ? terminalSystemLine("info", previousSessionId === undefined ? "已跟随当前受管进程输出会话。" : "Run 已切换到新的受管进程输出会话。", "SYSTEM", `session-${nextSessionId}`, serverTimeRef.current) - : terminalSystemLine("warn", "当前没有可跟随的受管进程输出；旧日志可从历史查看。", "SYSTEM", "session-empty", serverTimeRef.current) + : terminalSystemLine("warn", "当前没有可跟随的受管进程输出。", "SYSTEM", "session-empty", serverTimeRef.current) ])); lockTerminalFollow(); }); @@ -198,7 +168,6 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu if (!stream || !eventBelongsToLiveSession(stream.logSessionId, liveSessionRef.current)) return; ready = true; setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); - loadLiveStreamTail(stream); }); events.addEventListener("ready", () => { ready = true; @@ -217,36 +186,7 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu if (!ready) setStreams({ status: "error", reason: "实时日志推送连接失败" }); }; return () => events.close(); - }, [appendLines, loadLiveStreamTail, lockTerminalFollow, open, serverId]); - - useEffect(() => { - if (!open || !historyOpen) return; - let cancelled = false; - setHistoryStreams({ status: "loading" }); - void platformApiClient.listLogStreams(serverId).then((response) => { - if (!cancelled) setHistoryStreams({ status: "ready", data: [...response.items].sort((left, right) => Date.parse(right.updatedAt) - Date.parse(left.updatedAt)) }); - }).catch((error) => { - if (!cancelled) setHistoryStreams({ status: "error", reason: error instanceof Error ? error.message : "历史日志列表加载失败" }); - }); - return () => { cancelled = true; }; - }, [historyOpen, open, serverId]); - - async function selectHistoryStream(streamId: string) { - const stream = historyStreams.status === "ready" ? historyStreams.data.find((item) => item.id === streamId) : undefined; - if (!stream) return; - const requestId = historyRequestRef.current + 1; - historyRequestRef.current = requestId; - setSelectedHistoryStreamId(streamId); - setHistoryLines({ status: "loading" }); - try { - const response = await platformApiClient.queryLogStream({ logStreamId: streamId, afterSeq: Math.max(0, stream.latestSeq - terminalHistoryWindow), limit: terminalHistoryWindow }); - if (historyRequestRef.current !== requestId) return; - setHistoryLines({ status: "ready", data: response.entries.map((entry) => terminalLineFromLog(stream, entry)) }); - } catch (error) { - if (historyRequestRef.current !== requestId) return; - setHistoryLines({ status: "error", reason: error instanceof Error ? error.message : "历史日志加载失败" }); - } - } + }, [appendLines, lockTerminalFollow, open, serverId]); function selectQuickCommand(item: TerminalQuickCommand) { setCommand(item.command); @@ -276,20 +216,9 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu } function clearTerminalBuffer() { - if (historyOpen) { - setHistoryLines({ status: "ready", data: [] }); - return; - } setLines([]); } - function toggleHistory() { - historyRequestRef.current += 1; - setHistoryOpen((current) => !current); - setSelectedHistoryStreamId(""); - setHistoryLines({ status: "idle" }); - } - function handleTerminalScroll() { const output = outputRef.current; if (!output || initialHistoryPendingRef.current) return; @@ -347,19 +276,17 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu
{serverName} - {historyOpen ? "历史日志（独立于实时终端）" : `当前受管进程会话${liveSessionId ? " · SSE 实时推送" : " · 等待 Run 输出"}`} · {streams.status === "ready" ? "已连接" : streams.status === "loading" ? "连接日志流" : "日志流异常"} · {followLatest ? "自动置底" : "已解锁滚动"} + {`当前受管进程会话${liveSessionId ? " · SSE 实时推送" : " · 等待 Run 输出"}`} · {streams.status === "ready" ? "已连接" : streams.status === "loading" ? "连接日志流" : "日志流异常"} · {followLatest ? "自动置底" : "已解锁滚动"}
-
- {!historyOpen && streams.status === "error" &&
{streams.reason}
} - {historyOpen && } - {!historyOpen && streams.status === "ready" && lines.length === 0 &&
{liveSessionId ? "当前受管进程会话暂无输出，后续输出会自动追加。" : "当前没有可跟随的受管进程输出；旧日志可从历史查看。"}
} - {!historyOpen && lines.map((line) =>
{line.text}
)} + {streams.status === "error" &&
{streams.reason}
} + {streams.status === "ready" && lines.length === 0 &&
{liveSessionId ? "当前受管进程会话暂无输出，后续输出会自动追加。" : "当前没有可跟随的受管进程输出。"}
} + {lines.map((line) =>
{line.text}
)}
@@ -386,42 +313,6 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu ); } -interface HistoryLogViewProps { - streams: LoadState; - lines: HistoryLineState; - selectedStreamId: string; - onSelect: (streamId: string) => Promise; - serverTime?: string; -} - -function HistoryLogView({ streams, lines, selectedStreamId, onSelect, serverTime }: HistoryLogViewProps) { - if (streams.status === "loading") return ; - if (streams.status === "error") return ; - if (streams.data.length === 0) return ; - return ( - <> -
- - - - -
- {lines.status === "idle" && } - {lines.status === "loading" && } - {lines.status === "error" && } - {lines.status === "ready" && lines.data.length === 0 && } - {lines.status === "ready" && lines.data.map((line) =>
{line.text}
)} - - ); -} - -function TerminalStatusLine({ tone, label, serverTime }: { tone: "info" | "warn" | "error"; label: string; serverTime?: string }) { - return
{label}
; -} - function terminalQuickCommandsForPlugin(pluginId: string): TerminalQuickCommand[] { return terminalQuickCommandCatalog[pluginId] ?? []; } diff --git a/plugins/examples/scum-server-plugin/companion/cmd/scum-companion/main.go b/plugins/examples/scum-server-plugin/companion/cmd/scum-companion/main.go index d64402b..f18a025 100644 --- a/plugins/examples/scum-server-plugin/companion/cmd/scum-companion/main.go +++ b/plugins/examples/scum-server-plugin/companion/cmd/scum-companion/main.go @@ -30,6 +30,8 @@ func main() { trajectoryStatus := &companion.TrajectoryCollectionStatus{} collector, cleanup := buildTrajectoryCollector(config, trajectoryStatus) defer cleanup() + consoleCollector, consoleCleanup := buildConsoleLogCollector(config, client) + defer consoleCleanup() registry := companion.NewHandlerRegistry(defaultHandlerAvailability(config), companion.RuntimeAdapter{ BoundServerID: config.Component.ServerInstanceID, @@ -52,11 +54,14 @@ func main() { Health: trajectoryStatus.HealthReport, } - errorsCh := make(chan error, 2) + errorsCh := make(chan error, 3) go func() { errorsCh <- runtime.Run(ctx) }() if collector != nil { go func() { errorsCh <- collector.Run(ctx, trajectoryStatus) }() } + if consoleCollector != nil { + go func() { errorsCh <- consoleCollector.Run(ctx) }() + } err = <-errorsCh stop() @@ -75,6 +80,18 @@ func loadConfig() (companion.Config, error) { return companion.LoadConfig(file) } +func buildConsoleLogCollector(config companion.Config, client *companion.Client) (*companion.ConsoleLogCollector, func()) { + cleanup := func() {} + store, err := companion.OpenSCUMSQLStoreFromEnv(companion.PlatformMySQLDSNEnvironment) + if err != nil { + log.Printf("SCUM companion console log store unavailable: %v", err) + return nil, cleanup + } + cleanup = func() { _ = store.Close() } + secret := os.Getenv(config.Proof.MaterialEnv) + return companion.NewConsoleLogCollector(client, store, config.Component.ServerInstanceID, secret), cleanup +} + func buildTrajectoryCollector(config companion.Config, status *companion.TrajectoryCollectionStatus) (*companion.TrajectoryCollector, func()) { cleanup := func() {} if !config.Trajectory.Enabled { diff --git a/plugins/examples/scum-server-plugin/companion/console_collector.go b/plugins/examples/scum-server-plugin/companion/console_collector.go new file mode 100644 index 0000000..acba91f --- /dev/null +++ b/plugins/examples/scum-server-plugin/companion/console_collector.go @@ -0,0 +1,100 @@ +package companion + +import ( + "context" + "fmt" + "strings" + "time" +) + +type ConsoleLogStreamClient interface { + StreamLogEvents(context.Context, func(LogStreamEvent) error) error +} + +type ConsoleLogStore interface { + EnsureSchema(context.Context) error + StoreConsoleRecords(context.Context, []ConsoleRecord) (int, error) + StoreSemanticEventBatch(context.Context, SemanticEventBatch) (int, error) +} + +type ConsoleLogCollector struct { + Client ConsoleLogStreamClient + Store ConsoleLogStore + ServerInstanceID string + CorrelationSecret string + Backoff time.Duration +} + +func NewConsoleLogCollector(client ConsoleLogStreamClient, store ConsoleLogStore, serverInstanceID string, correlationSecret string) *ConsoleLogCollector { + return &ConsoleLogCollector{Client: client, Store: store, ServerInstanceID: serverInstanceID, CorrelationSecret: correlationSecret, Backoff: 2 * time.Second} +} + +func (collector *ConsoleLogCollector) Run(ctx context.Context) error { + if collector == nil || collector.Client == nil || collector.Store == nil || strings.TrimSpace(collector.ServerInstanceID) == "" { + return fmt.Errorf("console log collector is not configured") + } + if err := collector.Store.EnsureSchema(ctx); err != nil { + return err + } + backoff := collector.Backoff + if backoff <= 0 { + backoff = 2 * time.Second + } + for { + err := collector.Client.StreamLogEvents(ctx, collector.handleEvent(ctx)) + if ctx.Err() != nil { + return ctx.Err() + } + if err != nil { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + continue + } + } +} + +func (collector *ConsoleLogCollector) handleEvent(ctx context.Context) func(LogStreamEvent) error { + return func(event LogStreamEvent) error { + record, ok := consoleRecordFromLogEvent(collector.ServerInstanceID, event) + if !ok { + return nil + } + if _, err := collector.Store.StoreConsoleRecords(ctx, []ConsoleRecord{record}); err != nil { + return err + } + batch := ParseConsoleRecords(collector.ServerInstanceID, []ConsoleRecord{record}, collector.CorrelationSecret) + if _, err := collector.Store.StoreSemanticEventBatch(ctx, batch); err != nil { + return err + } + return nil + } +} + +func consoleRecordFromLogEvent(serverInstanceID string, event LogStreamEvent) (ConsoleRecord, bool) { + if event.ServerInstanceID != serverInstanceID || event.Entry.Seq == 0 || strings.TrimSpace(event.Entry.Line) == "" { + return ConsoleRecord{}, false + } + stream := consoleStreamName(event.StreamKey) + if stream == "" { + return ConsoleRecord{}, false + } + occurredAt := event.Entry.Timestamp + if occurredAt.IsZero() { + occurredAt = time.Now().UTC() + } + return ConsoleRecord{ServerID: serverInstanceID, Stream: stream, Sequence: event.Entry.Seq, OccurredAt: occurredAt.UTC(), Text: event.Entry.Line}, true +} + +func consoleStreamName(streamKey string) string { + key := strings.ToLower(strings.TrimSpace(streamKey)) + if strings.Contains(key, "stderr") || strings.HasSuffix(key, ".err") || strings.HasSuffix(key, "-err") { + return "stderr" + } + if strings.Contains(key, "stdout") || strings.Contains(key, "console") || strings.HasSuffix(key, ".out") || strings.HasSuffix(key, "-out") { + return "stdout" + } + return "" +} diff --git a/plugins/examples/scum-server-plugin/companion/console_collector_test.go b/plugins/examples/scum-server-plugin/companion/console_collector_test.go new file mode 100644 index 0000000..5339e43 --- /dev/null +++ b/plugins/examples/scum-server-plugin/companion/console_collector_test.go @@ -0,0 +1,62 @@ +package companion + +import ( + "context" + "testing" + "time" +) + +type recordingConsoleLogStore struct { + ensureCalls int + records []ConsoleRecord + batches []SemanticEventBatch +} + +func (store *recordingConsoleLogStore) EnsureSchema(context.Context) error { + store.ensureCalls++ + return nil +} + +func (store *recordingConsoleLogStore) StoreConsoleRecords(_ context.Context, records []ConsoleRecord) (int, error) { + store.records = append(store.records, records...) + return len(records), nil +} + +func (store *recordingConsoleLogStore) StoreSemanticEventBatch(_ context.Context, batch SemanticEventBatch) (int, error) { + store.batches = append(store.batches, batch) + return len(batch.Events), nil +} + +func TestConsoleLogCollectorStoresLiveConsoleEventAndSemanticBatch(t *testing.T) { + stamp := time.Date(2026, 8, 31, 4, 0, 0, 0, time.UTC) + store := &recordingConsoleLogStore{} + collector := NewConsoleLogCollector(nil, store, "server-1", "correlation-secret") + handle := collector.handleEvent(context.Background()) + + if err := handle(LogStreamEvent{ServerInstanceID: "server-1", StreamID: "stream-1", Source: "process", StreamKey: "stdout", Entry: LogEntry{Seq: 11, Timestamp: stamp, Line: "SCUM LOGIN 76561198000000001 10.0.0.1"}}); err != nil { + t.Fatalf("handle log event: %v", err) + } + if len(store.records) != 1 || store.records[0].ServerID != "server-1" || store.records[0].Stream != "stdout" || store.records[0].Sequence != 11 { + t.Fatalf("collector did not store raw console record: %#v", store.records) + } + if len(store.batches) != 1 || len(store.batches[0].Events) != 1 { + t.Fatalf("collector did not store semantic event batch: %#v", store.batches) + } + event := store.batches[0].Events[0] + if event.Type != "scum.login" || event.PlayerID != "76561198000000001" || event.NetworkCorrelation == "" || event.NetworkCorrelation == "10.0.0.1" { + t.Fatalf("unexpected semantic event: %#v", event) + } +} + +func TestConsoleLogCollectorIgnoresNonConsoleLiveLogEvents(t *testing.T) { + store := &recordingConsoleLogStore{} + collector := NewConsoleLogCollector(nil, store, "server-1", "correlation-secret") + handle := collector.handleEvent(context.Background()) + + if err := handle(LogStreamEvent{ServerInstanceID: "server-1", StreamID: "stream-1", Source: "process", StreamKey: "scum.file", Entry: LogEntry{Seq: 12, Timestamp: time.Now().UTC(), Line: "not console"}}); err != nil { + t.Fatalf("handle non-console event: %v", err) + } + if len(store.records) != 0 || len(store.batches) != 0 { + t.Fatalf("non-console event was stored: records=%#v batches=%#v", store.records, store.batches) + } +} diff --git a/plugins/examples/scum-server-plugin/companion/console_storage.go b/plugins/examples/scum-server-plugin/companion/console_storage.go new file mode 100644 index 0000000..5fdb855 --- /dev/null +++ b/plugins/examples/scum-server-plugin/companion/console_storage.go @@ -0,0 +1,127 @@ +package companion + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" +) + +func (store *SCUMSQLStore) StoreConsoleRecords(ctx context.Context, records []ConsoleRecord) (int, error) { + if store == nil || store.db == nil { + return 0, fmt.Errorf("SCUM plugin SQL store is not configured") + } + normalized := make([]ConsoleRecord, 0, len(records)) + for _, record := range records { + value, err := normalizeConsoleRecord(record) + if err != nil { + return 0, err + } + normalized = append(normalized, value) + } + if len(normalized) == 0 { + return 0, nil + } + tx, err := store.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("begin SCUM console log write: %w", err) + } + defer tx.Rollback() + stamp := time.Now().UTC() + for _, record := range normalized { + if _, err := tx.ExecContext(ctx, scumConsoleLogInsertSQL, scumConsoleRecordKey(record), record.ServerID, record.Stream, record.Sequence, record.OccurredAt, record.Text, stamp, stamp); err != nil { + return 0, fmt.Errorf("write SCUM console log: %w", err) + } + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("commit SCUM console logs: %w", err) + } + return len(normalized), nil +} + +func (store *SCUMSQLStore) StoreSemanticEventBatch(ctx context.Context, batch SemanticEventBatch) (int, error) { + if store == nil || store.db == nil { + return 0, fmt.Errorf("SCUM plugin SQL store is not configured") + } + normalized := make([]SemanticEvent, 0, len(batch.Events)) + for _, event := range batch.Events { + value, err := normalizeSemanticEvent(event) + if err != nil { + return 0, err + } + normalized = append(normalized, value) + } + if len(normalized) == 0 { + return 0, nil + } + tx, err := store.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("begin SCUM semantic event write: %w", err) + } + defer tx.Rollback() + stamp := time.Now().UTC() + for _, event := range normalized { + if _, err := tx.ExecContext(ctx, scumSemanticEventInsertSQL, scumSemanticEventRecordKey(event), event.ServerID, event.Sequence, event.Type, event.PlayerID, nullText(event.DisplayName), event.OccurredAt, nullText(event.NetworkCorrelation), stamp, stamp); err != nil { + return 0, fmt.Errorf("write SCUM semantic event: %w", err) + } + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("commit SCUM semantic events: %w", err) + } + return len(normalized), nil +} + +func normalizeConsoleRecord(record ConsoleRecord) (ConsoleRecord, error) { + record.ServerID = strings.TrimSpace(record.ServerID) + record.Stream = strings.TrimSpace(record.Stream) + record.Text = strings.TrimRight(record.Text, "\r\n") + if record.ServerID == "" || (record.Stream != "stdout" && record.Stream != "stderr") || record.Sequence == 0 || record.OccurredAt.IsZero() || strings.TrimSpace(record.Text) == "" || len(record.Text) > 8192 { + return ConsoleRecord{}, fmt.Errorf("SCUM console record is invalid") + } + record.OccurredAt = record.OccurredAt.UTC() + return record, nil +} + +func normalizeSemanticEvent(event SemanticEvent) (SemanticEvent, error) { + event.ServerID = strings.TrimSpace(event.ServerID) + event.Type = strings.TrimSpace(event.Type) + event.PlayerID = strings.TrimSpace(event.PlayerID) + event.DisplayName = strings.TrimSpace(event.DisplayName) + event.NetworkCorrelation = strings.TrimSpace(event.NetworkCorrelation) + if event.ServerID == "" || event.Sequence == 0 || event.Type == "" || event.PlayerID == "" || event.OccurredAt.IsZero() || len(event.Type) > 80 || len(event.PlayerID) > 80 || len(event.DisplayName) > 120 || len(event.NetworkCorrelation) > 128 { + return SemanticEvent{}, fmt.Errorf("SCUM semantic event is invalid") + } + event.OccurredAt = event.OccurredAt.UTC() + return event, nil +} + +func scumConsoleRecordKey(record ConsoleRecord) string { + digest := sha256.Sum256([]byte(strings.Join([]string{record.ServerID, record.Stream, fmt.Sprintf("%d", record.Sequence)}, "\x00"))) + return hex.EncodeToString(digest[:]) +} + +func scumSemanticEventRecordKey(event SemanticEvent) string { + digest := sha256.Sum256([]byte(strings.Join([]string{event.ServerID, event.Type, event.PlayerID, fmt.Sprintf("%d", event.Sequence)}, "\x00"))) + return hex.EncodeToString(digest[:]) +} + +const scumConsoleLogInsertSQL = ` +INSERT INTO scum_console_logs ( + record_key, server_instance_id, stream, sequence, occurred_at, line_text, created_at, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON DUPLICATE KEY UPDATE + occurred_at = VALUES(occurred_at), + line_text = VALUES(line_text), + updated_at = VALUES(updated_at)` + +const scumSemanticEventInsertSQL = ` +INSERT INTO scum_semantic_events ( + record_key, server_instance_id, sequence, event_type, player_id, display_name, occurred_at, network_correlation, created_at, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON DUPLICATE KEY UPDATE + display_name = VALUES(display_name), + occurred_at = VALUES(occurred_at), + network_correlation = VALUES(network_correlation), + updated_at = VALUES(updated_at)` diff --git a/plugins/examples/scum-server-plugin/companion/log_stream.go b/plugins/examples/scum-server-plugin/companion/log_stream.go new file mode 100644 index 0000000..4cb2cb1 --- /dev/null +++ b/plugins/examples/scum-server-plugin/companion/log_stream.go @@ -0,0 +1,125 @@ +package companion + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const logEventsPath = "/api/v1/game-client-bridge/companion/logs/events" + +type LogEntry struct { + Seq uint64 `json:"seq"` + Timestamp time.Time `json:"timestamp"` + Level string `json:"level,omitempty"` + Line string `json:"line"` + Fields map[string]string `json:"fields,omitempty"` + Redacted bool `json:"redacted"` +} + +type LogStreamEvent struct { + ServerInstanceID string `json:"serverInstanceId"` + StreamID string `json:"streamId"` + Source string `json:"source"` + StreamKey string `json:"streamKey"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` + LatestSeq uint64 `json:"latestSeq"` + Entry LogEntry `json:"entry"` +} + +type logStreamRequest struct { + SessionToken string `json:"sessionToken"` +} + +func (client *Client) StreamLogEvents(ctx context.Context, handle func(LogStreamEvent) error) error { + if handle == nil { + return fmt.Errorf("log event handler is required") + } + token, err := client.currentSession() + if err != nil { + return err + } + encoded, err := json.Marshal(logStreamRequest{SessionToken: token}) + if err != nil { + return fmt.Errorf("encode log stream request: %w", err) + } + if len(encoded) > maxRequestBytes { + return fmt.Errorf("platform request exceeds the bounded payload size") + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, client.config.Platform.BaseURL+logEventsPath, bytes.NewReader(encoded)) + if err != nil { + return fmt.Errorf("create log stream request: %w", err) + } + request.Header.Set("Accept", "text/event-stream") + request.Header.Set("Content-Type", "application/json") + response, err := client.httpClient.Do(request) + if err != nil { + return fmt.Errorf("open log stream: %w", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096)) + return HTTPError{StatusCode: response.StatusCode, ExpectedStatus: http.StatusOK} + } + return readLogEventStream(ctx, response.Body, handle) +} + +func readLogEventStream(ctx context.Context, body io.Reader, handle func(LogStreamEvent) error) error { + reader := bufio.NewReader(body) + var eventName string + var dataLines []string + flush := func() error { + if len(dataLines) == 0 { + eventName = "" + return nil + } + name := eventName + if name == "" { + name = "message" + } + payload := strings.Join(dataLines, "\n") + eventName = "" + dataLines = nil + if name != "log" { + return nil + } + var event LogStreamEvent + if err := json.Unmarshal([]byte(payload), &event); err != nil { + return fmt.Errorf("decode log event: %w", err) + } + return handle(event) + } + for { + line, err := reader.ReadString('\n') + if len(line) > 0 { + line = strings.TrimRight(line, "\r\n") + switch { + case line == "": + if flushErr := flush(); flushErr != nil { + return flushErr + } + case strings.HasPrefix(line, ":"): + case strings.HasPrefix(line, "event:"): + eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + case strings.HasPrefix(line, "data:"): + dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + } + } + if err != nil { + if ctx != nil && ctx.Err() != nil { + return ctx.Err() + } + if err == io.EOF { + return flush() + } + return fmt.Errorf("read log stream: %w", err) + } + } +} diff --git a/plugins/examples/scum-server-plugin/companion/log_stream_test.go b/plugins/examples/scum-server-plugin/companion/log_stream_test.go new file mode 100644 index 0000000..b08c9cd --- /dev/null +++ b/plugins/examples/scum-server-plugin/companion/log_stream_test.go @@ -0,0 +1,66 @@ +package companion + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" + "time" +) + +func TestClientStreamLogEventsUsesComponentSessionBodyAndSSE(t *testing.T) { + stamp := time.Date(2026, 8, 31, 2, 0, 0, 0, time.UTC) + config := loadTestConfig(t) + client := newTestClient(t, config, roundTripFunc(func(request *http.Request) (*http.Response, error) { + if request.Method != http.MethodPost || request.URL.Scheme != "https" || request.URL.Host != "platform.example.test" || request.URL.Path != logEventsPath { + t.Fatalf("unexpected log stream request: %s %s", request.Method, request.URL.String()) + } + if request.Header.Get("Authorization") != "" { + t.Fatalf("component session must stay in the typed JSON body, got Authorization header") + } + if request.Header.Get("Accept") != "text/event-stream" || request.Header.Get("Content-Type") != "application/json" { + t.Fatalf("unexpected log stream headers: %+v", request.Header) + } + var body logStreamRequest + decodeRequest(t, request, &body) + if body.SessionToken != "session-token" { + t.Fatalf("unexpected session token body: %#v", body) + } + logPayload, _ := json.Marshal(LogStreamEvent{ServerInstanceID: "server-1", StreamID: "stream-1", Source: "process", StreamKey: "stdout", LogSessionID: "session-live", SessionStartedAt: stamp, LatestSeq: 7, Entry: LogEntry{Seq: 7, Timestamp: stamp.Add(time.Second), Line: "SCUM LOGIN 76561198000000001 10.0.0.1", Redacted: true}}) + bodyText := strings.Join([]string{ + "event: ready", + "data: {\"serverInstanceId\":\"server-1\"}", + "", + "event: log", + "data: " + string(logPayload), + "", + ": heartbeat", + "", + }, "\n") + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(bodyText)), Request: request}, nil + }), stamp) + client.mu.Lock() + client.sessionToken = "session-token" + client.sessionExpiresAt = stamp.Add(time.Hour) + client.mu.Unlock() + + var events []LogStreamEvent + if err := client.StreamLogEvents(context.Background(), func(event LogStreamEvent) error { + events = append(events, event) + return nil + }); err != nil { + t.Fatalf("stream log events: %v", err) + } + if len(events) != 1 || events[0].Entry.Seq != 7 || events[0].Entry.Line != "SCUM LOGIN 76561198000000001 10.0.0.1" { + t.Fatalf("unexpected streamed log events: %#v", events) + } +} + +func TestReadLogEventStreamRejectsMalformedLogEvent(t *testing.T) { + err := readLogEventStream(context.Background(), strings.NewReader("event: log\ndata: {not-json}\n\n"), func(LogStreamEvent) error { return nil }) + if err == nil || !strings.Contains(err.Error(), "decode log event") { + t.Fatalf("expected malformed log event rejection, got %v", err) + } +} diff --git a/plugins/examples/scum-server-plugin/companion/storage.go b/plugins/examples/scum-server-plugin/companion/storage.go index 2cd0bc9..f1a7b18 100644 --- a/plugins/examples/scum-server-plugin/companion/storage.go +++ b/plugins/examples/scum-server-plugin/companion/storage.go @@ -348,6 +348,33 @@ CREATE TABLE IF NOT EXISTS scum_trajectories ( UNIQUE KEY scum_trajectories_sample_uq (server_instance_id, subject_type, subject_id, sampled_at), KEY scum_trajectories_subject_idx (server_instance_id, subject_type, subject_id, observed_at), KEY scum_trajectories_sampled_idx (server_instance_id, sampled_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, ` +CREATE TABLE IF NOT EXISTS scum_console_logs ( + record_key CHAR(64) PRIMARY KEY, + server_instance_id VARCHAR(96) NOT NULL, + stream VARCHAR(16) NOT NULL, + sequence BIGINT UNSIGNED NOT NULL, + occurred_at DATETIME(6) NOT NULL, + line_text TEXT NOT NULL, + created_at DATETIME(6) NOT NULL, + updated_at DATETIME(6) NOT NULL, + UNIQUE KEY scum_console_logs_stream_uq (server_instance_id, stream, sequence), + KEY scum_console_logs_time_idx (server_instance_id, occurred_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, ` +CREATE TABLE IF NOT EXISTS scum_semantic_events ( + record_key CHAR(64) PRIMARY KEY, + server_instance_id VARCHAR(96) NOT NULL, + sequence BIGINT UNSIGNED NOT NULL, + event_type VARCHAR(80) NOT NULL, + player_id VARCHAR(80) NOT NULL, + display_name VARCHAR(120) NULL, + occurred_at DATETIME(6) NOT NULL, + network_correlation VARCHAR(128) NULL, + created_at DATETIME(6) NOT NULL, + updated_at DATETIME(6) NOT NULL, + UNIQUE KEY scum_semantic_events_uq (server_instance_id, event_type, player_id, sequence), + KEY scum_semantic_events_player_idx (server_instance_id, player_id, occurred_at), + KEY scum_semantic_events_type_idx (server_instance_id, event_type, occurred_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`} const scumTrajectoryInsertSQL = ` diff --git a/plugins/examples/scum-server-plugin/companion/storage_test.go b/plugins/examples/scum-server-plugin/companion/storage_test.go index 0b01033..000493c 100644 --- a/plugins/examples/scum-server-plugin/companion/storage_test.go +++ b/plugins/examples/scum-server-plugin/companion/storage_test.go @@ -102,6 +102,61 @@ func TestSCUMSQLStoreWritesTrajectorySamplesWithoutCoordinateConversion(t *testi } } +func TestSCUMSQLStoreWritesConsoleAndSemanticEventsToPluginTables(t *testing.T) { + db, recorder := newRecordingSQLDB(t, "console-db") + store, err := NewSCUMSQLStore(db) + if err != nil { + t.Fatalf("create SCUM SQL store: %v", err) + } + stamp := time.Date(2026, 8, 31, 3, 0, 0, 0, time.UTC) + records := []ConsoleRecord{{ServerID: "server-1", Stream: "stdout", Sequence: 9, OccurredAt: stamp, Text: "SCUM LOGIN 76561198000000001 10.0.0.1"}} + written, err := store.StoreConsoleRecords(context.Background(), records) + if err != nil || written != 1 { + t.Fatalf("store console records: written=%d err=%v", written, err) + } + batch := ParseConsoleRecords("server-1", records, "correlation-secret") + if len(batch.Events) != 1 || batch.Events[0].NetworkCorrelation == "" { + t.Fatalf("expected one correlated semantic event: %#v", batch) + } + semanticWritten, err := store.StoreSemanticEventBatch(context.Background(), batch) + if err != nil || semanticWritten != 1 { + t.Fatalf("store semantic events: written=%d err=%v", semanticWritten, err) + } + if recorder.commits != 2 { + t.Fatalf("console and semantic writes did not commit once each: %d", recorder.commits) + } + + consoleIndex := findStatement(recorder.statements, "INSERT INTO scum_console_logs") + if consoleIndex < 0 { + t.Fatalf("missing console insert statement: %v", recorder.statements) + } + consoleInsert := recorder.statements[consoleIndex] + if !strings.Contains(consoleInsert, "line_text") || strings.Contains(consoleInsert, "platform_logs") { + t.Fatalf("console SQL must write the SCUM plugin table only: %s", consoleInsert) + } + consoleArgs := recorder.args[consoleIndex] + if consoleArgs[1].Value != "server-1" || consoleArgs[2].Value != "stdout" || !driverNumberEquals(consoleArgs[3].Value, 9) || consoleArgs[5].Value != records[0].Text { + t.Fatalf("unexpected console insert args: %+v", consoleArgs) + } + + semanticIndex := findStatement(recorder.statements, "INSERT INTO scum_semantic_events") + if semanticIndex < 0 { + t.Fatalf("missing semantic event insert statement: %v", recorder.statements) + } + semanticInsert := recorder.statements[semanticIndex] + if strings.Contains(semanticInsert, "platform_logs") || strings.Contains(semanticInsert, "run_logs") { + t.Fatalf("semantic SQL must write the SCUM plugin table only: %s", semanticInsert) + } + semanticArgs := recorder.args[semanticIndex] + if semanticArgs[1].Value != "server-1" || !driverNumberEquals(semanticArgs[2].Value, 9) || semanticArgs[3].Value != "scum.login" || semanticArgs[4].Value != "76561198000000001" { + t.Fatalf("unexpected semantic insert args: %+v", semanticArgs) + } + correlation, ok := semanticArgs[7].Value.(string) + if !ok || correlation == "10.0.0.1" || len(correlation) != 64 { + t.Fatalf("semantic event stored raw or missing network correlation: %+v", semanticArgs[7]) + } +} + func TestTrajectorySamplesFromSCUMRowsKeepWorldCoordinates(t *testing.T) { sampledAt := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC) positionSamples, err := TrajectorySamplesFromPositionRows("server-1", []map[string]any{ @@ -168,3 +223,16 @@ func findStatement(statements []string, prefix string) int { } return -1 } + +func driverNumberEquals(value any, want int64) bool { + switch typed := value.(type) { + case int: + return int64(typed) == want + case int64: + return typed == want + case uint64: + return typed == uint64(want) + default: + return false + } +}