From 302f1f64b7b1804f9d788d48a7095b875cfa4000 Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Mon, 17 Aug 2026 08:48:45 +0800 Subject: [PATCH] Implement platform management features --- .../.openspec.yaml | 2 + .../design.md | 75 +++ .../proposal.md | 31 ++ .../current-supervised-log-session/spec.md | 76 +++ .../tasks.md | 28 + .../api/channel_isolation_handlers_test.go | 3 + platform/api/log_events_handlers.go | 250 +++++++-- platform/api/log_ingest_handlers_test.go | 262 ++++++++- platform/domain/log_ingest.go | 2 + platform/domain/resources.go | 2 + platform/dto/log_ingest.go | 16 + platform/dto/resources.go | 8 + platform/model/resources.go | 4 + .../service/distribution_build_execution.go | 14 +- .../distribution_build_execution_test.go | 71 +++ platform/service/distribution_builder.go | 4 +- platform/service/distribution_builder_test.go | 8 + platform/service/log_events.go | 54 +- platform/service/log_ingest.go | 36 +- platform/service/log_ingest_privacy_test.go | 31 ++ platform/service/log_ingest_test.go | 84 ++- platform/service/resources.go | 49 +- platform/service/server_lifecycle.go | 9 + .../service/server_lifecycle_projection.go | 2 + platform/service/server_lifecycle_test.go | 56 ++ platform/validator/log_ingest.go | 8 + platform/validator/resources.go | 8 + platform_web/api/client.test.ts | 14 +- platform_web/api/client.ts | 5 +- platform_web/api/types.ts | 13 +- .../ServerManagementTerminalDrawer.test.tsx | 255 +++++++++ .../ServerManagementTerminalDrawer.tsx | 137 ++++- platform_web/schemas/serverManagement.test.ts | 13 +- platform_web/schemas/serverManagement.ts | 1 - platform_web/utils/logEvents.ts | 10 +- scripts/README.md | 1 + scripts/local-debug/smoke.sh | 501 ++++++++++++++++-- .../verify-current-log-session-sse.mjs | 152 ++++++ 38 files changed, 2185 insertions(+), 110 deletions(-) create mode 100644 openspec/changes/follow-current-supervised-log-session/.openspec.yaml create mode 100644 openspec/changes/follow-current-supervised-log-session/design.md create mode 100644 openspec/changes/follow-current-supervised-log-session/proposal.md create mode 100644 openspec/changes/follow-current-supervised-log-session/specs/current-supervised-log-session/spec.md create mode 100644 openspec/changes/follow-current-supervised-log-session/tasks.md create mode 100644 platform/service/log_ingest_privacy_test.go create mode 100644 platform_web/components/ServerManagementTerminalDrawer.test.tsx create mode 100644 scripts/local-debug/verify-current-log-session-sse.mjs diff --git a/openspec/changes/follow-current-supervised-log-session/.openspec.yaml b/openspec/changes/follow-current-supervised-log-session/.openspec.yaml new file mode 100644 index 0000000..4af8641 --- /dev/null +++ b/openspec/changes/follow-current-supervised-log-session/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-14 diff --git a/openspec/changes/follow-current-supervised-log-session/design.md b/openspec/changes/follow-current-supervised-log-session/design.md new file mode 100644 index 0000000..24abd21 --- /dev/null +++ b/openspec/changes/follow-current-supervised-log-session/design.md @@ -0,0 +1,75 @@ +## Context + +Run currently persists a supervised process identity and forwards plugin-declared `process.stdout` and `process.stderr` through its durable spool. Autonomous lifecycle stream IDs are stable per Run endpoint, server, and stream key, so Platform cannot separate a later process start from the previous generation. The terminal then lists and replays all server streams, including old jobs, before it listens for SSE appends. + +Platform remains the authorization and persistence boundary. Run remains the sole authority for the supervised process. The design must expose only a logical session identifier and timestamps, never output paths, PIDs, command lines, credentials, or direct process handles. + +## Goals / Non-Goals + +**Goals:** + +- Follow only the current plugin-declared supervised process output in the live terminal. +- Preserve the same session when Run restarts and resumes an already-running supervised process. +- Switch an open terminal atomically when the supervised process starts a new generation. +- Keep bounded recent replay for the selected session and keep all older streams readable as explicit history. +- Keep RCON and any future plugin command transport independent of the log data path. + +**Non-Goals:** + +- Expose browser shell access, host paths, raw sockets, or process stdin by default. +- Infer game logs from arbitrary files or special-case SCUM, Minecraft, or another game. +- Delete, migrate, or reinterpret legacy log bodies. + +## Decisions + +### Process-generation session is generated and persisted by Run + +`ProcessIdentity` receives a random/logical session ID and its start timestamp when a new managed process is started. The identity is journaled, so a Run restart resumes tailing and continues using the identical session for the still-running process. Starting a replacement process generates a new session. + +The session ID is added to process log batches and autonomous process stream IDs. This avoids sequence collisions and gives Platform a durable grouping key. A stable stream ID with an inferred time was rejected because delayed spool upload could make an old process appear newer. + +### Platform selects the current session only from supervised process streams + +Platform persists the logical session ID and process-start timestamp with each stream. Run identifies an observed generation as `log-session:`, and a persisted session is eligible as current only while the bound Run endpoint is online, Platform's latest Run-reported supervised-process fact is `running`, and that fact names the same session. This generation binding prevents a delayed batch from an older or not-yet-observed process from becoming current merely because it has a newer timestamp. Among the matching session's process streams, timestamp and deterministic ID ordering remain useful only for stable replay ordering. Only `source=process` streams with a non-empty session ID participate. Legacy, job, file-tail, and management-program logs remain historical and cannot displace a current terminal session. + +Run reports observations for every process it supervises, including servers handled by a general worker rather than only a generated single-server Run. Platform publishes process-state changes into the existing server log subscription. A stopped, exited, or failed fact emits an empty session boundary; a later running fact re-evaluates the persisted streams. Run endpoint disconnect alone does not end a session because the game process may survive a Run restart, but a newly opened terminal treats an offline endpoint as having no proven current session until Run reconnects and reports the process fact. + +Dispatching a new start clears Platform's previous managed-process binding before the job is queued. Historical output from the prior generation therefore cannot reappear during the interval between desired start intent and Run's first observation for the replacement generation. + +Using the newest arbitrary stream was rejected because a file backfill or command result is not evidence of the process that the terminal must follow. + +### SSE carries an explicit session boundary + +The server log event endpoint begins with a `session` event, then exposes only the selected session's streams and a bounded replay. On each incoming process event Platform re-evaluates the active session. When it changed, Platform emits a new `session` event, its stream metadata, and a new selected-session replay before normal appends. The client clears the live buffer on the session boundary and remains connected; it never has to guess based on timestamps. + +Re-opening a new EventSource on each restart was rejected because it races with output and leaves the old terminal contents visible while reconnecting. + +### Durable source cursors make resumed file reads idempotent + +Each process-output append carries its generation-local source-file cursor into the durable spool. The spool commits sequence allocation and the source cursor with the batch, restores both from pending segments, and treats a replayed cursor as already committed. This closes the crash window between a durable spool append and the process journal offset checkpoint. + +The process journal retains superseded generations until both stdout and stderr files are fully drained. Starting generation B therefore cannot erase generation A recovery state if Run exits before A's final output is spooled. Existing journals without a session ID are upgraded in place for a still-running process so deployment of the new Run does not require restarting the game process. + +### History is explicitly requested + +The existing log-stream list/cursor contracts remain the history source. The terminal includes a deliberate history mode that reads selected historical stream cursors; it is not fed by the live SSE endpoint. This retains operator access without contaminating the live view. + +## Risks / Trade-offs + +- [Old Run versions emit no session ID] → Platform keeps their logs accessible in history but excludes them from live-follow selection; deployment is backward-safe and becomes live when Run is updated. +- [A process emits before the initial SSE replay completes] → line identity is stream ID plus sequence and client-side merge de-duplicates replay/live overlap. +- [Run exits after spooling a line but before checkpointing its file offset] → the durable source cursor makes the repeated read idempotent. +- [A new process starts before the old output files finish draining] → Run journals the retired generation until both channels are complete. +- [Session metadata tampering] → Platform accepts it only through the existing authenticated Run ingest channel and requires it to match the stream's immutable metadata. +- [Unbounded current logs] → live replay remains bounded while the full selected stream remains available through explicit history cursors. + +## Migration Plan + +1. Deploy Platform so it accepts session metadata and treats old streams as history. +2. Deploy Run so new managed process starts generate session-scoped stream IDs and resumptions preserve the persisted session. +3. Deploy the frontend terminal that understands `session` SSE boundaries and offers explicit history. +4. Rollback is safe: historical streams and batches remain immutable; an older frontend ignores the additional SSE event and Run can continue uploading session-scoped streams to the compatible Platform. + +## Open Questions + +None. The session scope is the plugin-declared process output, and the terminal's default is the current session. diff --git a/openspec/changes/follow-current-supervised-log-session/proposal.md b/openspec/changes/follow-current-supervised-log-session/proposal.md new file mode 100644 index 0000000..f0da5d8 --- /dev/null +++ b/openspec/changes/follow-current-supervised-log-session/proposal.md @@ -0,0 +1,31 @@ +## Why + +The server terminal currently replays recent entries across every log stream ever created for an instance. That makes an open terminal appear stuck on stale startup output and leaves it unable to distinguish the process currently supervised by Run from a prior Run or process generation. + +Operators need a terminal that continuously follows the current managed process even when Run or the process restarts, while keeping historical logs available deliberately and keeping game command execution separate from log transport. + +## What Changes + +- Define a current supervised-log session identity for each Run-managed process generation and attach it to process stdout/stderr log streams. +- Make the live terminal subscribe to the active supervised-log session by default, including a bounded recent replay from that session only. +- Notify terminal subscribers when the active session changes so an already-open terminal automatically replaces the old session output with the new process generation and continues following it. +- Clear the live terminal when Run reports that the supervised process stopped or exited, and restore the same session when a restarted Run confirms that the process survived. +- Retain historical logs as an explicit history view instead of mixing them into the live terminal. +- Preserve RCON and other plugin-declared command transports solely for command dispatch; they are not log sources. +- Extend the local end-to-end smoke coverage with a process/Run generation switch and output markers proving that the terminal feed follows the new generation. + +## Capabilities + +### New Capabilities + +- `current-supervised-log-session`: identifies and follows the current plugin-declared supervised process stdout/stderr session, including generation changes and separate explicit history access. + +### Modified Capabilities + + + +## Impact + +- Affected code: Run process supervision and log metadata, Platform log ingest/domain/API/SSE handling, the server-management terminal API client and drawer, and the local debug smoke path. +- Affected API: Run log ingest, supervised-process observations, and server log SSE gain a bounded session identity/filtering contract; a history-only query is exposed or retained separately. +- Compatibility: existing persisted log streams remain readable through history, but no longer appear by default in the live terminal. diff --git a/openspec/changes/follow-current-supervised-log-session/specs/current-supervised-log-session/spec.md b/openspec/changes/follow-current-supervised-log-session/specs/current-supervised-log-session/spec.md new file mode 100644 index 0000000..b2f3827 --- /dev/null +++ b/openspec/changes/follow-current-supervised-log-session/specs/current-supervised-log-session/spec.md @@ -0,0 +1,76 @@ +## ADDED Requirements + +### Requirement: Live terminal follows the current supervised process session +The system SHALL identify each newly started Run-supervised process generation with a durable logical session ID, and SHALL associate its plugin-declared `process.stdout` and `process.stderr` streams with that session. A Run restart that resumes the same running process SHALL retain that session ID. + +#### Scenario: Process output is collected independently of the terminal +- **WHEN** a plugin-declared supervised process writes stdout or stderr while no browser terminal is open +- **THEN** Run SHALL durably spool and upload that output under the process's logical session without opening an RCON connection or depending on browser state + +#### Scenario: Run resumes a running process after restart +- **WHEN** Run restarts while its persisted supervised process is still running +- **THEN** Run SHALL resume collecting output with the persisted process session ID and the terminal SHALL continue following that session + +#### Scenario: Run crashes between durable append and output checkpoint +- **WHEN** a process line is already durable in the local spool but Run restarts before its output-file offset checkpoint is persisted +- **THEN** Run SHALL recognize the repeated source cursor and SHALL NOT allocate or upload a duplicate log entry + +#### Scenario: A replacement starts before the prior output drain completes +- **WHEN** Run starts a replacement process while the prior generation still has unread stdout or stderr +- **THEN** Run SHALL retain and resume the prior generation's drain state independently of the replacement generation + +#### Scenario: Existing process journal predates session metadata +- **WHEN** an upgraded Run loads a legacy journal for a supervised process that is still alive +- **THEN** Run SHALL assign and persist a session for that same process without requiring the game process to restart + +### Requirement: Live SSE exposes only the active supervised session +The server log SSE endpoint SHALL select the supervised-process session named by the current `running` process fact from an online bound Run endpoint, publish an explicit session boundary, and replay only a bounded set of entries from that selected session. The process fact SHALL identify the generation as `log-session:`. The endpoint SHALL exclude legacy, completed-job, file-tail, management-program, unbound, and older supervised-session entries from the default live feed. + +#### Scenario: Terminal opens after prior process generations +- **WHEN** an instance has historical job streams and older process sessions and an operator opens the terminal +- **THEN** the endpoint SHALL publish only the active session metadata and its bounded recent output before live appends + +#### Scenario: No current session exists +- **WHEN** no session-scoped supervised process stream has been accepted for an instance +- **THEN** the endpoint SHALL publish a ready empty live session and SHALL NOT replay unrelated historical streams + +#### Scenario: Persisted session belongs to a stopped process +- **WHEN** the newest persisted session belongs to a process that Run reported stopped, exited, or failed +- **THEN** a newly opened terminal SHALL receive an empty live session and SHALL expose the persisted output only through history + +#### Scenario: Output arrives before its generation observation +- **WHEN** session-scoped process output is accepted before Run reports a matching `log-session:` running observation +- **THEN** the endpoint SHALL retain the output as history and SHALL NOT select it as the current live session until that matching observation arrives + +#### Scenario: A new start is dispatched from a stopped instance +- **WHEN** Platform accepts a new start request for an instance with a previous process-session binding +- **THEN** Platform SHALL clear the previous binding before dispatch so historical output cannot become current while the replacement generation is unobserved + +### Requirement: Open terminal changes session without reconnection +The SSE endpoint SHALL detect a newer supervised-process session on accepted process output and SHALL emit a new session boundary, its stream metadata, and the new session replay to existing subscribers before sending subsequent output for that session. + +#### Scenario: Process restarts while terminal is open +- **WHEN** a replacement supervised process writes its first stdout or stderr entry while an operator's terminal is open +- **THEN** the terminal SHALL discard the old live buffer, label the new current session, and display the new generation's output without the operator reopening the terminal + +#### Scenario: Current process exits while terminal is open +- **WHEN** Run reports that the current supervised process stopped, exited, or failed +- **THEN** the same SSE connection SHALL emit an empty session boundary and the terminal SHALL remove the ended session from its live buffer + +#### Scenario: Run reconnects to a surviving process +- **WHEN** a Run endpoint reconnects and reports that its persisted supervised process is still running +- **THEN** the same SSE connection SHALL reselect that process's existing session without creating a replacement session + +### Requirement: Historical logs remain explicit and separate +The system SHALL retain persisted log streams and cursor queries for historical inspection. The terminal SHALL expose history only through an explicit operator action or view and SHALL NOT merge it into the current live buffer. + +#### Scenario: Operator reviews an older stream +- **WHEN** an operator explicitly selects a historical log stream +- **THEN** the terminal SHALL fetch and display that stream's retained cursor entries separately from the live session + +### Requirement: Command transports are not log sources +The system SHALL use RCON or another plugin-declared command transport only to execute a requested command. It SHALL collect live terminal output only through plugin-declared log sources, prioritizing the supervised process stdout/stderr channels. + +#### Scenario: Operator sends an RCON command +- **WHEN** an operator submits a SCUM management command +- **THEN** the command SHALL use the protected RCON command path while live terminal output continues to arrive independently from the supervised process stream diff --git a/openspec/changes/follow-current-supervised-log-session/tasks.md b/openspec/changes/follow-current-supervised-log-session/tasks.md new file mode 100644 index 0000000..a11ece1 --- /dev/null +++ b/openspec/changes/follow-current-supervised-log-session/tasks.md @@ -0,0 +1,28 @@ +## 1. Run process-session protocol + +- [x] 1.1 Persist a new logical session ID and start timestamp for each newly supervised process while retaining it during Run output resumption. +- [x] 1.2 Include session metadata in durable process log batches and make autonomous process stream IDs generation-scoped without changing non-process/job log behavior. +- [x] 1.3 Add Run tests for new process generations and Run restart/resumption retaining the session. +- [x] 1.4 Make source-file replay idempotent across crashes, retain undrained retired generations, atomically replace aggregated spool segments, and upgrade live legacy journal entries. + +## 2. Platform active-session feed + +- [x] 2.1 Persist and validate immutable session metadata on log streams, including session-scoped Run stream recognition. +- [x] 2.2 Select active supervised-process streams and add session-boundary SSE events with selected-session-only replay and live filtering. +- [x] 2.3 Add Platform service/API tests covering stale history exclusion, new-session switches, and command/log separation. +- [x] 2.4 Gate initial current-session replay on online Run process facts and publish empty/recovered session boundaries on supervised process observations. + +## 3. Terminal live and history views + +- [x] 3.1 Extend frontend log contracts and SSE parsing for session boundaries. +- [x] 3.2 Make the terminal clear and follow a switched session automatically, and provide an explicit historical-stream view. +- [x] 3.3 Add frontend tests for current-session replay, live session switch, and separated history. + +## 4. End-to-end verification + +- [x] 4.1 Extend local debug smoke to prove first-generation output, process/Run generation switch, and post-switch terminal output. +- [ ] 4.2 Run OpenSpec validation, focused unit tests, structural validation, and the complete local smoke; record any external-environment blocker precisely. + - 2026-08-15 verification passed: `openspec validate follow-current-supervised-log-session --strict`; `go test ./api ./service ./domain ./dto ./validator` in `platform/`; `go test ./runtime ./spool ./protocol` in `run/`; `npm --prefix platform_web test -- ServerManagementTerminalDrawer client schemas/serverManagement`; `scripts/check-structure.sh`; `bash -n scripts/local-debug/smoke.sh`. + - 2026-08-15 complete local smoke command attempted: `LOCAL_DEBUG_ROOT=/private/tmp/browser-local-debug-current-session-smoke LOCAL_DEBUG_PLATFORM_PORT=18198 LOCAL_DEBUG_WEB_PORT=5192 LOCAL_DEBUG_SELF_START=true scripts/dev-smoke.sh`. + - Current supervised log-session proof passed inside that smoke, including generation A replay, Run restart/resumption, generation B switch on the same SSE connection, and explicit-history exclusion; evidence file: `/private/tmp/browser-local-debug-current-session-smoke/smoke/current-log-session-verification.json`. + - External blocker: the complete smoke later failed in the platform Docker distribution builder at the host-native Run generation step because `go mod download` timed out fetching `github.com/dustin/go-humanize@v1.0.1` from `proxy.golang.org` (`dial tcp 142.251.33.209:443: i/o timeout`). diff --git a/platform/api/channel_isolation_handlers_test.go b/platform/api/channel_isolation_handlers_test.go index b1618d8..8f7cd4f 100644 --- a/platform/api/channel_isolation_handlers_test.go +++ b/platform/api/channel_isolation_handlers_test.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" "testing" + "time" "browser.local/platform/domain" "browser.local/platform/dto" @@ -67,6 +68,8 @@ func TestRunChannelAPIInterleavedRequestsMutateIndependentState(t *testing.T) { logBatch := validLogBatchRequest(t, hello.SessionToken, 1, 1) logBatch.LogStreamID = "log-channel-isolation" + logBatch.LogSessionID = "" + logBatch.SessionStartedAt = time.Time{} logRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", logBatch) assertStatus(t, logRecorder, http.StatusOK) logAck := decodeBody[dto.LogBatchIngestResponse](t, logRecorder) diff --git a/platform/api/log_events_handlers.go b/platform/api/log_events_handlers.go index 49ed30a..9bb432d 100644 --- a/platform/api/log_events_handlers.go +++ b/platform/api/log_events_handlers.go @@ -18,6 +18,7 @@ const ( defaultLogEventHistoryLimit = 100 maxLogEventHistoryLimit = 10000 logEventHeartbeatInterval = 15 * time.Second + managedLogSessionIDPrefix = "log-session:" ) // serverLogEvents streams platform-accepted server log history and live append events for the terminal drawer. @@ -26,7 +27,7 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { writeMethodNotAllowed(w, http.MethodGet) return } - instance, streams, subscription, err := h.openLogEventSubscription(r) + instance, streams, liveEligible, subscription, err := h.openLogEventSubscription(r) if err != nil { writeServiceError(w, err) return @@ -46,25 +47,17 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) historyLimit := parseLogEventHistoryLimit(r.URL.Query().Get("historyLimit")) - for _, stream := range streams { - if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil { - return - } + active := supervisedLogSession{} + if liveEligible { + active = activeSupervisedLogSession(streams) } - if historyLimit > 0 { - history, err := h.loadLogEventHistory(streams, historyLimit) - if err != nil { - _ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()}) - flusher.Flush() - return - } - for _, event := range history { - if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { - return - } - } + emittedThrough, err := h.writeCurrentLogSession(w, instance.ID, active, historyLimit) + if err != nil { + _ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()}) + flusher.Flush() + return } - if err := writeSSEJSON(w, "ready", "", dto.LogStreamEventsReadyResponse{ServerInstanceID: instance.ID, StreamCount: len(streams), ServerTime: time.Now().UTC()}); err != nil { + if err := writeSSEJSON(w, "ready", "", dto.LogStreamEventsReadyResponse{ServerInstanceID: instance.ID, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil { return } flusher.Flush() @@ -75,13 +68,85 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { select { case <-r.Context().Done(): return - case event, ok := <-subscription.Events: + case subscriptionEvent, ok := <-subscription.Events: if !ok { return } + if subscriptionEvent.Kind == service.LogEventSubscriptionEventProcessState { + if subscriptionEvent.ServerInstanceID != instance.ID { + continue + } + if subscriptionEvent.ProcessState != domain.ServerInstanceStateRunning { + liveEligible = false + if active.sessionID == "" { + continue + } + active = supervisedLogSession{} + emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active, historyLimit) + if err != nil { + return + } + flusher.Flush() + continue + } + streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID) + if err != nil { + return + } + next := supervisedLogSession{} + if liveEligible { + next = activeSupervisedLogSession(streams) + } + if sameSupervisedLogSession(active, next) { + continue + } + active = next + emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active, historyLimit) + if err != nil { + return + } + flusher.Flush() + continue + } + if subscriptionEvent.Kind != service.LogEventSubscriptionEventLog || !liveEligible { + continue + } + event := subscriptionEvent.LogEvent + candidate := activeSupervisedLogSession([]domain.LogStream{event.Stream}) + if candidate.sessionID != "" && newerLogSession(candidate, active) { + streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID) + if err != nil { + return + } + next := supervisedLogSession{} + if liveEligible { + next = activeSupervisedLogSession(streams) + } + if !sameSupervisedLogSession(active, next) { + active = next + emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active, historyLimit) + if err != nil { + return + } + flusher.Flush() + } + } + if !active.contains(event.Stream) { + continue + } + if !active.hasStream(event.Stream.ID) { + active.streams = append(active.streams, event.Stream) + if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(event.Stream)); err != nil { + return + } + } + if event.Entry.Seq <= emittedThrough[event.Stream.ID] { + continue + } if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { return } + emittedThrough[event.Stream.ID] = event.Entry.Seq flusher.Flush() case <-heartbeat.C: if _, err := fmt.Fprintf(w, ": heartbeat %s\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil { @@ -92,34 +157,157 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) { } } -func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, service.LogEventSubscription, error) { +type supervisedLogSession struct { + sessionID string + startedAt time.Time + streams []domain.LogStream +} + +func activeSupervisedLogSession(streams []domain.LogStream) supervisedLogSession { + active := supervisedLogSession{} + for _, stream := range streams { + if stream.Source != domain.LogStreamSourceProcess || strings.TrimSpace(stream.LogSessionID) == "" || stream.SessionStartedAt.IsZero() { + continue + } + candidate := supervisedLogSession{sessionID: stream.LogSessionID, startedAt: stream.SessionStartedAt} + if active.sessionID == "" || newerLogSession(candidate, active) { + active = candidate + } + } + if active.sessionID == "" { + return active + } + for _, stream := range streams { + if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == active.sessionID && stream.SessionStartedAt.Equal(active.startedAt) { + active.streams = append(active.streams, stream) + } + } + return active +} + +func newerLogSession(candidate supervisedLogSession, current supervisedLogSession) bool { + if candidate.sessionID == "" || candidate.sessionID == current.sessionID { + return false + } + if current.sessionID == "" { + return true + } + if !candidate.startedAt.Equal(current.startedAt) { + return candidate.startedAt.After(current.startedAt) + } + return candidate.sessionID > current.sessionID +} + +func sameSupervisedLogSession(left supervisedLogSession, right supervisedLogSession) bool { + return left.sessionID == right.sessionID && left.startedAt.Equal(right.startedAt) +} + +func (session supervisedLogSession) contains(stream domain.LogStream) bool { + return session.sessionID != "" && stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == session.sessionID && stream.SessionStartedAt.Equal(session.startedAt) +} + +func (session supervisedLogSession) hasStream(streamID string) bool { + for _, stream := range session.streams { + if stream.ID == streamID { + return true + } + } + return false +} + +func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInstanceID string, active supervisedLogSession, historyLimit int) (map[string]uint64, error) { + emittedThrough := make(map[string]uint64, len(active.streams)) + if err := writeSSEJSON(w, "session", "", dto.LogStreamEventsSessionResponse{ServerInstanceID: serverInstanceID, LogSessionID: active.sessionID, SessionStartedAt: active.startedAt, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil { + return nil, err + } + for _, stream := range active.streams { + if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil { + return nil, err + } + } + if historyLimit == 0 || len(active.streams) == 0 { + return emittedThrough, nil + } + history, err := h.loadLogEventHistory(active.streams, historyLimit) + if err != nil { + return nil, err + } + for _, event := range history { + if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil { + return nil, err + } + if event.Entry.Seq > emittedThrough[event.Stream.ID] { + emittedThrough[event.Stream.ID] = event.Entry.Seq + } + } + return emittedThrough, nil +} + +func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, bool, service.LogEventSubscription, error) { var instance domain.ServerInstance var streams []domain.LogStream + var liveEligible bool var subscription service.LogEventSubscription var err error if h.enforceAuthorization { sessionID := bearerToken(r) instance, err = h.core.GetServerInstanceForSession(sessionID, r.PathValue("id")) if err != nil { - return domain.ServerInstance{}, nil, subscription, err - } - streams, err = h.core.ListLogStreamsForSession(sessionID, domain.LogStreamFilter{ServerInstanceID: instance.ID}) - if err != nil { - return domain.ServerInstance{}, nil, subscription, err + return domain.ServerInstance{}, nil, false, subscription, err } subscription, err = h.core.SubscribeLogEventsForSession(sessionID, instance.ID) + if err != nil { + return domain.ServerInstance{}, nil, false, subscription, err + } } else { instance, err = h.core.GetServerInstance(r.PathValue("id")) if err != nil { - return domain.ServerInstance{}, nil, subscription, err - } - streams, err = h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID}) - if err != nil { - return domain.ServerInstance{}, nil, subscription, err + return domain.ServerInstance{}, nil, false, subscription, err } subscription, err = h.core.SubscribeLogEvents(instance.ID) + if err != nil { + return domain.ServerInstance{}, nil, false, subscription, err + } } - return instance, streams, subscription, err + if err == nil { + streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID) + } + if err != nil && subscription.Close != nil { + subscription.Close() + } + return instance, streams, liveEligible, subscription, err +} + +func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string) ([]domain.LogStream, bool, error) { + instance, err := h.core.GetServerInstance(serverInstanceID) + if err != nil { + return nil, false, err + } + if instance.State != domain.ServerInstanceStateRunning || strings.TrimSpace(instance.RunEndpointID) == "" { + return nil, false, nil + } + endpoint, err := h.core.GetRunEndpoint(instance.RunEndpointID) + if err != nil { + return nil, false, err + } + if endpoint.Status != domain.RunEndpointStatusOnline { + return nil, false, nil + } + logSessionID := strings.TrimPrefix(instance.LifecycleProcessID, managedLogSessionIDPrefix) + if logSessionID == instance.LifecycleProcessID || strings.TrimSpace(logSessionID) == "" { + return nil, false, nil + } + streams, err := h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID}) + if err != nil { + return nil, false, err + } + current := make([]domain.LogStream, 0, len(streams)) + for _, stream := range streams { + if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == logSessionID { + current = append(current, stream) + } + } + return current, true, nil } func (h *coreHandlers) loadLogEventHistory(streams []domain.LogStream, limit int) ([]domain.LogStreamEvent, error) { diff --git a/platform/api/log_ingest_handlers_test.go b/platform/api/log_ingest_handlers_test.go index a7cc0b7..4456f04 100644 --- a/platform/api/log_ingest_handlers_test.go +++ b/platform/api/log_ingest_handlers_test.go @@ -1,18 +1,61 @@ package api import ( + "bufio" "context" + "io" "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" "browser.local/platform/domain" "browser.local/platform/dto" + "browser.local/platform/repo" + "browser.local/platform/service" "browser.local/platform/validator" ) +type logStreamListHookCore struct { + service.Core + once sync.Once + hook func() +} + +type ssePipeResponseWriter struct { + header http.Header + pipe *io.PipeWriter + status chan int + once sync.Once +} + +func newSSEPipeResponseWriter() (*ssePipeResponseWriter, *io.PipeReader) { + reader, writer := io.Pipe() + return &ssePipeResponseWriter{header: make(http.Header), pipe: writer, status: make(chan int, 1)}, reader +} + +func (writer *ssePipeResponseWriter) Header() http.Header { return writer.header } + +func (writer *ssePipeResponseWriter) WriteHeader(status int) { + writer.once.Do(func() { writer.status <- status }) +} + +func (writer *ssePipeResponseWriter) Write(payload []byte) (int, error) { + writer.WriteHeader(http.StatusOK) + return writer.pipe.Write(payload) +} + +func (writer *ssePipeResponseWriter) Flush() {} + +func (writer *ssePipeResponseWriter) Close() error { return writer.pipe.Close() } + +func (core *logStreamListHookCore) ListLogStreams(filter domain.LogStreamFilter) ([]domain.LogStream, error) { + core.once.Do(core.hook) + return core.Core.ListLogStreams(filter) +} + func TestLogIngestAPIWorkflow(t *testing.T) { router := newTestRouter() hello := createLogIngestAPIFixtures(t, router) @@ -57,6 +100,7 @@ func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) { hello := createLogIngestAPIFixtures(t, router) postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{ ID: "log-2", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stderr", + LogSessionID: "session-current", SessionStartedAt: time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC), StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", }) assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-1", "stdout", 1, 2, 0)), http.StatusOK) @@ -70,6 +114,213 @@ func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) { } } +func TestLogEventsSSEExcludesOlderSupervisedSessions(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + oldStream := dto.LogStreamCreateRequest{ID: "log-old", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stdout", LogSessionID: "session-old", SessionStartedAt: time.Date(2026, 7, 3, 11, 0, 0, 0, time.UTC), StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"} + postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", oldStream) + oldBatch := validLogBatchRequestForStream(t, hello.SessionToken, "log-old", "stdout", 1, 1, 0) + oldBatch.LogSessionID = "session-old" + oldBatch.SessionStartedAt = oldStream.SessionStartedAt + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", oldBatch), http.StatusOK) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK) + + recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10") + body := recorder.Body.String() + if !strings.Contains(body, `"logSessionId":"session-current"`) || strings.Contains(body, `"streamId":"log-old"`) { + t.Fatalf("expected only current session in live SSE, body=%s", body) + } +} + +func TestLogEventsSSEInitialSnapshotRequiresRunningOnlineProcessFact(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK) + + stopped := dto.RunLifecycleReportRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, Progress: dto.JobProgressBody{Percent: 100}, ExecutionResult: dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "stopped", ExitClassification: "requested-stop"}} + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", stopped), http.StatusOK) + assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10")) + + running := stopped + running.ManagedProcessID = "log-session:session-missing" + running.ObservationSeq = 1 + running.ObservedAt = time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC) + running.ExecutionResult = dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"} + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", running), http.StatusOK) + assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10")) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", dto.RunControlHeartbeatRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, Version: "0.1.0", Status: domain.RunEndpointStatusOffline, CapabilityFingerprint: "cap-logs", Capacity: dto.RunCapacityResponse{MaxJobs: 1}}), http.StatusOK) + assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10")) +} + +func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconnect(t *testing.T) { + router := newTestRouter() + hello := createLogIngestAPIFixtures(t, router) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=10", 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, "log", `"seq":1`) + assertSSEEvent(t, reader, "ready", `"streamCount":1`) + + statusReport := dto.RunLifecycleReportRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, Progress: dto.JobProgressBody{Percent: 100}, ExecutionResult: dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "stopped", ExitClassification: "requested-stop"}} + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", statusReport), http.StatusOK) + name, data := readSSEEvent(t, reader) + if name != "session" || strings.Contains(data, `"logSessionId"`) || !strings.Contains(data, `"streamCount":0`) { + t.Fatalf("expected stopped process to emit an empty session boundary, name=%q data=%s", name, data) + } + + nextStartedAt := time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC) + next := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stdout", "stdout", 1, 1, 20) + next.LogSessionID = "session-next" + next.SessionStartedAt = nextStartedAt + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", next), http.StatusOK) + + statusReport.ManagedProcessID = "log-session:session-next" + statusReport.ObservationSeq = 1 + statusReport.ObservedAt = nextStartedAt + statusReport.ExecutionResult = dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"} + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", statusReport), http.StatusOK) + assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`) + assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`) + assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`) +} + +func TestLogEventsSSEOrdersReplaySessionSwitchAndAdditionalStreamWithoutDuplicates(t *testing.T) { + core := service.NewCoreService(repo.NewMemoryStore()) + if err := core.SeedLocalPlatformAdmin(); err != nil { + t.Fatalf("seed platform admin: %v", err) + } + setupRouter := NewTestRouterWithCore(core) + hello := createLogIngestAPIFixtures(t, setupRouter) + initial := validLogBatchRequest(t, hello.SessionToken, 1, 1) + hookResult := make(chan error, 1) + hooked := &logStreamListHookCore{Core: core, hook: func() { + _, err := core.IngestLogBatch(initial.ToDomain()) + hookResult <- err + }} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=10", nil).WithContext(ctx) + streamWriter, streamReader := newSSEPipeResponseWriter() + done := make(chan struct{}) + go func() { + NewTestRouterWithCore(hooked).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, "log", `"seq":1`) + assertSSEEvent(t, reader, "ready", `"streamCount":1`) + if err := <-hookResult; err != nil { + t.Fatalf("ingest during stream snapshot: %v", err) + } + + nextStartedAt := time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC) + next := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stdout", "stdout", 1, 1, 20) + next.LogSessionID = "session-next" + next.SessionStartedAt = nextStartedAt + if _, err := core.IngestLogBatch(next.ToDomain()); err != nil { + t.Fatalf("ingest next session stdout: %v", err) + } + if _, err := core.ReportRunLifecycle(domain.RunLifecycleReport{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, ManagedProcessID: "log-session:session-next", ObservationSeq: 1, ObservedAt: nextStartedAt, ExecutionResult: domain.JobExecutionResult{Kind: "process", ProcessState: "running"}}); err != nil { + t.Fatalf("report next managed process: %v", err) + } + assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`) + assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`) + assertSSEEvent(t, reader, "log", `"seq":1`) + + stderr := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stderr", "stderr", 1, 1, 21) + stderr.LogSessionID = "session-next" + stderr.SessionStartedAt = nextStartedAt + if _, err := core.IngestLogBatch(stderr.ToDomain()); err != nil { + t.Fatalf("ingest next session stderr: %v", err) + } + assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stderr"`) + assertSSEEvent(t, reader, "log", `"streamKey":"stderr"`) + + delayedOld := validLogBatchRequest(t, hello.SessionToken, 2, 2) + if _, err := core.IngestLogBatch(delayedOld.ToDomain()); err != nil { + t.Fatalf("ingest delayed old session batch: %v", err) + } + nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 22) + nextLive.LogSessionID = "session-next" + nextLive.SessionStartedAt = nextStartedAt + if _, err := core.IngestLogBatch(nextLive.ToDomain()); err != nil { + t.Fatalf("ingest next session live append: %v", err) + } + assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`) +} + +func assertSSEEvent(t *testing.T, reader *bufio.Reader, wantName string, wantData string) { + t.Helper() + name, data := readSSEEvent(t, reader) + if name != wantName || !strings.Contains(data, wantData) { + t.Fatalf("unexpected SSE event: name=%q data=%s; want name=%q containing %s", name, data, wantName, wantData) + } +} + +func assertEmptyInitialLogSession(t *testing.T, recorder *httptest.ResponseRecorder) { + t.Helper() + assertStatus(t, recorder, http.StatusOK) + body := recorder.Body.String() + if !strings.Contains(body, "event: session") || strings.Contains(body, `"logSessionId"`) || strings.Contains(body, "event: log") || !strings.Contains(body, `"streamCount":0`) { + t.Fatalf("expected an empty initial live session, body=%s", body) + } +} + +func readSSEEvent(t *testing.T, reader *bufio.Reader) (string, string) { + t.Helper() + for { + name, data := "", "" + for { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("read SSE event: %v", err) + } + line = strings.TrimRight(line, "\r\n") + if line == "" { + if name != "" { + return name, data + } + break + } + if strings.HasPrefix(line, "event: ") { + name = strings.TrimPrefix(line, "event: ") + } + if strings.HasPrefix(line, "data: ") { + data = strings.TrimPrefix(line, "data: ") + } + } + } +} + func performCancelledSSE(t *testing.T, router http.Handler, path string) *httptest.ResponseRecorder { t.Helper() ctx, cancel := context.WithCancel(context.Background()) @@ -110,15 +361,22 @@ func createLogIngestAPIFixtures(t *testing.T, router http.Handler) dto.RunContro hello := decodeBody[dto.RunControlHelloResponse](t, performRunControlHello(t, router, helloRequest)) adminSession := createAdminSession(t, router) postJSON[dto.GamePluginResponse](t, router, "/api/v1/game-plugins", validGamePluginRequest()) - postJSONWithAuth[dto.ServerInstanceResponse](t, router, "/api/v1/server-instances", dto.ServerInstanceCreateRequest{ID: "server-1", PluginID: "server.scum", RunEndpointID: "run-local", Name: "SCUM #1"}, adminSession) + postJSONWithAuth[dto.ServerInstanceResponse](t, router, "/api/v1/server-instances", dto.ServerInstanceCreateRequest{ID: "server-1", PluginID: "server.scum", RunEndpointID: "run-local", Name: "SCUM #1", State: domain.ServerInstanceStateRunning}, adminSession) postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{ ID: "log-1", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stdout", + LogSessionID: "session-current", + SessionStartedAt: time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC), StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", }) + assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", dto.RunLifecycleReportRequest{ + RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, + ManagedProcessID: "log-session:session-current", ObservationSeq: 1, ObservedAt: time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC), + ExecutionResult: dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"}, + }), http.StatusOK) return hello } @@ -146,6 +404,8 @@ func validLogBatchRequestForStream(t *testing.T, sessionToken string, streamID s ServerInstanceID: "server-1", StreamKey: streamKey, Source: domain.LogStreamSourceProcess, + LogSessionID: "session-current", + SessionStartedAt: time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC), FirstSeq: firstSeq, LastSeq: lastSeq, Compression: "none", diff --git a/platform/domain/log_ingest.go b/platform/domain/log_ingest.go index 3ae9078..1c28f4a 100644 --- a/platform/domain/log_ingest.go +++ b/platform/domain/log_ingest.go @@ -18,6 +18,8 @@ type LogBatchIngest struct { ServerInstanceID string StreamKey string Source LogStreamSource + LogSessionID string + SessionStartedAt time.Time FirstSeq uint64 LastSeq uint64 Compression string diff --git a/platform/domain/resources.go b/platform/domain/resources.go index 8bcc7af..0af779c 100644 --- a/platform/domain/resources.go +++ b/platform/domain/resources.go @@ -1507,6 +1507,8 @@ type LogStream struct { ServerInstanceID string Source LogStreamSource StreamKey string + LogSessionID string + SessionStartedAt time.Time LatestSeq uint64 StorageBackend LogStorageBackend RetentionPolicy string diff --git a/platform/dto/log_ingest.go b/platform/dto/log_ingest.go index f3e736e..e78528d 100644 --- a/platform/dto/log_ingest.go +++ b/platform/dto/log_ingest.go @@ -22,6 +22,8 @@ type LogBatchIngestRequest struct { ServerInstanceID string `json:"serverInstanceId"` StreamKey string `json:"streamKey"` Source domain.LogStreamSource `json:"source"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` FirstSeq uint64 `json:"firstSeq"` LastSeq uint64 `json:"lastSeq"` Compression string `json:"compression"` @@ -74,6 +76,8 @@ type LogStreamEventResponse struct { StreamID string `json:"streamId"` Source domain.LogStreamSource `json:"source"` StreamKey string `json:"streamKey"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` LatestSeq uint64 `json:"latestSeq"` Entry LogEntryBody `json:"entry"` } @@ -84,6 +88,14 @@ type LogStreamEventsReadyResponse struct { ServerTime time.Time `json:"serverTime"` } +type LogStreamEventsSessionResponse struct { + ServerInstanceID string `json:"serverInstanceId"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` + StreamCount int `json:"streamCount"` + ServerTime time.Time `json:"serverTime"` +} + func (request LogBatchIngestRequest) ToDomain() domain.LogBatchIngest { return domain.LogBatchIngest{ RunEndpointID: request.RunEndpointID, @@ -92,6 +104,8 @@ func (request LogBatchIngestRequest) ToDomain() domain.LogBatchIngest { ServerInstanceID: request.ServerInstanceID, StreamKey: request.StreamKey, Source: request.Source, + LogSessionID: request.LogSessionID, + SessionStartedAt: request.SessionStartedAt, FirstSeq: request.FirstSeq, LastSeq: request.LastSeq, Compression: request.Compression, @@ -145,6 +159,8 @@ func LogStreamEventFromDomain(event domain.LogStreamEvent) LogStreamEventRespons StreamID: event.Stream.ID, Source: event.Stream.Source, StreamKey: event.Stream.StreamKey, + LogSessionID: event.Stream.LogSessionID, + SessionStartedAt: event.Stream.SessionStartedAt, LatestSeq: event.LatestSeq, Entry: logEntryFromDomain(event.Entry), } diff --git a/platform/dto/resources.go b/platform/dto/resources.go index 67665cc..2b88622 100644 --- a/platform/dto/resources.go +++ b/platform/dto/resources.go @@ -893,6 +893,8 @@ type LogStreamCreateRequest struct { ServerInstanceID string `json:"serverInstanceId"` Source domain.LogStreamSource `json:"source"` StreamKey string `json:"streamKey"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` StorageBackend domain.LogStorageBackend `json:"storageBackend"` RetentionPolicy string `json:"retentionPolicy"` } @@ -902,6 +904,8 @@ type LogStreamResponse struct { ServerInstanceID string `json:"serverInstanceId"` Source domain.LogStreamSource `json:"source"` StreamKey string `json:"streamKey"` + LogSessionID string `json:"logSessionId,omitempty"` + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty"` LatestSeq uint64 `json:"latestSeq"` StorageBackend domain.LogStorageBackend `json:"storageBackend"` RetentionPolicy string `json:"retentionPolicy"` @@ -1363,6 +1367,8 @@ func (request LogStreamCreateRequest) ToDomain() domain.LogStream { ServerInstanceID: request.ServerInstanceID, Source: request.Source, StreamKey: request.StreamKey, + LogSessionID: request.LogSessionID, + SessionStartedAt: request.SessionStartedAt, StorageBackend: request.StorageBackend, RetentionPolicy: request.RetentionPolicy, } @@ -1938,6 +1944,8 @@ func LogStreamFromDomain(stream domain.LogStream) LogStreamResponse { ServerInstanceID: stream.ServerInstanceID, Source: stream.Source, StreamKey: stream.StreamKey, + LogSessionID: stream.LogSessionID, + SessionStartedAt: stream.SessionStartedAt, LatestSeq: stream.LatestSeq, StorageBackend: stream.StorageBackend, RetentionPolicy: stream.RetentionPolicy, diff --git a/platform/model/resources.go b/platform/model/resources.go index 419a44f..d52be0a 100644 --- a/platform/model/resources.go +++ b/platform/model/resources.go @@ -415,6 +415,10 @@ type LogStream struct { Source domain.LogStreamSource `json:"source" db:"source"` // StreamKey is stable within the server instance. StreamKey string `json:"streamKey" db:"stream_key"` + // LogSessionID groups plugin-declared supervised-process streams for one process generation. + LogSessionID string `json:"logSessionId,omitempty" db:"log_session_id"` + // SessionStartedAt is Run's persisted start timestamp for the supervised process generation. + SessionStartedAt time.Time `json:"sessionStartedAt,omitempty" db:"session_started_at"` // LatestSeq is the latest accepted sequence number. LatestSeq uint64 `json:"latestSeq" db:"latest_seq"` // StorageBackend identifies the log body backend. diff --git a/platform/service/distribution_build_execution.go b/platform/service/distribution_build_execution.go index b8f202d..3de782b 100644 --- a/platform/service/distribution_build_execution.go +++ b/platform/service/distribution_build_execution.go @@ -66,7 +66,17 @@ func (svc *CoreService) enqueueDistributionBuild(job domain.Job) { delete(svc.distributionBuilds, job.ID) svc.distributionBuildMu.Unlock() }() - _ = svc.executeDistributionBuild(job) + defer func() { + if recover() != nil { + _ = svc.failDistributionBuildJob(job, "platform builder failed unexpectedly") + } + }() + if err := svc.executeDistributionBuild(job); err != nil { + // executeDistributionBuild may fail after persisting the running state. + // A second terminalization attempt is idempotent when the specific + // failure path already marked the job failed. + _ = svc.failDistributionBuildJob(job, "platform builder failed before completion") + } }() } @@ -99,7 +109,7 @@ func (svc *CoreService) executeDistributionBuild(job domain.Job) error { payload, buildErr = builder.Build(input) } if buildErr != nil { - return svc.failDistributionBuildJob(job, builderJobFailureMessage(buildErr)) + return svc.failDistributionBuildJob(job, builderJobFailureMessage(buildErr, input.AuthKey)) } if _, err := svc.platformDistributionBuildInput(job); err != nil { return svc.failDistributionBuildJob(job, "platform builder discarded output because the component key is no longer current") diff --git a/platform/service/distribution_build_execution_test.go b/platform/service/distribution_build_execution_test.go index 4c5d5d3..b8b5498 100644 --- a/platform/service/distribution_build_execution_test.go +++ b/platform/service/distribution_build_execution_test.go @@ -25,6 +25,34 @@ type progressDistributionBuilder struct { payload []byte } +type failingBuildProgressStore struct { + repo.Store + buildJobs repo.ClientManagerBuildJobRepository +} + +func (store failingBuildProgressStore) ClientManagerBuildJobs() repo.ClientManagerBuildJobRepository { + return store.buildJobs +} + +type failWhenDistributionJobRunningRepository struct { + repo.ClientManagerBuildJobRepository + jobs repo.JobRepository + err error +} + +func (repository failWhenDistributionJobRunningRepository) List(filter domain.ClientManagerBuildJobFilter) ([]domain.ClientManagerBuildJob, error) { + jobs, err := repository.jobs.List(domain.JobFilter{ServerInstanceID: filter.ServerInstanceID}) + if err != nil { + return nil, err + } + for _, job := range jobs { + if job.Capability == domain.JobCapabilityDistributionBuild && job.State == domain.JobStateRunning { + return nil, repository.err + } + } + return repository.ClientManagerBuildJobRepository.List(filter) +} + func (builder captureDistributionBuilder) Readiness() (bool, string) { return true, "" } @@ -379,6 +407,49 @@ func TestCoreServiceProjectsPlatformBuilderProgressBeforeCompletion(t *testing.T completeDistributionBuild(t, svc, distribution, nil) } +func TestCoreServiceTerminalizesPlatformBuildWhenRunningProjectionFails(t *testing.T) { + svc, session, instance := newDistributionTestFixture(t) + const leakedDetail = "sensitive-builder-token /private/platform/build/input/auth-key" + baseStore := svc.store + svc.store = failingBuildProgressStore{ + Store: baseStore, + buildJobs: failWhenDistributionJobRunningRepository{ + ClientManagerBuildJobRepository: baseStore.ClientManagerBuildJobs(), + jobs: baseStore.Jobs(), + err: errors.New(leakedDetail), + }, + } + + distribution, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{ + ServerInstanceID: instance.ID, + TargetOS: "linux", + TargetArch: "amd64", + IdempotencyKey: "running-projection-failure-terminalizes", + }) + if err != nil { + t.Fatalf("queue platform distribution: %v", err) + } + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + job, jobErr := svc.GetJob(distribution.BuildJobID) + updated, distributionErr := svc.store.RunDistributions().Get(distribution.ID) + if jobErr != nil || distributionErr != nil { + t.Fatalf("read failed build state: jobErr=%v distributionErr=%v", jobErr, distributionErr) + } + if isTerminalJobState(job.State) { + if job.State != domain.JobStateFailed || updated.Status != domain.DistributionStatusFailed { + t.Fatalf("projection failure did not end in failed state: job=%+v distribution=%+v", job, updated) + } + if strings.Contains(job.Progress.Message, "sensitive-builder-token") || strings.Contains(job.Progress.Message, "/private/") || strings.Contains(job.TerminalFingerprint, "sensitive-builder-token") { + t.Fatalf("terminal build state leaked internal failure details: %+v", job) + } + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("platform build remained non-terminal after running projection failure") +} + func TestCoreServiceDiscardsBuildCompletedAfterKeyReset(t *testing.T) { svc, session, instance := newDistributionTestFixture(t) inputs := make(chan domain.DistributionBuildInput, 1) diff --git a/platform/service/distribution_builder.go b/platform/service/distribution_builder.go index 178f3ea..1450a93 100644 --- a/platform/service/distribution_builder.go +++ b/platform/service/distribution_builder.go @@ -553,11 +553,11 @@ func redactBuilderHostPaths(line string) string { return strings.Join(fields, " ") } -func builderJobFailureMessage(err error) string { +func builderJobFailureMessage(err error, sensitiveValues ...string) string { if err == nil { return "platform builder failed" } - message := strings.TrimSpace(err.Error()) + message := safeBuilderFailure([]byte(err.Error()), sensitiveValues...) if message == "" { return "platform builder failed" } diff --git a/platform/service/distribution_builder_test.go b/platform/service/distribution_builder_test.go index 3c15552..5bbadd4 100644 --- a/platform/service/distribution_builder_test.go +++ b/platform/service/distribution_builder_test.go @@ -289,6 +289,14 @@ func TestDockerDistributionBuilderRedactsFailureAndTimeout(t *testing.T) { } } +func TestBuilderJobFailureMessageRedactsSensitiveValuesAndHostPaths(t *testing.T) { + const secret = "sensitive-component-key" + message := builderJobFailureMessage(errors.New(secret+" /private/platform/build/input/auth-key"), secret) + if strings.Contains(message, secret) || strings.Contains(message, "/private/") || strings.Contains(message, "auth-key") { + t.Fatalf("builder job failure leaked sensitive details: %s", message) + } +} + func TestPackageClientManagerDistributionProducesProtectedArchives(t *testing.T) { for _, packageFormat := range []string{"zip", "tar.gz"} { t.Run(packageFormat, func(t *testing.T) { diff --git a/platform/service/log_events.go b/platform/service/log_events.go index 21f98d2..992862d 100644 --- a/platform/service/log_events.go +++ b/platform/service/log_events.go @@ -8,14 +8,28 @@ import ( const logEventSubscriberBuffer = 512 +type LogEventSubscriptionEventKind string + +const ( + LogEventSubscriptionEventLog LogEventSubscriptionEventKind = "log" + LogEventSubscriptionEventProcessState LogEventSubscriptionEventKind = "process-state" +) + +type LogEventSubscriptionEvent struct { + Kind LogEventSubscriptionEventKind + LogEvent domain.LogStreamEvent + ServerInstanceID string + ProcessState domain.ServerInstanceState +} + type LogEventSubscription struct { - Events <-chan domain.LogStreamEvent + Events <-chan LogEventSubscriptionEvent Close func() } type logEventSubscriber struct { serverInstanceID string - events chan domain.LogStreamEvent + events chan LogEventSubscriptionEvent } func (svc *CoreService) SubscribeLogEvents(serverInstanceID string) (LogEventSubscription, error) { @@ -26,7 +40,7 @@ func (svc *CoreService) SubscribeLogEvents(serverInstanceID string) (LogEventSub if _, err := svc.store.ServerInstances().Get(serverInstanceID); err != nil { return LogEventSubscription{}, err } - events := make(chan domain.LogStreamEvent, logEventSubscriberBuffer) + events := make(chan LogEventSubscriptionEvent, logEventSubscriberBuffer) svc.logEventMu.Lock() svc.logEventSubscriberSeq++ id := svc.logEventSubscriberSeq @@ -55,18 +69,36 @@ func (svc *CoreService) publishLogEvents(stream domain.LogStream, entries []doma if len(entries) == 0 { return } - events := make([]domain.LogStreamEvent, len(entries)) + events := make([]LogEventSubscriptionEvent, len(entries)) for index, entry := range entries { - events[index] = domain.CopyLogStreamEvent(domain.LogStreamEvent{ - ServerInstanceID: stream.ServerInstanceID, - Stream: stream, - Entry: entry, - LatestSeq: stream.LatestSeq, - }) + events[index] = LogEventSubscriptionEvent{ + Kind: LogEventSubscriptionEventLog, + LogEvent: domain.CopyLogStreamEvent(domain.LogStreamEvent{ + ServerInstanceID: stream.ServerInstanceID, + Stream: stream, + Entry: entry, + LatestSeq: stream.LatestSeq, + }), + } + } + svc.publishLogSubscriptionEvents(stream.ServerInstanceID, events) +} + +func (svc *CoreService) publishLogProcessState(instance domain.ServerInstance) { + svc.publishLogSubscriptionEvents(instance.ID, []LogEventSubscriptionEvent{{ + Kind: LogEventSubscriptionEventProcessState, + ServerInstanceID: instance.ID, + ProcessState: instance.State, + }}) +} + +func (svc *CoreService) publishLogSubscriptionEvents(serverInstanceID string, events []LogEventSubscriptionEvent) { + if len(events) == 0 { + return } svc.logEventMu.Lock() for id, subscriber := range svc.logEventSubscribers { - if subscriber.serverInstanceID != stream.ServerInstanceID { + if subscriber.serverInstanceID != serverInstanceID { continue } dropped := false diff --git a/platform/service/log_ingest.go b/platform/service/log_ingest.go index 19f46c0..8e688a9 100644 --- a/platform/service/log_ingest.go +++ b/platform/service/log_ingest.go @@ -20,6 +20,9 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil { return domain.LogBatchIngestResult{}, err } + lock := svc.logIngestLock(batch.ServerInstanceID) + lock.Lock() + defer lock.Unlock() stamp := svc.now() stream, err := svc.store.LogStreams().Get(batch.LogStreamID) @@ -58,7 +61,7 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB } storedBatch := domain.CopyLogBatchIngest(batch) - sanitizeGamePlayerNetworkFields(&storedBatch) + sanitizeLogNetworkFields(&storedBatch) record := domain.CopyLogBatchRecord(domain.LogBatchRecord{ Checksum: batch.Checksum, FirstSeq: batch.FirstSeq, @@ -96,7 +99,7 @@ func (svc *CoreService) ensureJobLogStreamForBatch(batch domain.LogBatchIngest, if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID { return validationError("log batch job scope does not match stream") } - return svc.ensureJobLogStreams(job, stamp) + return svc.ensureJobLogStreamsUnlocked(job, stamp) } func (svc *CoreService) ensureLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error { @@ -107,7 +110,7 @@ func (svc *CoreService) ensureLogStreamForBatch(batch domain.LogBatchIngest, sta if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID { return validationError("log batch job scope does not match stream") } - return svc.ensureJobLogStreams(job, stamp) + return svc.ensureJobLogStreamsUnlocked(job, stamp) } if !errors.Is(err, repo.ErrNotFound) || !strings.HasPrefix(jobID, "autonomous-") { return err @@ -120,8 +123,14 @@ func (svc *CoreService) ensureRunLogStreamForBatch(batch domain.LogBatchIngest, if batch.Source != domain.LogStreamSourceProcess && batch.Source != domain.LogStreamSourceFile && batch.Source != domain.LogStreamSourceManagementProgram { return repo.ErrNotFound } - if batch.LogStreamID != runLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.StreamKey) && !legacyAutonomousLogStream(batch) { - return repo.ErrNotFound + expectedStreamID := runLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.StreamKey) + if batch.LogSessionID != "" { + expectedStreamID = runSessionLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.LogSessionID, batch.StreamKey) + } + if batch.LogStreamID != expectedStreamID { + if batch.LogSessionID != "" || !legacyAutonomousLogStream(batch) { + return repo.ErrNotFound + } } instance, err := svc.store.ServerInstances().Get(batch.ServerInstanceID) if err != nil { @@ -130,7 +139,7 @@ func (svc *CoreService) ensureRunLogStreamForBatch(batch domain.LogBatchIngest, if instance.RunEndpointID != batch.RunEndpointID { return validationError("server instance run endpoint must match log batch endpoint") } - _, err = svc.CreateLogStream(domain.LogStream{ID: batch.LogStreamID, ServerInstanceID: batch.ServerInstanceID, Source: batch.Source, StreamKey: batch.StreamKey, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp}) + _, err = svc.createLogStream(domain.LogStream{ID: batch.LogStreamID, ServerInstanceID: batch.ServerInstanceID, Source: batch.Source, StreamKey: batch.StreamKey, LogSessionID: batch.LogSessionID, SessionStartedAt: batch.SessionStartedAt, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp}) if errors.Is(err, repo.ErrDuplicate) { return nil } @@ -166,18 +175,16 @@ func logBatchRecordMatches(record domain.LogBatchRecord, batch domain.LogBatchIn return false } -// sanitizeGamePlayerNetworkFields removes raw network material before the durable log body is written. -func sanitizeGamePlayerNetworkFields(batch *domain.LogBatchIngest) { +// sanitizeLogNetworkFields removes raw network material before the durable log body is written. +func sanitizeLogNetworkFields(batch *domain.LogBatchIngest) { for index := range batch.Entries { fields := batch.Entries[index].Fields if fields == nil { continue } - if fields["eventType"] == "scum.login" { - delete(fields, "networkFingerprint") - delete(fields, "ip") - delete(fields, "ipAddress") - } + delete(fields, "networkFingerprint") + delete(fields, "ip") + delete(fields, "ipAddress") } } @@ -249,5 +256,8 @@ func validateLogBatchStream(batch domain.LogBatchIngest, stream domain.LogStream if stream.Source != batch.Source { return validationError("source must match stream") } + if stream.LogSessionID != batch.LogSessionID || !stream.SessionStartedAt.Equal(batch.SessionStartedAt) { + return validationError("log session metadata must match stream") + } return nil } diff --git a/platform/service/log_ingest_privacy_test.go b/platform/service/log_ingest_privacy_test.go new file mode 100644 index 0000000..125d6ca --- /dev/null +++ b/platform/service/log_ingest_privacy_test.go @@ -0,0 +1,31 @@ +package service + +import ( + "testing" + + "browser.local/platform/domain" +) + +func TestSanitizeLogNetworkFieldsIsGameAgnostic(t *testing.T) { + batch := domain.LogBatchIngest{Entries: []domain.LogEntry{ + {Fields: map[string]string{ + "eventType": "game.session.opened", + "networkFingerprint": "fingerprint", + "ip": "192.0.2.1", + "ipAddress": "2001:db8::1", + "playerId": "player-1", + }}, + }} + + sanitizeLogNetworkFields(&batch) + + fields := batch.Entries[0].Fields + for _, key := range []string{"networkFingerprint", "ip", "ipAddress"} { + if _, exists := fields[key]; exists { + t.Fatalf("expected %s to be removed", key) + } + } + if fields["playerId"] != "player-1" { + t.Fatal("expected unrelated fields to be preserved") + } +} diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index 39273a5..27824de 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -3,6 +3,7 @@ package service import ( "path/filepath" "strings" + "sync" "testing" "time" @@ -74,7 +75,7 @@ func TestCoreServicePublishesLogEventsForAcceptedBatch(t *testing.T) { } select { case event := <-subscription.Events: - if event.Stream.ID != "log-1" || event.Entry.Seq != 1 || event.LatestSeq != 1 { + if event.Kind != LogEventSubscriptionEventLog || event.LogEvent.Stream.ID != "log-1" || event.LogEvent.Entry.Seq != 1 || event.LogEvent.LatestSeq != 1 { t.Fatalf("unexpected log event: %+v", event) } case <-time.After(time.Second): @@ -91,6 +92,76 @@ func TestCoreServicePublishesLogEventsForAcceptedBatch(t *testing.T) { } } +func TestCoreServicePersistsAndEnforcesImmutableProcessLogSessionMetadata(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) + streamID := runSessionLogStreamID("run-local", "server-1", "session-a", "stdout") + batch := validLogBatch(t, sessionToken, 1, 1) + batch.LogStreamID = streamID + batch.LogSessionID = "session-a" + batch.SessionStartedAt = startedAt + if _, err := svc.IngestLogBatch(batch); err != nil { + t.Fatalf("ingest session-scoped process batch: %v", err) + } + stream, err := svc.GetLogStream(streamID) + if err != nil || stream.LogSessionID != "session-a" || !stream.SessionStartedAt.Equal(startedAt) { + t.Fatalf("session metadata was not persisted: stream=%+v err=%v", stream, err) + } + + conflict := validLogBatch(t, sessionToken, 2, 2) + conflict.LogStreamID = streamID + conflict.LogSessionID = "session-a" + conflict.SessionStartedAt = startedAt.Add(time.Second) + if _, err := svc.IngestLogBatch(conflict); err == nil || !strings.Contains(err.Error(), "metadata must match") { + t.Fatalf("expected immutable stream metadata rejection, got %v", err) + } + + legacy := createLogStreamFixture(t, svc) + legacyBatch := validLogBatch(t, sessionToken, 1, 1) + legacyBatch.LogStreamID = legacy.ID + legacyBatch.LogSessionID = "session-a" + legacyBatch.SessionStartedAt = startedAt + if _, err := svc.IngestLogBatch(legacyBatch); err == nil || !strings.Contains(err.Error(), "metadata must match") { + t.Fatalf("expected legacy stream to reject attached session metadata, got %v", err) + } +} + +func TestCoreServiceSerializesConsistentSessionStreamCreation(t *testing.T) { + svc, _ := newRegisteredLogIngestService(t) + startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) + streams := []domain.LogStream{ + {ID: "session-stream-stdout", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stdout", LogSessionID: "session-a", SessionStartedAt: startedAt, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}, + {ID: "session-stream-stderr", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stderr", LogSessionID: "session-a", SessionStartedAt: startedAt, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}, + } + errorsByStream := make(chan error, len(streams)) + var wait sync.WaitGroup + for _, stream := range streams { + stream := stream + wait.Add(1) + go func() { + defer wait.Done() + _, err := svc.CreateLogStream(stream) + errorsByStream <- err + }() + } + wait.Wait() + close(errorsByStream) + for err := range errorsByStream { + if err != nil { + t.Fatalf("create consistent session stream: %v", err) + } + } + + _, err := svc.CreateLogStream(domain.LogStream{ID: "session-stream-conflict", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "console", LogSessionID: "session-a", SessionStartedAt: startedAt.Add(time.Second), StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}) + if err == nil || !strings.Contains(err.Error(), "conflicts") { + t.Fatalf("expected conflicting session timestamp rejection, got %v", err) + } + _, err = svc.CreateLogStream(domain.LogStream{ID: "session-file-tail", ServerInstanceID: "server-1", Source: domain.LogStreamSourceFile, StreamKey: "file", LogSessionID: "session-file", SessionStartedAt: startedAt, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}) + if err == nil || !strings.Contains(err.Error(), "only valid for process") { + t.Fatalf("expected file-tail session metadata rejection, got %v", err) + } +} + func TestCoreServiceLogBatchDuplicateAck(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) createLogStreamFixture(t, svc) @@ -346,6 +417,17 @@ func TestCoreServiceAcceptsLegacyAutonomousJobLogStreamWithoutPlatformJob(t *tes } } +func TestCoreServiceRejectsSessionMetadataOnLegacyAutonomousStreamID(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + batch := validLogBatch(t, sessionToken, 1, 1) + batch.LogStreamID = jobLogStreamID("autonomous-bootstrap-start", "stdout") + batch.LogSessionID = "session-a" + batch.SessionStartedAt = time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) + if _, err := svc.IngestLogBatch(batch); err == nil { + t.Fatal("expected session-scoped batch with legacy autonomous stream ID to be rejected") + } +} + func TestCoreServiceRejectsOutOfOrderAndConflictingLogBatches(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) createLogStreamFixture(t, svc) diff --git a/platform/service/resources.go b/platform/service/resources.go index 8ad5a55..43f6ea2 100644 --- a/platform/service/resources.go +++ b/platform/service/resources.go @@ -233,6 +233,7 @@ type CoreService struct { bridgeMu sync.Mutex bridgeSeq uint64 logStore LogBodyStore + logIngestMu [64]sync.Mutex logEventMu sync.Mutex logEventSubscribers map[uint64]logEventSubscriber logEventSubscriberSeq uint64 @@ -2476,6 +2477,13 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) { } func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) error { + lock := svc.logIngestLock(job.ServerInstanceID) + lock.Lock() + defer lock.Unlock() + return svc.ensureJobLogStreamsUnlocked(job, stamp) +} + +func (svc *CoreService) ensureJobLogStreamsUnlocked(job domain.Job, stamp time.Time) error { if strings.TrimSpace(job.ServerInstanceID) == "" || strings.TrimSpace(job.ID) == "" { return nil } @@ -2524,7 +2532,7 @@ func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) err CreatedAt: stamp, UpdatedAt: stamp, } - if _, err := svc.CreateLogStream(stream); err != nil && !errors.Is(err, repo.ErrDuplicate) { + if _, err := svc.createLogStream(stream); err != nil && !errors.Is(err, repo.ErrDuplicate) { return err } } @@ -2539,6 +2547,10 @@ func runLogStreamID(runEndpointID string, serverInstanceID string, streamKey str return fmt.Sprintf("run.%s.%s.%s", runEndpointID, serverInstanceID, streamKey) } +func runSessionLogStreamID(runEndpointID string, serverInstanceID string, logSessionID string, streamKey string) string { + return fmt.Sprintf("run.%s.%s.%s.%s", runEndpointID, serverInstanceID, logSessionID, streamKey) +} + func (svc *CoreService) GetJob(id string) (domain.Job, error) { job, err := svc.store.Jobs().Get(id) if err != nil { @@ -2587,6 +2599,22 @@ func (svc *CoreService) ListArtifacts(filter domain.ArtifactFilter) ([]domain.Ar } func (svc *CoreService) CreateLogStream(stream domain.LogStream) (domain.LogStream, error) { + lock := svc.logIngestLock(stream.ServerInstanceID) + lock.Lock() + defer lock.Unlock() + return svc.createLogStream(stream) +} + +func (svc *CoreService) logIngestLock(serverInstanceID string) *sync.Mutex { + hash := uint32(2166136261) + for index := 0; index < len(serverInstanceID); index++ { + hash ^= uint32(serverInstanceID[index]) + hash *= 16777619 + } + return &svc.logIngestMu[hash%uint32(len(svc.logIngestMu))] +} + +func (svc *CoreService) createLogStream(stream domain.LogStream) (domain.LogStream, error) { instance, err := svc.store.ServerInstances().Get(stream.ServerInstanceID) if err != nil { return domain.LogStream{}, fmt.Errorf("get server instance dependency: %w", err) @@ -2604,12 +2632,31 @@ func (svc *CoreService) CreateLogStream(stream domain.LogStream) (domain.LogStre if err := validator.ValidateLogStream(stream); err != nil { return domain.LogStream{}, err } + if err := svc.validateLogStreamSession(stream); err != nil { + return domain.LogStream{}, err + } if err := svc.store.LogStreams().Create(stream); err != nil { return domain.LogStream{}, err } return domain.CopyLogStream(stream), nil } +func (svc *CoreService) validateLogStreamSession(stream domain.LogStream) error { + if stream.LogSessionID == "" { + return nil + } + streams, err := svc.store.LogStreams().List(domain.LogStreamFilter{ServerInstanceID: stream.ServerInstanceID}) + if err != nil { + return err + } + for _, existing := range streams { + if existing.LogSessionID == stream.LogSessionID && !existing.SessionStartedAt.Equal(stream.SessionStartedAt) { + return validationError("log session metadata conflicts with an existing stream") + } + } + return nil +} + func (svc *CoreService) GetLogStream(id string) (domain.LogStream, error) { return svc.store.LogStreams().Get(id) } diff --git a/platform/service/server_lifecycle.go b/platform/service/server_lifecycle.go index a12ae29..b2f4a83 100644 --- a/platform/service/server_lifecycle.go +++ b/platform/service/server_lifecycle.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "time" "browser.local/platform/domain" "browser.local/platform/repo" @@ -255,6 +256,14 @@ func (svc *CoreService) dispatchExistingServerLifecycle(command domain.ServerLif if err := svc.validateRunnableEndpoint(endpoint, domain.LifecycleCapabilityForAction(action)); err != nil { return domain.ServerLifecycleResult{}, err } + if action == domain.ServerLifecycleActionStart { + instance.LifecycleProcessID = "" + instance.LifecycleObservationSeq = 0 + instance.LifecycleObservedAt = time.Time{} + if err := svc.store.ServerInstances().Update(instance); err != nil { + return domain.ServerLifecycleResult{}, err + } + } job, err := svc.dispatchLifecycleJob(instance, action, command.IdempotencyKey) if err != nil { diff --git a/platform/service/server_lifecycle_projection.go b/platform/service/server_lifecycle_projection.go index c41783d..f27ac01 100644 --- a/platform/service/server_lifecycle_projection.go +++ b/platform/service/server_lifecycle_projection.go @@ -50,6 +50,7 @@ func (svc *CoreService) ReportRunLifecycle(report domain.RunLifecycleReport) (do if err := svc.store.ServerInstances().Update(instance); err != nil { return domain.RunLifecycleReportResult{}, err } + svc.publishLogProcessState(instance) } auditResult := domain.AuditResultSuccess if report.State == domain.JobStateFailed || report.State == domain.JobStateCancelled { @@ -140,6 +141,7 @@ func (svc *CoreService) projectLifecycleJobResult(job domain.Job, stamp time.Tim if err := svc.store.ServerInstances().Update(instance); err != nil { return err } + svc.publishLogProcessState(instance) auditResult := domain.AuditResultSuccess if job.State == domain.JobStateFailed || job.State == domain.JobStateCancelled { auditResult = domain.AuditResultFailed diff --git a/platform/service/server_lifecycle_test.go b/platform/service/server_lifecycle_test.go index 150f1d4..64aebf3 100644 --- a/platform/service/server_lifecycle_test.go +++ b/platform/service/server_lifecycle_test.go @@ -3,6 +3,7 @@ package service import ( "strings" "testing" + "time" "browser.local/platform/domain" ) @@ -103,6 +104,61 @@ func TestLifecycleProjectedStateUsesRunProcessFacts(t *testing.T) { } } +func TestLifecycleJobResultsPublishProcessStateEvents(t *testing.T) { + svc, sessionToken := newLifecycleRunService(t) + createLifecyclePlugin(t, svc) + created, err := svc.CreateServerInstanceWorkflow(domain.ServerLifecycleCreate{ID: "server-state-events", PluginID: "server.scum", RunEndpointID: "run-local", Name: "State Events", IdempotencyKey: "state-events-create", ProfileKey: "local"}) + if err != nil { + t.Fatalf("create lifecycle server: %v", err) + } + claimAndCompleteLifecycleJobForServer(t, svc, sessionToken, created.Instance.ID, domain.LifecycleCapabilityInstall, domain.JobStateSucceeded) + ready, err := svc.GetServerInstance(created.Instance.ID) + if err != nil { + t.Fatalf("get ready server: %v", err) + } + if _, err := svc.StartServerInstance(domain.ServerLifecycleCommand{ServerInstanceID: ready.ID, ExpectedConfigVersion: ready.ConfigVersion, IdempotencyKey: "state-events-start"}); err != nil { + t.Fatalf("dispatch start: %v", err) + } + claimAndCompleteLifecycleJobForServer(t, svc, sessionToken, ready.ID, domain.LifecycleCapabilityStart, domain.JobStateSucceeded) + + subscription, err := svc.SubscribeLogEvents(ready.ID) + if err != nil { + t.Fatalf("subscribe state events: %v", err) + } + defer subscription.Close() + running, err := svc.GetServerInstance(ready.ID) + if err != nil { + t.Fatalf("get running server: %v", err) + } + if _, err := svc.StopServerInstance(domain.ServerLifecycleCommand{ServerInstanceID: running.ID, ExpectedConfigVersion: running.ConfigVersion, IdempotencyKey: "state-events-stop"}); err != nil { + t.Fatalf("dispatch stop: %v", err) + } + claimAndCompleteLifecycleJobForServer(t, svc, sessionToken, running.ID, domain.LifecycleCapabilityStop, domain.JobStateSucceeded) + assertLogProcessStateEvent(t, subscription, domain.ServerInstanceStateStopped) + + stopped, err := svc.GetServerInstance(running.ID) + if err != nil { + t.Fatalf("get stopped server: %v", err) + } + if _, err := svc.StartServerInstance(domain.ServerLifecycleCommand{ServerInstanceID: stopped.ID, ExpectedConfigVersion: stopped.ConfigVersion, IdempotencyKey: "state-events-restart"}); err != nil { + t.Fatalf("dispatch restart: %v", err) + } + claimAndCompleteLifecycleJobForServer(t, svc, sessionToken, stopped.ID, domain.LifecycleCapabilityStart, domain.JobStateSucceeded) + assertLogProcessStateEvent(t, subscription, domain.ServerInstanceStateRunning) +} + +func assertLogProcessStateEvent(t *testing.T, subscription LogEventSubscription, want domain.ServerInstanceState) { + t.Helper() + select { + case event := <-subscription.Events: + if event.Kind != LogEventSubscriptionEventProcessState || event.ProcessState != want { + t.Fatalf("unexpected process state event: %+v", event) + } + case <-time.After(time.Second): + t.Fatalf("expected process state event %q", want) + } +} + func TestCoreServiceCreatesTargetBoundDraftAndRequiresDedicatedRunRegistration(t *testing.T) { svc, _ := newLifecycleRunService(t) plugin := createLifecyclePlugin(t, svc) diff --git a/platform/validator/log_ingest.go b/platform/validator/log_ingest.go index 042b43f..d4b478b 100644 --- a/platform/validator/log_ingest.go +++ b/platform/validator/log_ingest.go @@ -27,6 +27,14 @@ func ValidateLogBatchIngest(batch domain.LogBatchIngest) error { if !validLogStreamSource(batch.Source) { violations = append(violations, "source is invalid") } + hasSessionID := strings.TrimSpace(batch.LogSessionID) != "" + hasSessionStart := !batch.SessionStartedAt.IsZero() + if hasSessionID != hasSessionStart { + violations = append(violations, "logSessionId and sessionStartedAt must be provided together") + } + if (hasSessionID || hasSessionStart) && batch.Source != domain.LogStreamSourceProcess { + violations = append(violations, "log session metadata is only valid for process streams") + } if batch.FirstSeq == 0 || batch.LastSeq == 0 { violations = append(violations, "sequence range must be positive") } diff --git a/platform/validator/resources.go b/platform/validator/resources.go index bb7f837..e755e3d 100644 --- a/platform/validator/resources.go +++ b/platform/validator/resources.go @@ -1667,6 +1667,14 @@ func ValidateLogStream(stream domain.LogStream) error { if !validLogStorageBackend(stream.StorageBackend) { violations = append(violations, "storageBackend is invalid") } + hasSessionID := strings.TrimSpace(stream.LogSessionID) != "" + hasSessionStart := !stream.SessionStartedAt.IsZero() + if hasSessionID != hasSessionStart { + violations = append(violations, "logSessionId and sessionStartedAt must be provided together") + } + if (hasSessionID || hasSessionStart) && stream.Source != domain.LogStreamSourceProcess { + violations = append(violations, "log session metadata is only valid for process streams") + } return finish(violations) } diff --git a/platform_web/api/client.test.ts b/platform_web/api/client.test.ts index ce40d7a..969b1f2 100644 --- a/platform_web/api/client.test.ts +++ b/platform_web/api/client.test.ts @@ -736,8 +736,18 @@ describe("PlatformApiClient AI providers", () => { expect(client.serverLogEventsUrl("server-1")).toBe("/api/v1/server-instances/server-1/logs/events"); }); + it("scopes explicit terminal history lists to the selected server", async () => { + const fetchMock = vi.fn(async (_input: RequestInfo | URL) => new Response(JSON.stringify({ items: [], count: 0 }), { status: 200, headers: { "Content-Type": "application/json" } })); + vi.stubGlobal("fetch", fetchMock); + const client = new PlatformApiClient("/api/v1", () => "terminal-session"); + + await expect(client.listLogStreams("server/scum 1")).resolves.toEqual({ items: [], count: 0 }); + + expect(String(fetchMock.mock.calls[0]?.[0])).toBe("/api/v1/log-streams?serverInstanceId=server%2Fscum%201"); + }); + it("streams terminal log SSE with bearer authorization and dispatches live events", async () => { - const payload = "event: stream\ndata: {\"id\":\"log-1\",\"serverInstanceId\":\"server-1\"}\n\nevent: log\ndata: {\"streamId\":\"log-1\",\"entry\":{\"seq\":3,\"line\":\"live line\"}}\n\nevent: ready\ndata: {\"serverInstanceId\":\"server-1\"}\n\n"; + const payload = "event: session\ndata: {\"serverInstanceId\":\"server-1\",\"logSessionId\":\"session-2\"}\n\nevent: stream\ndata: {\"id\":\"log-1\",\"serverInstanceId\":\"server-1\"}\n\nevent: log\ndata: {\"streamId\":\"log-1\",\"entry\":{\"seq\":3,\"line\":\"live line\"}}\n\nevent: ready\ndata: {\"serverInstanceId\":\"server-1\"}\n\n"; const fetchMock = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { expect(new Headers(init?.headers).get("Authorization")).toBe("Bearer terminal-session"); expect(new Headers(init?.headers).get("Accept")).toBe("text/event-stream"); @@ -747,11 +757,13 @@ describe("PlatformApiClient AI providers", () => { const client = new PlatformApiClient("/api/v1", () => "terminal-session"); const events: string[] = []; const stream = client.openServerLogEvents("server-1", { historyLimit: 2 }); + stream.addEventListener("session", (event) => events.push(`session:${event.data}`)); stream.addEventListener("stream", (event) => events.push(`stream:${event.data}`)); stream.addEventListener("log", (event) => events.push(`log:${event.data}`)); stream.addEventListener("ready", (event) => events.push(`ready:${event.data}`)); await vi.waitFor(() => expect(events).toEqual([ + 'session:{"serverInstanceId":"server-1","logSessionId":"session-2"}', 'stream:{"id":"log-1","serverInstanceId":"server-1"}', 'log:{"streamId":"log-1","entry":{"seq":3,"line":"live line"}}', 'ready:{"serverInstanceId":"server-1"}' diff --git a/platform_web/api/client.ts b/platform_web/api/client.ts index 26fad10..e697192 100644 --- a/platform_web/api/client.ts +++ b/platform_web/api/client.ts @@ -602,8 +602,9 @@ export class PlatformApiClient { }); } - async listLogStreams(): Promise { - return this.request("/log-streams"); + async listLogStreams(serverInstanceId?: string): Promise { + const query = serverInstanceId ? `?serverInstanceId=${encodeURIComponent(serverInstanceId)}` : ""; + return this.request(`/log-streams${query}`); } openServerLogEvents(id: string, options: LogStreamEventOptions = {}): PlatformEventStream { diff --git a/platform_web/api/types.ts b/platform_web/api/types.ts index 3be7c26..1290a58 100644 --- a/platform_web/api/types.ts +++ b/platform_web/api/types.ts @@ -600,7 +600,6 @@ export interface RuntimeBindingResponse { export interface ServerLifecycleCommandRequest { expectedConfigVersion: number; - expectedChecksum?: string; idempotencyKey: string; } @@ -1478,6 +1477,8 @@ export interface LogStreamResponse { serverInstanceId: string; source: string; streamKey: string; + logSessionId?: string; + sessionStartedAt?: string; latestSeq: number; storageBackend: string; retentionPolicy: string; @@ -1517,6 +1518,8 @@ export interface LogStreamEventResponse { streamId: string; source: string; streamKey: string; + logSessionId?: string; + sessionStartedAt?: string; latestSeq: number; entry: LogEntryBody; } @@ -1527,6 +1530,14 @@ export interface LogStreamEventsReadyResponse { serverTime: string; } +export interface LogStreamEventsSessionResponse { + serverInstanceId: string; + logSessionId?: string; + sessionStartedAt?: string; + streamCount: number; + serverTime: string; +} + export interface LogStreamEventOptions { historyLimit?: number; } diff --git a/platform_web/components/ServerManagementTerminalDrawer.test.tsx b/platform_web/components/ServerManagementTerminalDrawer.test.tsx new file mode 100644 index 0000000..ad36612 --- /dev/null +++ b/platform_web/components/ServerManagementTerminalDrawer.test.tsx @@ -0,0 +1,255 @@ +/** @vitest-environment jsdom */ + +import { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { GameClientBridgeCommandResponse, LogEntryBody, LogStreamResponse } from "../api/types"; +import { ServerManagementTerminalDrawer } from "./ServerManagementTerminalDrawer"; + +const apiMocks = vi.hoisted(() => ({ + getGameClientBridgeCommand: vi.fn(), + listLogStreams: vi.fn(), + openServerLogEvents: vi.fn(), + queryLogStream: vi.fn(), + queueGameClientBridgeCommand: vi.fn() +})); + +vi.mock("../api/client", () => ({ platformApiClient: apiMocks })); + +class FakeEventStream { + onerror: ((event: Event) => void) | null = null; + closed = false; + private readonly listeners = new Map void>>(); + + addEventListener(type: string, listener: (event: MessageEvent) => void) { + const listeners = this.listeners.get(type) ?? new Set<(event: MessageEvent) => void>(); + listeners.add(listener); + this.listeners.set(type, listeners); + } + + removeEventListener(type: string, listener: (event: MessageEvent) => void) { + this.listeners.get(type)?.delete(listener); + } + + close() { + this.closed = true; + this.listeners.clear(); + } + + emit(type: string, payload: unknown) { + const event = new MessageEvent(type, { data: JSON.stringify(payload) }); + this.listeners.get(type)?.forEach((listener) => listener(event)); + } +} + +let root: Root | null = null; +let container: HTMLDivElement | null = null; +let eventStream: FakeEventStream; + +(globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + +beforeEach(() => { + eventStream = new FakeEventStream(); + apiMocks.openServerLogEvents.mockReturnValue(eventStream); + apiMocks.listLogStreams.mockResolvedValue({ items: [], count: 0 }); + apiMocks.queryLogStream.mockResolvedValue({ logStreamId: "", entries: [], nextSeq: 0, latestSeq: 0 }); + vi.stubGlobal("requestAnimationFrame", (callback: FrameRequestCallback) => { callback(0); return 1; }); + vi.stubGlobal("cancelAnimationFrame", vi.fn()); +}); + +afterEach(async () => { + if (root) await act(async () => root?.unmount()); + container?.remove(); + root = null; + container = null; + vi.clearAllMocks(); + vi.unstubAllGlobals(); +}); + +describe("ServerManagementTerminalDrawer", () => { + it("shows current-session replay and keeps it on a repeated boundary for the same session", async () => { + await renderDrawer(); + + await emitSession("session-a"); + await emitStream(logStream("stdout-a", "session-a", "process.stdout")); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(1, "generation A current replay")); + await emitReady(); + + expect(container?.textContent).toContain("generation A current replay"); + expect(container?.textContent).toContain("当前受管进程会话 · SSE 实时推送"); + + await emitSession("session-a"); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(1, "generation A current replay")); + expect(container?.textContent).toContain("generation A current replay"); + expect(Array.from(container?.querySelectorAll(".terminal-text") ?? []).filter((node) => node.textContent === "generation A current replay")).toHaveLength(1); + expect(apiMocks.openServerLogEvents).toHaveBeenCalledTimes(1); + expect(apiMocks.listLogStreams).not.toHaveBeenCalled(); + expect(apiMocks.queueGameClientBridgeCommand).not.toHaveBeenCalled(); + }); + + it("renders an empty current session without accepting unrelated or sessionless logs", async () => { + await renderDrawer(); + + await emitSession(); + await emitLog("legacy-job", undefined, "job", logEntry(1, "legacy output must stay historical")); + await emitReady(); + + expect(container?.textContent).toContain("当前没有可跟随的受管进程输出"); + expect(container?.textContent).not.toContain("legacy output must stay historical"); + expect(apiMocks.queueGameClientBridgeCommand).not.toHaveBeenCalled(); + }); + + it("clears generation A on a new session and rejects late generation A events", async () => { + await renderDrawer(); + await emitSession("session-a"); + await emitStream(logStream("stdout-a", "session-a", "process.stdout")); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(1, "generation A output")); + + await emitSession("session-b"); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(2, "late generation A output")); + await emitStream(logStream("stdout-b", "session-b", "process.stdout")); + await emitLog("stdout-b", "session-b", "process.stdout", logEntry(1, "generation B output")); + + expect(container?.textContent).not.toContain("generation A output"); + expect(container?.textContent).not.toContain("late generation A output"); + expect(container?.textContent).toContain("generation B output"); + expect(container?.textContent).toContain("Run 已切换到新的受管进程输出会话"); + expect(apiMocks.openServerLogEvents).toHaveBeenCalledTimes(1); + }); + + it("clears a stopped session and restores the running session on the same SSE connection", async () => { + await renderDrawer(); + await emitSession("session-a"); + await emitStream(logStream("stdout-a", "session-a", "process.stdout")); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(1, "running output before stop")); + + await emitSession(); + await emitLog("stdout-a", "session-a", "process.stdout", logEntry(2, "late output after stop")); + expect(container?.textContent).not.toContain("running output before stop"); + expect(container?.textContent).not.toContain("late output after stop"); + expect(container?.textContent).toContain("当前没有可跟随的受管进程输出"); + + await emitSession("session-b"); + await emitStream(logStream("stdout-b", "session-b", "process.stdout")); + await emitLog("stdout-b", "session-b", "process.stdout", logEntry(1, "running output after recovery")); + expect(container?.textContent).not.toContain("running output before stop"); + expect(container?.textContent).toContain("running output after recovery"); + 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.mockResolvedValue({ logStreamId: oldStream.id, entries: [logEntry(1, "selected historical output", "2026-08-01T00:00:01Z")], nextSeq: 1, latestSeq: 1 }); + 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(apiMocks.queueGameClientBridgeCommand).not.toHaveBeenCalled(); + }); + + it("uses RCON only after an operator submits a command", async () => { + const pending = bridgeCommand("pending"); + const succeeded = bridgeCommand("succeeded"); + apiMocks.queueGameClientBridgeCommand.mockResolvedValue(pending); + apiMocks.getGameClientBridgeCommand.mockResolvedValue(succeeded); + await renderDrawer(); + await emitSession("session-current"); + await emitLog("stdout-current", "session-current", "process.stdout", logEntry(1, "ordinary live output")); + expect(apiMocks.queueGameClientBridgeCommand).not.toHaveBeenCalled(); + + const input = container?.querySelector('.terminal-command-form input'); + const form = container?.querySelector('.terminal-command-form'); + if (!input || !form) throw new Error("command form not found"); + await act(async () => setInputValue(input, "#ListPlayers")); + await act(async () => form.dispatchEvent(new SubmitEvent("submit", { bubbles: true, cancelable: true }))); + await flushPromises(); + + expect(apiMocks.queueGameClientBridgeCommand).toHaveBeenCalledTimes(1); + expect(apiMocks.getGameClientBridgeCommand).toHaveBeenCalledWith("server-1", pending.id); + expect(container?.textContent).toContain("ordinary live output"); + }); +}); + +async function renderDrawer() { + container = document.createElement("div"); + document.body.append(container); + root = createRoot(container); + await act(async () => { + root?.render( undefined} />); + }); + expect(apiMocks.openServerLogEvents).toHaveBeenCalledWith("server-1", { historyLimit: 500 }); +} + +async function emitSession(logSessionId?: string) { + await act(async () => eventStream.emit("session", { serverInstanceId: "server-1", logSessionId, streamCount: logSessionId ? 1 : 0, serverTime: "2026-08-14T00:00:00Z" })); +} + +async function emitStream(stream: LogStreamResponse) { + await act(async () => eventStream.emit("stream", stream)); +} + +async function emitLog(streamId: string, logSessionId: string | undefined, streamKey: string, entry: LogEntryBody) { + await act(async () => eventStream.emit("log", { serverInstanceId: "server-1", streamId, source: "process", streamKey, logSessionId, latestSeq: entry.seq, entry })); +} + +async function emitReady() { + await act(async () => eventStream.emit("ready", { serverInstanceId: "server-1", streamCount: 1, serverTime: "2026-08-14T00:00:00Z" })); +} + +async function clickButton(label: string) { + const button = Array.from(container?.querySelectorAll("button") ?? []).find((item) => item.textContent?.includes(label)); + if (!button) throw new Error(`button not found: ${label}`); + await act(async () => button.dispatchEvent(new MouseEvent("click", { bubbles: true, cancelable: true }))); +} + +async function flushPromises() { + await act(async () => { await Promise.resolve(); }); +} + +function setInputValue(input: HTMLInputElement, value: string) { + Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, "value")?.set?.call(input, value); + 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 }; +} + +function logEntry(seq: number, line: string, timestamp = `2026-08-14T00:00:0${seq}Z`): LogEntryBody { + return { seq, timestamp, line, redacted: true }; +} + +function bridgeCommand(state: GameClientBridgeCommandResponse["state"]): GameClientBridgeCommandResponse { + return { + id: "command-1", serverInstanceId: "server-1", pluginId: "game.scum", profileKey: "scum-rcon", commandType: "management.command", priority: 50, state, approvalState: "not_required", + result: state === "succeeded" ? { status: "succeeded", summary: "command completed", completedAt: "2026-08-14T00:00:03Z" } : undefined, + expiresAt: "2026-08-14T00:01:00Z", createdAt: "2026-08-14T00:00:00Z", updatedAt: "2026-08-14T00:00:03Z", completedAt: state === "succeeded" ? "2026-08-14T00:00:03Z" : undefined + }; +} diff --git a/platform_web/components/ServerManagementTerminalDrawer.tsx b/platform_web/components/ServerManagementTerminalDrawer.tsx index 774a20d..c68e64e 100644 --- a/platform_web/components/ServerManagementTerminalDrawer.tsx +++ b/platform_web/components/ServerManagementTerminalDrawer.tsx @@ -1,14 +1,15 @@ -import { ListChecks, Send, Sparkles, Terminal, Trash2, X } from "lucide-react"; +import { History, 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"; import type { GameClientBridgeCommandResponse, LogEntryBody, LogStreamResponse } from "../api/types"; import { scumManagementRCONCommandRequest } from "../schemas/scumManagementRcon"; import { cx } from "../utils/classes"; -import { mergeLogStreams, parseLogStreamEvent, parseServerLogEvent, streamFromServerLogEvent } from "../utils/logEvents"; +import { mergeLogStreams, parseLogSessionEvent, parseLogStreamEvent, parseServerLogEvent, streamFromServerLogEvent } from "../utils/logEvents"; 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 }; @@ -89,9 +90,16 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu const [historyIndex, setHistoryIndex] = useState(null); 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 initialHistoryPendingRef = useRef(false); + const liveSessionRef = useRef(undefined); + const historyRequestRef = useRef(0); const quickCommands = useMemo(() => terminalQuickCommandsForPlugin(pluginId), [pluginId]); const supportsCommands = quickCommands.length > 0; @@ -121,10 +129,17 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu setPending(false); setResult(null); setHistoryIndex(null); + setLiveSessionId(null); + setHistoryOpen(false); + setHistoryStreams({ status: "loading" }); + setHistoryLines({ status: "idle" }); + setSelectedHistoryStreamId(""); + liveSessionRef.current = undefined; + historyRequestRef.current += 1; initialHistoryPendingRef.current = true; followLatestRef.current = true; setFollowLatest(true); - setLines([terminalSystemLine("info", supportsCommands ? "连接平台日志推送，先补最近历史再实时追加。" : "该插件暂未声明可用的终端命令通道。", "SYSTEM")]); + setLines([terminalSystemLine("info", supportsCommands ? "正在连接当前受管进程输出。" : "该插件暂未声明可用的终端命令通道。", "SYSTEM")]); }, [open, supportsCommands]); useEffect(() => { @@ -140,9 +155,24 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu if (!open) return undefined; let ready = false; const events = platformApiClient.openServerLogEvents(serverId, { historyLimit: terminalInitialHistoryWindow }); + events.addEventListener("session", (event) => { + const session = parseLogSessionEvent(event); + if (!session) return; + ready = true; + const nextSessionId = normalizeLogSessionId(session.logSessionId); + const previousSessionId = liveSessionRef.current; + liveSessionRef.current = nextSessionId; + setLiveSessionId(nextSessionId); + if (previousSessionId === nextSessionId) return; + setStreams({ status: "ready", data: [] }); + setLines(nextSessionId + ? [terminalSystemLine("info", previousSessionId === undefined ? "已跟随当前受管进程输出会话。" : "Run 已切换到新的受管进程输出会话。", "SYSTEM", `session-${nextSessionId}`)] + : [terminalSystemLine("warn", "当前没有可跟随的受管进程输出；旧日志可从历史查看。", "SYSTEM", "session-empty")]); + lockTerminalFollow(); + }); events.addEventListener("stream", (event) => { const stream = parseLogStreamEvent(event); - if (!stream) return; + if (!stream || !eventBelongsToLiveSession(stream.logSessionId, liveSessionRef.current)) return; ready = true; setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); }); @@ -153,7 +183,7 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu }); events.addEventListener("log", (event) => { const payload = parseServerLogEvent(event); - if (!payload) return; + if (!payload || !eventBelongsToLiveSession(payload.logSessionId, liveSessionRef.current)) return; ready = true; const stream = streamFromServerLogEvent(payload); setStreams((current) => ({ status: "ready", data: mergeLogStreams(current.status === "ready" ? current.data : [], stream) })); @@ -165,6 +195,35 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu return () => events.close(); }, [appendLines, 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 - terminalInitialHistoryWindow), limit: terminalInitialHistoryWindow }); + 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 : "历史日志加载失败" }); + } + } + function selectQuickCommand(item: TerminalQuickCommand) { setCommand(item.command); setHistoryIndex(null); @@ -193,9 +252,20 @@ 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; @@ -253,17 +323,19 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu
{serverName} - 当前服务器 Run 日志 · 已接受历史 + SSE 实时推送 · {streams.status === "ready" ? "等待当前输出" : streams.status === "loading" ? "连接日志流" : "日志流异常"} · {followLatest ? "自动置底" : "已解锁滚动"} + {historyOpen ? "历史日志（独立于实时终端）" : `当前受管进程会话${liveSessionId ? " · SSE 实时推送" : " · 等待 Run 输出"}`} · {streams.status === "ready" ? "已连接" : streams.status === "loading" ? "连接日志流" : "日志流异常"} · {followLatest ? "自动置底" : "已解锁滚动"}
+
- {streams.status === "error" &&
{streams.reason}
} - {streams.status === "ready" && streams.data.length === 0 &&
暂无已接受日志。Run 恢复连接并完成日志水位校准后，新输出会继续追加。
} - {lines.map((line) =>
{line.text}
)} + {!historyOpen && streams.status === "error" &&
{streams.reason}
} + {historyOpen && } + {!historyOpen && streams.status === "ready" && lines.length === 0 &&
{liveSessionId ? "当前受管进程会话暂无输出，后续输出会自动追加。" : "当前没有可跟随的受管进程输出；旧日志可从历史查看。"}
} + {!historyOpen && lines.map((line) =>
{line.text}
)}
@@ -279,7 +351,7 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu
void submitCommand(event)}> @@ -290,6 +362,41 @@ export function ServerManagementTerminalDrawer({ open, serverId, serverName, plu ); } +interface HistoryLogViewProps { + streams: LoadState; + lines: HistoryLineState; + selectedStreamId: string; + onSelect: (streamId: string) => Promise; +} + +function HistoryLogView({ streams, lines, selectedStreamId, onSelect }: 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 }: { tone: "info" | "warn" | "error"; label: string }) { + return
{label}
; +} + function terminalQuickCommandsForPlugin(pluginId: string): TerminalQuickCommand[] { return terminalQuickCommandCatalog[pluginId] ?? []; } @@ -331,6 +438,16 @@ function terminalSourceClass(value?: string): string { return "log"; } +function normalizeLogSessionId(value?: string): string | null { + const normalized = value?.trim(); + return normalized || null; +} + +function eventBelongsToLiveSession(eventSessionId: string | undefined, liveSessionId: string | null | undefined): boolean { + const normalizedEventSessionId = normalizeLogSessionId(eventSessionId); + return Boolean(normalizedEventSessionId && normalizedEventSessionId === liveSessionId); +} + function isTerminalBridgeCommandState(state: GameClientBridgeCommandResponse["state"]): boolean { return state === "succeeded" || state === "failed" || state === "cancelled" || state === "expired" || state === "unknown"; } diff --git a/platform_web/schemas/serverManagement.test.ts b/platform_web/schemas/serverManagement.test.ts index ebda958..3a9995b 100644 --- a/platform_web/schemas/serverManagement.test.ts +++ b/platform_web/schemas/serverManagement.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from "vitest"; -import type { GamePluginResponse } from "../api/types"; +import type { GamePluginResponse, ServerInstanceResponse } from "../api/types"; import { defaultServerCreateForm, runtimeBindingFields } from "../contracts/serverManagement"; -import { minimalServerCreateRequestFromForm, serverCreateRequestFromForm, serverInstanceIdFromName } from "./serverManagement"; +import { minimalServerCreateRequestFromForm, serverCreateRequestFromForm, serverInstanceIdFromName, serverLifecycleCommandRequest } from "./serverManagement"; const plugin: GamePluginResponse = { id: "game.runtime", @@ -109,6 +109,15 @@ describe("runtime profile server creation contracts", () => { expect(serverInstanceIdFromName("测试服", 18)).toBe("server-18"); }); + it("matches the platform lifecycle command DTO exactly", () => { + const instance = { id: "server-1", configVersion: 7, configChecksum: "sha256:unused" } as ServerInstanceResponse; + + expect(serverLifecycleCommandRequest(instance, "start", 20)).toEqual({ + expectedConfigVersion: 7, + idempotencyKey: "web:start:server-1:20" + }); + }); + it("keeps complete paths and commands in a write-only deployment payload", () => { const form = defaultServerCreateForm([plugin], []); const request = serverCreateRequestFromForm({ ...form, name: "Venv Server", deploymentMode: "custom-command", serverRoot: "/srv/venv-server", workingDirectory: "/srv/venv-server", startCommand: "/srv/venv-server/.venv/bin/python server.py", shell: "" }, 19); diff --git a/platform_web/schemas/serverManagement.ts b/platform_web/schemas/serverManagement.ts index de4a8f4..608aaef 100644 --- a/platform_web/schemas/serverManagement.ts +++ b/platform_web/schemas/serverManagement.ts @@ -59,7 +59,6 @@ export function serverInstanceIdFromName(name: string, sequence = Date.now()): s export function serverLifecycleCommandRequest(instance: ServerInstanceResponse, action: "deploy" | "start" | "stop" | "status", sequence = Date.now()): ServerLifecycleCommandRequest { return { expectedConfigVersion: instance.configVersion, - expectedChecksum: instance.configChecksum, idempotencyKey: lifecycleIdempotencyKey(action, instance.id, sequence) }; } diff --git a/platform_web/utils/logEvents.ts b/platform_web/utils/logEvents.ts index abfde81..eab123f 100644 --- a/platform_web/utils/logEvents.ts +++ b/platform_web/utils/logEvents.ts @@ -1,4 +1,4 @@ -import type { LogEntryBody, LogStreamEventResponse, LogStreamResponse } from "../api/types"; +import type { LogEntryBody, LogStreamEventResponse, LogStreamEventsSessionResponse, LogStreamResponse } from "../api/types"; export type LiveLogEntry = LogEntryBody & { source: string; streamId: string; streamKey: string }; @@ -14,6 +14,12 @@ export function parseServerLogEvent(event: MessageEvent): LogStreamEventResponse return value as unknown as LogStreamEventResponse; } +export function parseLogSessionEvent(event: MessageEvent): LogStreamEventsSessionResponse | null { + const value = parseEventData(event); + if (!isRecord(value) || typeof value.serverInstanceId !== "string") return null; + return value as unknown as LogStreamEventsSessionResponse; +} + export function entryFromServerLogEvent(event: LogStreamEventResponse): LiveLogEntry { return { ...event.entry, source: event.source || event.streamKey, streamId: event.streamId, streamKey: event.streamKey }; } @@ -24,6 +30,8 @@ export function streamFromServerLogEvent(event: LogStreamEventResponse): LogStre serverInstanceId: event.serverInstanceId, source: event.source, streamKey: event.streamKey, + logSessionId: event.logSessionId, + sessionStartedAt: event.sessionStartedAt, latestSeq: event.latestSeq, storageBackend: "", retentionPolicy: "", diff --git a/scripts/README.md b/scripts/README.md index 8d3c1cb..f5d076a 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -4,6 +4,7 @@ Use these root-level commands for day-to-day work: - `scripts/dev-start.sh`: start the local platform API, run worker, and platform_web console. - `scripts/dev-smoke.sh`: seed and verify the API-backed local plugin/server fixtures. +- `LOCAL_DEBUG_SELF_START=true LOCAL_DEBUG_LOG_SESSION_SMOKE_ONLY=true scripts/dev-smoke.sh`: run the focused current supervised-log session restart/SSE proof without distribution builds. - `scripts/dev-stop.sh`: stop the managed local debug stack. - `scripts/dev-reset.sh`: stop the stack and delete only the safe local debug data root. - `scripts/browser-acceptance.sh`: run the full self-starting browser acceptance suite. diff --git a/scripts/local-debug/smoke.sh b/scripts/local-debug/smoke.sh index 6138e25..013c016 100755 --- a/scripts/local-debug/smoke.sh +++ b/scripts/local-debug/smoke.sh @@ -10,6 +10,7 @@ API_URL="$PLATFORM_URL/api/v1" WORK_DIR="$LOCAL_DEBUG_ROOT/smoke" SMOKE_INVOCATION_ID="${SMOKE_INVOCATION_ID:-$(date -u +%Y%m%d%H%M%S)-$$}" SERVER_LOCAL_ID="server-local-debug-$SMOKE_INVOCATION_ID" +LOG_SESSION_SERVER_ID="server-log-session-$SMOKE_INVOCATION_ID" SCUM_ALPHA_ID="scum-alpha-$SMOKE_INVOCATION_ID" SCUM_BETA_ID="scum-beta-$SMOKE_INVOCATION_ID" SCUM_DYNAMIC_ID="scum-dynamic-$SMOKE_INVOCATION_ID" @@ -17,11 +18,21 @@ GENERATED_RUN_ENDPOINT_ID="server-run-$SERVER_LOCAL_ID" GENERATED_RUN_BIN="$WORK_DIR/generated-$SERVER_LOCAL_ID-run" GENERATED_RUN_LOG="$LOCAL_DEBUG_LOG_DIR/generated-$SERVER_LOCAL_ID-run.log" GENERATED_RUN_PID_FILE="$LOCAL_DEBUG_PID_DIR/generated-$SERVER_LOCAL_ID-run.pid" +LOG_SESSION_SCOPE="$RUN_WORKSPACE_ROOT/instances/$LOG_SESSION_SERVER_ID/run-local" +LOG_SESSION_MARKER_PREFIX="SMOKE-CURRENT-SESSION-$SMOKE_INVOCATION_ID" +LOG_SESSION_SSE_FILE="$WORK_DIR/current-log-session.events.sse" +LOG_SESSION_SSE_ERROR_FILE="$WORK_DIR/current-log-session.events.stderr.log" +LOG_SESSION_SSE_PID="" +LOG_SESSION_LEGACY_STREAM_ID="legacy.$LOG_SESSION_SERVER_ID.stdout" +LOG_SESSION_JOB_STREAM_ID="job.smoke-$SMOKE_INVOCATION_ID.stdout" +LOG_SESSION_FILE_STREAM_ID="file.$LOG_SESSION_SERVER_ID.backfill" +LOG_SESSION_REAL_JOB_STREAM_IDS="" mkdir -p "$WORK_DIR" cat >"$WORK_DIR/run-build-config.env" <"$WORK_DIR/current-log-session-fixture.sh" <&2 +printf '%s\n' "\$\$" >smoke-process.pid +last_command="" +while :; do + command="\$(cat smoke-command 2>/dev/null || true)" + if [ -n "\$command" ] && [ "\$command" != "\$last_command" ]; then + case "\$command" in + resume) + printf '%s-%s-RUN-RESUME-STDOUT\n' '$LOG_SESSION_MARKER_PREFIX' "\$generation" + printf '%s-%s-RUN-RESUME-STDERR\n' '$LOG_SESSION_MARKER_PREFIX' "\$generation" >&2 + ;; + esac + last_command="\$command" + printf '%s\n' "\$command" >smoke-last-command + fi + sleep 0.2 +done +EOF +cat >"$WORK_DIR/current-log-session-install-fixture.sh" <&2 +EOF +cp "$WORK_DIR/current-log-session-fixture.sh" "$LOG_SESSION_SCOPE/bin/game-server" +cp "$WORK_DIR/current-log-session-install-fixture.sh" "$LOG_SESSION_SCOPE/bin/install-server" +chmod 700 "$LOG_SESSION_SCOPE/bin/game-server" +chmod 700 "$LOG_SESSION_SCOPE/bin/install-server" +printf 'A\n' >"$LOG_SESSION_SCOPE/smoke-generation" +rm -f "$LOG_SESSION_SCOPE/smoke-command" "$LOG_SESSION_SCOPE/smoke-last-command" "$LOG_SESSION_SCOPE/smoke-process.pid" + SELF_STARTED_PIDS=() cleanup_self_started() { @@ -63,6 +112,83 @@ cleanup_self_started() { done } +cleanup_log_session_fixture() { + if [[ -n "${LOG_SESSION_SSE_PID:-}" ]] && kill -0 "$LOG_SESSION_SSE_PID" 2>/dev/null; then + kill "$LOG_SESSION_SSE_PID" 2>/dev/null || true + wait "$LOG_SESSION_SSE_PID" 2>/dev/null || true + fi + local fixture_pid_file="$LOG_SESSION_SCOPE/smoke-process.pid" + if [[ -f "$fixture_pid_file" ]]; then + local fixture_pid + fixture_pid="$(<"$fixture_pid_file")" + if [[ "$fixture_pid" =~ ^[0-9]+$ ]] && kill -0 "$fixture_pid" 2>/dev/null; then + local fixture_command + fixture_command="$(ps -p "$fixture_pid" -o command= 2>/dev/null || true)" + if [[ "$fixture_command" == *"$LOG_SESSION_SCOPE/bin/game-server"* ]]; then + kill "$fixture_pid" 2>/dev/null || true + fi + fi + fi +} + +cleanup_smoke() { + cleanup_log_session_fixture + cleanup_self_started +} + +trap cleanup_smoke EXIT + +launch_bootstrap_run() { + local log_mode="${1:-overwrite}" + mkdir -p "$LOCAL_DEBUG_LOG_DIR" "$LOCAL_DEBUG_PID_DIR" + if [[ "$log_mode" == "append" ]]; then + ( + cd "$(dirname "$RUN_BOOTSTRAP_BIN")" + exec env \ + GOCACHE="$GOCACHE" \ + RUN_MODE="$RUN_MODE" \ + RUN_PLATFORM_URL="$RUN_PLATFORM_URL" \ + RUN_ENDPOINT_ID="$RUN_ENDPOINT_ID" \ + RUN_DISPLAY_NAME="$RUN_DISPLAY_NAME" \ + RUN_VERSION="$RUN_VERSION" \ + RUN_REGISTRATION_TOKEN="$RUN_REGISTRATION_TOKEN" \ + RUN_WORKSPACE_ROOT="$RUN_WORKSPACE_ROOT" \ + RUN_BUILD_SOURCE_ROOT="$RUN_BUILD_SOURCE_ROOT" \ + RUN_SPOOL_ROOT="$RUN_SPOOL_ROOT" \ + RUN_MAX_JOBS="$RUN_MAX_JOBS" \ + RUN_HEARTBEAT_INTERVAL_MS="$RUN_HEARTBEAT_INTERVAL_MS" \ + RUN_POLL_INTERVAL_MS="$RUN_POLL_INTERVAL_MS" \ + RUN_RETRY_BACKOFF_MS="$RUN_RETRY_BACKOFF_MS" \ + "$RUN_BOOTSTRAP_BIN" + ) >>"$LOCAL_DEBUG_LOG_DIR/run.log" 2>&1 & + else + ( + cd "$(dirname "$RUN_BOOTSTRAP_BIN")" + exec env \ + GOCACHE="$GOCACHE" \ + RUN_MODE="$RUN_MODE" \ + RUN_PLATFORM_URL="$RUN_PLATFORM_URL" \ + RUN_ENDPOINT_ID="$RUN_ENDPOINT_ID" \ + RUN_DISPLAY_NAME="$RUN_DISPLAY_NAME" \ + RUN_VERSION="$RUN_VERSION" \ + RUN_REGISTRATION_TOKEN="$RUN_REGISTRATION_TOKEN" \ + RUN_WORKSPACE_ROOT="$RUN_WORKSPACE_ROOT" \ + RUN_BUILD_SOURCE_ROOT="$RUN_BUILD_SOURCE_ROOT" \ + RUN_SPOOL_ROOT="$RUN_SPOOL_ROOT" \ + RUN_MAX_JOBS="$RUN_MAX_JOBS" \ + RUN_HEARTBEAT_INTERVAL_MS="$RUN_HEARTBEAT_INTERVAL_MS" \ + RUN_POLL_INTERVAL_MS="$RUN_POLL_INTERVAL_MS" \ + RUN_RETRY_BACKOFF_MS="$RUN_RETRY_BACKOFF_MS" \ + "$RUN_BOOTSTRAP_BIN" + ) >"$LOCAL_DEBUG_LOG_DIR/run.log" 2>&1 & + fi + local run_pid="$!" + printf '%s' "$run_pid" >"$LOCAL_DEBUG_PID_DIR/run.pid" + if [[ "${LOCAL_DEBUG_SELF_START:-false}" == "true" ]]; then + SELF_STARTED_PIDS+=("$run_pid") + fi +} + wait_for_url() { local name="$1" local url="$2" @@ -127,8 +253,9 @@ assert_no_managed_run_process() { start_self_hosted_stack() { mkdir -p "$LOCAL_DEBUG_LOG_DIR" "$LOCAL_DEBUG_PID_DIR" "$PLATFORM_DATA_DIR" "$PLATFORM_LOG_DIR" "$PLATFORM_ARTIFACT_DIR" "$RUN_WORKSPACE_ROOT" "$RUN_SPOOL_ROOT" "$RUN_BUILD_BUCKET_ROOT" "$GOCACHE" - local_debug_prepare_distribution_builder - trap cleanup_self_started EXIT + if [[ "${LOCAL_DEBUG_LOG_SESSION_SMOKE_ONLY:-false}" != "true" ]]; then + local_debug_prepare_distribution_builder + fi printf 'self-starting platform for local debug smoke\n' ( @@ -172,27 +299,7 @@ start_self_hosted_stack() { printf 'self-starting run worker for local debug smoke\n' local_debug_build_bootstrap_run - ( - cd "$(dirname "$RUN_BOOTSTRAP_BIN")" - exec env \ - GOCACHE="$GOCACHE" \ - RUN_MODE="$RUN_MODE" \ - RUN_PLATFORM_URL="$RUN_PLATFORM_URL" \ - RUN_ENDPOINT_ID="$RUN_ENDPOINT_ID" \ - RUN_DISPLAY_NAME="$RUN_DISPLAY_NAME" \ - RUN_VERSION="$RUN_VERSION" \ - RUN_REGISTRATION_TOKEN="$RUN_REGISTRATION_TOKEN" \ - RUN_WORKSPACE_ROOT="$RUN_WORKSPACE_ROOT" \ - RUN_BUILD_SOURCE_ROOT="$RUN_BUILD_SOURCE_ROOT" \ - RUN_SPOOL_ROOT="$RUN_SPOOL_ROOT" \ - RUN_MAX_JOBS="$RUN_MAX_JOBS" \ - RUN_HEARTBEAT_INTERVAL_MS="$RUN_HEARTBEAT_INTERVAL_MS" \ - RUN_POLL_INTERVAL_MS="$RUN_POLL_INTERVAL_MS" \ - RUN_RETRY_BACKOFF_MS="$RUN_RETRY_BACKOFF_MS" \ - "$RUN_BOOTSTRAP_BIN" - ) >"$LOCAL_DEBUG_LOG_DIR/run.log" 2>&1 & - SELF_STARTED_PIDS+=("$!") - printf '%s' "$!" >"$LOCAL_DEBUG_PID_DIR/run.pid" + launch_bootstrap_run overwrite } if [[ "${LOCAL_DEBUG_SELF_START:-false}" == "true" ]]; then @@ -329,7 +436,7 @@ if (!logSources.some((source) => source.kind === "process.stderr" && source.stre if (!plugin.gameClientBridge?.commands?.length) { missing.push("game client bridge declarations"); } -for (const pageKey of ["files-config", "players", "squads", "live-map", "gifts"]) { +for (const pageKey of ["players", "squads", "live-map", "gifts"]) { if (!pageKeys.includes(pageKey)) { missing.push(`SCUM page ${pageKey}`); } @@ -636,6 +743,201 @@ NODE exit 1 } +wait_for_job_success() { + local response_file="$1" + local output_file="$2" + local label="$3" + local job_id + job_id="$(node -e 'const fs=require("fs"); const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); const id=data.job?.id || data.id; if (!id) process.exit(2); process.stdout.write(id);' "$response_file")" + rm -f "$output_file" + for _ in $(seq 1 60); do + json_get "$API_URL/jobs/$job_id" "$output_file" "${AUTH_HEADER[@]}" || true + if [[ -s "$output_file" ]]; then + local state + state="$(node -e 'const fs=require("fs"); const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); process.stdout.write(data.state || "unknown");' "$output_file")" + case "$state" in + succeeded) + return 0 + ;; + failed | cancelled) + printf '%s job %s reached %s\n' "$label" "$job_id" "$state" >&2 + sed -n '1,200p' "$output_file" >&2 + return 1 + ;; + esac + fi + sleep 1 + done + printf '%s job %s did not succeed before timeout\n' "$label" "$job_id" >&2 + [[ -s "$output_file" ]] && sed -n '1,200p' "$output_file" >&2 + return 1 +} + +dump_log_session_diagnostics() { + local reason="$1" + printf 'current log session smoke diagnostics: %s\n' "$reason" >&2 + for file in \ + "$WORK_DIR/log-session-streams.response.json" \ + "$LOG_SESSION_SSE_FILE" \ + "$LOG_SESSION_SSE_ERROR_FILE" \ + "$LOCAL_DEBUG_LOG_DIR/run.log"; do + if [[ -s "$file" ]]; then + printf '%s (last 160 lines):\n' "$file" >&2 + tail -n 160 "$file" >&2 + fi + done +} + +wait_for_file_literal() { + local label="$1" + local file="$2" + local literal="$3" + local attempts="${4:-60}" + for _ in $(seq 1 "$attempts"); do + if [[ -f "$file" ]] && grep -Fq -- "$literal" "$file"; then + return 0 + fi + if [[ -n "${LOG_SESSION_SSE_PID:-}" ]] && ! kill -0 "$LOG_SESSION_SSE_PID" 2>/dev/null; then + dump_log_session_diagnostics "$label: SSE client exited" + return 1 + fi + sleep 1 + done + dump_log_session_diagnostics "$label: timed out waiting for $literal" + return 1 +} + +wait_for_persisted_log_marker() { + local marker="$1" + local evidence_file="$2" + local stream_scope="${3:-session}" + local streams_file="$WORK_DIR/log-session-streams.response.json" + for attempt in $(seq 1 60); do + json_get "$API_URL/log-streams?serverInstanceId=$LOG_SESSION_SERVER_ID" "$streams_file" "${AUTH_HEADER[@]}" || true + if [[ -s "$streams_file" ]]; then + local stream_ids + stream_ids="$(node -e 'const fs=require("fs"); const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); const scope=process.argv[2]; for (const item of data.items || []) if (scope === "history" || item.logSessionId) console.log(item.id);' "$streams_file" "$stream_scope" 2>/dev/null || true)" + local stream_id + for stream_id in $stream_ids; do + local request_file="$WORK_DIR/log-session-query-$attempt.request.json" + printf '{"logStreamId":"%s","afterSeq":0,"limit":200}\n' "$stream_id" >"$request_file" + if json_post "$API_URL/log-streams/query" "$request_file" "$evidence_file" "${AUTH_HEADER[@]}" && node -e 'const fs=require("fs"); const marker=process.argv[2]; const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); process.exit((data.entries || []).some((entry) => entry.line === marker) ? 0 : 1);' "$evidence_file" "$marker"; then + return 0 + fi + done + fi + sleep 1 + done + dump_log_session_diagnostics "persisted marker timeout: $marker" + return 1 +} + +current_log_session_id() { + json_get "$API_URL/log-streams?serverInstanceId=$LOG_SESSION_SERVER_ID" "$WORK_DIR/log-session-streams.response.json" "${AUTH_HEADER[@]}" + node - "$WORK_DIR/log-session-streams.response.json" <<'NODE' +const fs = require("fs"); +const response = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); +const current = (response.items || []) + .filter((item) => item.source === "process" && item.logSessionId && item.sessionStartedAt) + .sort((left, right) => Date.parse(right.sessionStartedAt) - Date.parse(left.sessionStartedAt) || right.logSessionId.localeCompare(left.logSessionId))[0]; +if (!current) process.exit(2); +process.stdout.write(current.logSessionId); +NODE +} + +wait_for_run_heartbeat_after() { + local previous_heartbeat="$1" + local endpoint_file="$WORK_DIR/log-session-run-endpoint.response.json" + for _ in $(seq 1 45); do + if json_get "$API_URL/run/endpoints/$RUN_ENDPOINT_ID" "$endpoint_file" "${AUTH_HEADER[@]}" 2>/dev/null && node - "$endpoint_file" "$RUN_ENDPOINT_ID" "$previous_heartbeat" <<'NODE' +const fs = require("fs"); +const endpoint = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); +const current = Date.parse(endpoint.lastHeartbeatAt || endpoint.LastHeartbeatAt || ""); +const previous = Date.parse(process.argv[4] || ""); +process.exit(endpoint.id === process.argv[3] && endpoint.status === "online" && Number.isFinite(current) && current > previous ? 0 : 1); +NODE + then + return 0 + fi + sleep 1 + done + dump_log_session_diagnostics "Run did not heartbeat after restart" + return 1 +} + +restart_bootstrap_run_preserving_process() { + local endpoint_file="$WORK_DIR/log-session-run-endpoint-before-restart.response.json" + json_get "$API_URL/run/endpoints/$RUN_ENDPOINT_ID" "$endpoint_file" "${AUTH_HEADER[@]}" + local previous_heartbeat + previous_heartbeat="$(node -e 'const fs=require("fs"); const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); process.stdout.write(data.lastHeartbeatAt || data.LastHeartbeatAt || "");' "$endpoint_file")" + local run_pid + run_pid="$(<"$LOCAL_DEBUG_PID_DIR/run.pid")" + if [[ ! "$run_pid" =~ ^[0-9]+$ ]] || ! kill -0 "$run_pid" 2>/dev/null; then + dump_log_session_diagnostics "managed Run pid is unavailable before restart" + return 1 + fi + printf 'stopping Run pid %s while leaving its supervised process alive\n' "$run_pid" + kill "$run_pid" + for _ in $(seq 1 40); do + if ! kill -0 "$run_pid" 2>/dev/null; then + break + fi + sleep 0.25 + done + if kill -0 "$run_pid" 2>/dev/null; then + dump_log_session_diagnostics "Run pid $run_pid did not stop" + return 1 + fi + printf 'resume\n' >"$LOG_SESSION_SCOPE/smoke-command.next" + mv "$LOG_SESSION_SCOPE/smoke-command.next" "$LOG_SESSION_SCOPE/smoke-command" + wait_for_file_literal "supervised process output while Run is down" "$LOG_SESSION_SCOPE/smoke-last-command" "resume" 40 + launch_bootstrap_run append + wait_for_run_heartbeat_after "$previous_heartbeat" +} + +write_server_lifecycle_request() { + local action="$1" + local output_file="$2" + json_get "$API_URL/server-instances/$LOG_SESSION_SERVER_ID" "$WORK_DIR/log-session-server-current.response.json" "${AUTH_HEADER[@]}" + local config_version + config_version="$(node -e 'const fs=require("fs"); const data=JSON.parse(fs.readFileSync(process.argv[1],"utf8")); if (!Number.isInteger(data.configVersion)) process.exit(2); process.stdout.write(String(data.configVersion));' "$WORK_DIR/log-session-server-current.response.json")" + cat >"$output_file" <"$request_file" <"$WORK_DIR/server-runtime-binding.request.json" <"$WORK_DIR/create-log-session-server.request.json" <"$WORK_DIR/log-session-runtime-binding.request.json" <"$WORK_DIR/log-session-deployment.request.json" <"$WORK_DIR/server-run-generate.request.json" <&2 exit 1 fi @@ -929,9 +1266,10 @@ curl -fsS -X PUT -H 'Content-Type: application/json' "${AUTH_HEADER[@]}" \ reject_forbidden_fragments "$WORK_DIR/server-runtime-binding.response.json" require_file_contains "$WORK_DIR/server-runtime-binding.response.json" '"status"[[:space:]]*:[[:space:]]*"complete"' -printf 'checking example server platform-builder action\n' -json_get "$API_URL/server-instances/$SERVER_ID/runtime/actions" "$WORK_DIR/server-runtime-actions.response.json" "${AUTH_HEADER[@]}" -node - "$WORK_DIR/server-runtime-actions.response.json" <<'NODE' +if [[ "${LOCAL_DEBUG_LOG_SESSION_SMOKE_ONLY:-false}" != "true" ]]; then + printf 'checking example server platform-builder action\n' + json_get "$API_URL/server-instances/$SERVER_ID/runtime/actions" "$WORK_DIR/server-runtime-actions.response.json" "${AUTH_HEADER[@]}" + node - "$WORK_DIR/server-runtime-actions.response.json" <<'NODE' const fs = require("fs"); const response = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); const action = (response.actions || []).find((candidate) => candidate.key === "generate-run"); @@ -941,7 +1279,110 @@ if (!action || action.available !== true || (action.reason || "").includes("run process.exit(1); } NODE -reject_forbidden_fragments "$WORK_DIR/server-runtime-actions.response.json" + reject_forbidden_fragments "$WORK_DIR/server-runtime-actions.response.json" +fi + +printf 'configuring current log session fixture on the local Run endpoint\n' +curl -fsS -X PUT -H 'Content-Type: application/json' "${AUTH_HEADER[@]}" \ + --data-binary "@$WORK_DIR/log-session-runtime-binding.request.json" \ + "$API_URL/server-instances/$LOG_SESSION_SERVER_ID/runtime-binding" >"$WORK_DIR/log-session-runtime-binding.response.json" +reject_forbidden_fragments "$WORK_DIR/log-session-runtime-binding.response.json" +require_file_contains "$WORK_DIR/log-session-runtime-binding.response.json" '"status"[[:space:]]*:[[:space:]]*"complete"' +curl -fsS -X PUT -H 'Content-Type: application/json' "${AUTH_HEADER[@]}" \ + --data-binary "@$WORK_DIR/log-session-deployment.request.json" \ + "$API_URL/server-instances/$LOG_SESSION_SERVER_ID/deployment" >"$WORK_DIR/log-session-deployment.response.json" +reject_forbidden_fragments "$WORK_DIR/log-session-deployment.response.json" + +write_server_lifecycle_request deploy "$WORK_DIR/log-session-deploy.request.json" +json_post "$API_URL/server-instances/$LOG_SESSION_SERVER_ID/deploy" "$WORK_DIR/log-session-deploy.request.json" "$WORK_DIR/log-session-deploy.response.json" "${AUTH_HEADER[@]}" +wait_for_job_success "$WORK_DIR/log-session-deploy.response.json" "$WORK_DIR/log-session-deploy-job.response.json" "log session deploy" +LOG_SESSION_DEPLOY_JOB_ID="$(json_id "$WORK_DIR/log-session-deploy-job.response.json")" +LOG_SESSION_REAL_JOB_STREAM_IDS="job.$LOG_SESSION_DEPLOY_JOB_ID.stdout,job.$LOG_SESSION_DEPLOY_JOB_ID.stderr" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-HISTORICAL-JOB-STDOUT" "$WORK_DIR/log-session-job-stdout-history.response.json" history +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-HISTORICAL-JOB-STDERR" "$WORK_DIR/log-session-job-stderr-history.response.json" history + +printf 'seeding explicit legacy, job, and file-tail history stream metadata\n' +create_historical_log_stream "$LOG_SESSION_LEGACY_STREAM_ID" process legacy.stdout +create_historical_log_stream "$LOG_SESSION_JOB_STREAM_ID" process job.stdout +create_historical_log_stream "$LOG_SESSION_FILE_STREAM_ID" file file.tail + +printf 'starting generation A without an open terminal or SSE client\n' +dispatch_log_session_lifecycle start "$WORK_DIR/log-session-start-a.response.json" "$WORK_DIR/log-session-start-a-job.response.json" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-A-START-STDOUT" "$WORK_DIR/log-session-a-stdout-history.response.json" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-A-START-STDERR" "$WORK_DIR/log-session-a-stderr-history.response.json" +LOG_SESSION_A="$(current_log_session_id)" +if [[ -z "$LOG_SESSION_A" ]]; then + dump_log_session_diagnostics "generation A did not expose a current session" + exit 1 +fi +printf '%s\n' "$LOG_SESSION_A" >"$WORK_DIR/current-log-session-a.id" + +printf 'opening one persistent Platform SSE connection after generation A was already uploaded\n' +: >"$LOG_SESSION_SSE_FILE" +: >"$LOG_SESSION_SSE_ERROR_FILE" +curl -fsS --no-buffer --max-time 240 -H "Authorization: Bearer $SESSION_ID" -H 'Accept: text/event-stream' \ + "$API_URL/server-instances/$LOG_SESSION_SERVER_ID/logs/events?historyLimit=200" \ + >"$LOG_SESSION_SSE_FILE" 2>"$LOG_SESSION_SSE_ERROR_FILE" & +LOG_SESSION_SSE_PID="$!" +wait_for_file_literal "initial SSE ready event" "$LOG_SESSION_SSE_FILE" 'event: ready' 60 +wait_for_file_literal "generation A stdout replay" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-START-STDOUT" 60 +wait_for_file_literal "generation A stderr replay" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-START-STDERR" 60 +LOG_SESSION_A_FROM_SSE="$(node "$ROOT_DIR/scripts/local-debug/verify-current-log-session-sse.mjs" session-for-marker "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-START-STDOUT")" +if [[ "$LOG_SESSION_A_FROM_SSE" != "$LOG_SESSION_A" ]]; then + dump_log_session_diagnostics "generation A history and SSE selected different sessions" + exit 1 +fi + +printf 'restarting Run while generation A remains alive\n' +restart_bootstrap_run_preserving_process +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-A-RUN-RESUME-STDOUT" "$WORK_DIR/log-session-a-resume-stdout-history.response.json" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-A-RUN-RESUME-STDERR" "$WORK_DIR/log-session-a-resume-stderr-history.response.json" +wait_for_file_literal "generation A stdout after Run restart" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-RUN-RESUME-STDOUT" 60 +wait_for_file_literal "generation A stderr after Run restart" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-RUN-RESUME-STDERR" 60 +LOG_SESSION_A_AFTER_RUN_RESTART="$(node "$ROOT_DIR/scripts/local-debug/verify-current-log-session-sse.mjs" session-for-marker "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-A-RUN-RESUME-STDOUT")" +if [[ "$LOG_SESSION_A_AFTER_RUN_RESTART" != "$LOG_SESSION_A" || "$(current_log_session_id)" != "$LOG_SESSION_A" ]]; then + dump_log_session_diagnostics "Run restart changed the supervised process session" + exit 1 +fi + +printf 'stopping generation A and starting generation B on the same SSE connection\n' +dispatch_log_session_lifecycle stop "$WORK_DIR/log-session-stop-a.response.json" "$WORK_DIR/log-session-stop-a-job.response.json" +printf 'B\n' >"$LOG_SESSION_SCOPE/smoke-generation.next" +mv "$LOG_SESSION_SCOPE/smoke-generation.next" "$LOG_SESSION_SCOPE/smoke-generation" +rm -f "$LOG_SESSION_SCOPE/smoke-command" "$LOG_SESSION_SCOPE/smoke-last-command" "$LOG_SESSION_SCOPE/smoke-process.pid" +dispatch_log_session_lifecycle start "$WORK_DIR/log-session-start-b.response.json" "$WORK_DIR/log-session-start-b-job.response.json" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-B-START-STDOUT" "$WORK_DIR/log-session-b-stdout-history.response.json" +wait_for_persisted_log_marker "$LOG_SESSION_MARKER_PREFIX-B-START-STDERR" "$WORK_DIR/log-session-b-stderr-history.response.json" +wait_for_file_literal "generation B stdout" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-B-START-STDOUT" 60 +wait_for_file_literal "generation B stderr" "$LOG_SESSION_SSE_FILE" "$LOG_SESSION_MARKER_PREFIX-B-START-STDERR" 60 +LOG_SESSION_B="$(current_log_session_id)" +if [[ -z "$LOG_SESSION_B" || "$LOG_SESSION_B" == "$LOG_SESSION_A" ]]; then + dump_log_session_diagnostics "generation B did not create a new current session" + exit 1 +fi +printf '%s\n' "$LOG_SESSION_B" >"$WORK_DIR/current-log-session-b.id" + +node "$ROOT_DIR/scripts/local-debug/verify-current-log-session-sse.mjs" verify \ + "$LOG_SESSION_SSE_FILE" \ + "$LOG_SESSION_SERVER_ID" \ + "$LOG_SESSION_MARKER_PREFIX" \ + "$LOG_SESSION_A" \ + "$LOG_SESSION_LEGACY_STREAM_ID,$LOG_SESSION_JOB_STREAM_ID,$LOG_SESSION_FILE_STREAM_ID,$LOG_SESSION_REAL_JOB_STREAM_IDS" \ + >"$WORK_DIR/current-log-session-verification.json" +reject_forbidden_fragments "$LOG_SESSION_SSE_FILE" +reject_forbidden_fragments "$WORK_DIR/current-log-session-verification.json" +dispatch_log_session_lifecycle stop "$WORK_DIR/log-session-stop-b.response.json" "$WORK_DIR/log-session-stop-b-job.response.json" +if kill -0 "$LOG_SESSION_SSE_PID" 2>/dev/null; then + kill "$LOG_SESSION_SSE_PID" 2>/dev/null || true + wait "$LOG_SESSION_SSE_PID" 2>/dev/null || true +fi +LOG_SESSION_SSE_PID="" +printf 'current supervised log session smoke passed; evidence: %s\n' "$WORK_DIR/current-log-session-verification.json" +if [[ "${LOCAL_DEBUG_LOG_SESSION_SMOKE_ONLY:-false}" == "true" ]]; then + printf 'focused current supervised log session smoke passed\n' + printf 'evidence directory: %s\n' "$WORK_DIR" + exit 0 +fi printf 'generating host-native example Run through platform Docker builder\n' curl -fsS -H 'Content-Type: application/json' "${AUTH_HEADER[@]}" --data-binary "@$WORK_DIR/server-run-generate.request.json" "$API_URL/server-instances/$SERVER_ID/run/generate" >"$WORK_DIR/server-run-generate.response.json" diff --git a/scripts/local-debug/verify-current-log-session-sse.mjs b/scripts/local-debug/verify-current-log-session-sse.mjs new file mode 100644 index 0000000..9de8fee --- /dev/null +++ b/scripts/local-debug/verify-current-log-session-sse.mjs @@ -0,0 +1,152 @@ +#!/usr/bin/env node + +import fs from "node:fs"; + +function fail(message, details = undefined) { + console.error(message); + if (details !== undefined) console.error(JSON.stringify(details, null, 2)); + process.exit(1); +} + +function parseEvents(file) { + const body = fs.readFileSync(file, "utf8").replace(/\r\n/g, "\n"); + const events = []; + for (const block of body.split("\n\n")) { + const lines = block.split("\n"); + let event = "message"; + let id = ""; + const data = []; + for (const line of lines) { + if (line.startsWith("event:")) event = line.slice(6).trim(); + if (line.startsWith("id:")) id = line.slice(3).trim(); + if (line.startsWith("data:")) data.push(line.slice(5).trimStart()); + } + if (data.length === 0) continue; + try { + events.push({ event, id, data: JSON.parse(data.join("\n")) }); + } catch { + // The final block may be incomplete while curl is still appending. + } + } + return events; +} + +function markerEvent(events, marker) { + return events.find((item) => item.event === "log" && item.data?.entry?.line === marker); +} + +function requireMarker(events, marker) { + const item = markerEvent(events, marker); + if (!item) fail(`missing SSE log marker: ${marker}`); + return item; +} + +const [command, file, ...args] = process.argv.slice(2); +if (!command || !file) { + fail("usage: verify-current-log-session-sse.mjs [...args]"); +} + +const events = parseEvents(file); +if (command === "session-for-marker") { + const [marker] = args; + const item = requireMarker(events, marker); + if (!item.data.logSessionId) fail(`marker has no logSessionId: ${marker}`, item); + process.stdout.write(item.data.logSessionId); + process.exit(0); +} + +if (command !== "verify") fail(`unknown command: ${command}`); + +const [serverID, prefix, expectedSessionA, forbiddenIDsText = ""] = args; +if (!serverID || !prefix || !expectedSessionA) fail("verify requires server id, marker prefix, and generation A session id"); + +const markers = { + aStdout: `${prefix}-A-START-STDOUT`, + aStderr: `${prefix}-A-START-STDERR`, + aResumeStdout: `${prefix}-A-RUN-RESUME-STDOUT`, + aResumeStderr: `${prefix}-A-RUN-RESUME-STDERR`, + bStdout: `${prefix}-B-START-STDOUT`, + bStderr: `${prefix}-B-START-STDERR` +}; +const markerItems = Object.fromEntries(Object.entries(markers).map(([key, marker]) => [key, requireMarker(events, marker)])); +const eventIndex = (needle) => events.indexOf(needle); +const sessionEvents = events.filter((item) => item.event === "session" && item.data?.logSessionId); +const sessionIDs = sessionEvents.map((item) => item.data.logSessionId); + +if (sessionIDs.length !== 2 || sessionIDs[0] !== expectedSessionA) { + fail("expected exactly one generation A boundary followed by one generation B boundary", { sessionIDs, expectedSessionA }); +} +const sessionB = sessionIDs[1]; +if (!sessionB || sessionB === expectedSessionA) fail("generation B did not receive a new session id", { sessionIDs }); + +for (const key of ["aStdout", "aStderr", "aResumeStdout", "aResumeStderr"]) { + if (markerItems[key].data.logSessionId !== expectedSessionA) { + fail(`generation A marker changed session at ${key}`, markerItems[key]); + } +} +for (const key of ["bStdout", "bStderr"]) { + if (markerItems[key].data.logSessionId !== sessionB) fail(`generation B marker has the wrong session at ${key}`, markerItems[key]); +} + +const boundaryBIndex = eventIndex(sessionEvents[1]); +const firstBLogIndex = Math.min(eventIndex(markerItems.bStdout), eventIndex(markerItems.bStderr)); +if (boundaryBIndex < 0 || boundaryBIndex >= firstBLogIndex) { + fail("generation B session boundary was not delivered before generation B output", { boundaryBIndex, firstBLogIndex }); +} +if (eventIndex(markerItems.aResumeStdout) >= boundaryBIndex || eventIndex(markerItems.aResumeStderr) >= boundaryBIndex) { + fail("Run-resume output arrived after the process-generation boundary"); +} + +const allowedStreamKeys = new Set(["smoke.console.stdout", "smoke.console.stderr"]); +const forbiddenIDs = new Set(forbiddenIDsText.split(",").filter(Boolean)); +const streamIDsBySession = new Map(); +for (let index = 0; index < events.length; index += 1) { + const item = events[index]; + if (item.data?.serverInstanceId && item.data.serverInstanceId !== serverID) { + fail("SSE feed included another server instance", item); + } + const streamID = item.event === "stream" ? item.data?.id : item.data?.streamId; + if (streamID && forbiddenIDs.has(streamID)) fail("SSE feed included an explicitly historical stream", item); + if (item.event !== "stream" && item.event !== "log") continue; + const sessionID = item.data?.logSessionId; + const source = item.data?.source; + const streamKey = item.data?.streamKey; + if (item.data?.entry?.line === `${prefix}-HISTORICAL-JOB-STDOUT` || item.data?.entry?.line === `${prefix}-HISTORICAL-JOB-STDERR`) { + fail("live SSE included historical job output", item); + } + if (source !== "process" || !allowedStreamKeys.has(streamKey)) { + fail("live SSE included a legacy, job, file-tail, or undeclared stream", item); + } + if (index < boundaryBIndex && sessionID !== expectedSessionA) { + fail("generation A feed mixed output from another or sessionless stream", item); + } + if (index > boundaryBIndex && sessionID !== sessionB) { + fail("post-switch SSE mixed output from another session", item); + } + if (index > boundaryBIndex && item.data?.entry?.line?.startsWith(`${prefix}-A-`)) { + fail("post-switch SSE retained generation A output", item); + } + if (streamID && sessionID) { + if (!streamIDsBySession.has(sessionID)) streamIDsBySession.set(sessionID, new Set()); + streamIDsBySession.get(sessionID).add(streamID); + } +} + +const streamsA = streamIDsBySession.get(expectedSessionA) ?? new Set(); +const streamsB = streamIDsBySession.get(sessionB) ?? new Set(); +if (streamsA.size < 2 || streamsB.size < 2) fail("stdout/stderr stream metadata was incomplete", { streamsA: [...streamsA], streamsB: [...streamsB] }); +for (const streamID of streamsA) { + if (streamsB.has(streamID)) fail("process generations reused a stream id", { streamID }); +} + +process.stdout.write(`${JSON.stringify({ + serverInstanceId: serverID, + sessionA: expectedSessionA, + sessionB, + sessionBoundaries: sessionIDs, + markerOrder: Object.fromEntries(Object.entries(markerItems).map(([key, item]) => [key, eventIndex(item)])), + generationAStreamIds: [...streamsA].sort(), + generationBStreamIds: [...streamsB].sort(), + excludedHistoricalStreamIds: [...forbiddenIDs].sort(), + eventCount: events.length +}, null, 2)}\n`);