Make terminal log stream live-only by default

2026-08-25 09:42:24 +08:00
parent 199fa8b41d
commit 725b71b6b6
7 changed files with 72 additions and 9 deletions
@@ -95,6 +95,63 @@ func TestLogEventsSSEReplaysHistory(t *testing.T) {
 	}
 }

+func TestLogEventsSSEDefaultsToLiveOnly(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, 2)), http.StatusOK)
+
+	recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events")
+	assertStatus(t, recorder, http.StatusOK)
+	body := recorder.Body.String()
+	if !strings.Contains(body, "event: stream") || !strings.Contains(body, "event: ready") || strings.Contains(body, "event: log") {
+		t.Fatalf("expected live-only SSE snapshot without history logs, body=%s", body)
+	}
+}
+
+func TestLogEventsSSELiveOnlyStartsAfterSnapshotTail(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=0", 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, "ready", `"streamCount":1`)
+	if err := <-hookResult; err != nil {
+		t.Fatalf("ingest during stream snapshot: %v", err)
+	}
+	next := validLogBatchRequest(t, hello.SessionToken, 2, 2)
+	if _, err := core.IngestLogBatch(next.ToDomain()); err != nil {
+		t.Fatalf("ingest next live batch: %v", err)
+	}
+	assertSSEEvent(t, reader, "log", `"seq":2`)
+}
+
 func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) {
 	router := newTestRouter()
 	hello := createLogIngestAPIFixtures(t, router)