browser/platform/api/log_ingest_handlers_test.go

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)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 2)

	ackRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, ackRecorder, http.StatusOK)
	ack := decodeBody[dto.LogBatchIngestResponse](t, ackRecorder)
	if !ack.Accepted || ack.AcceptedFrom != 1 || ack.AcceptedTo != 2 || ack.LatestSeq != 2 {
		t.Fatalf("unexpected ack: %+v", ack)
	}

	stream := getJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams/log-1")
	if stream.LatestSeq != 2 {
		t.Fatalf("expected latest seq update, got %+v", stream)
	}

	queryRecorder := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: "log-1", AfterSeq: 0, Limit: 1})
	assertStatus(t, queryRecorder, http.StatusOK)
	query := decodeBody[dto.LogStreamCursorResponse](t, queryRecorder)
	if len(query.Entries) != 1 || query.Entries[0].Seq != 1 || query.NextSeq != 1 || query.LatestSeq != 2 {
		t.Fatalf("unexpected query: %+v", query)
	}
}

func TestLiveLogRelayAPIForwardsWithoutStoredOutput(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 1)

	relay := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", batch)
	assertStatus(t, relay, http.StatusOK)
	ack := decodeBody[dto.LogBatchIngestResponse](t, relay)
	if !ack.Accepted || ack.AcceptedFrom != 1 || ack.AcceptedTo != 1 {
		t.Fatalf("unexpected live relay ack: %+v", ack)
	}

	query := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: batch.LogStreamID, AfterSeq: 0, Limit: 10})
	assertStatus(t, query, http.StatusOK)
	body := decodeBody[dto.LogStreamCursorResponse](t, query)
	if len(body.Entries) != 0 || body.LatestSeq != 1 {
		t.Fatalf("live relay stored platform log output: %+v", body)
	}
}

func TestLogEventsSSEDoesNotReplayHistory(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 2)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch), 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(recorder.Header().Get("Content-Type"), "text/event-stream") || !strings.Contains(body, "event: stream") || !strings.Contains(body, "event: ready") || strings.Contains(body, "event: log") || strings.Contains(body, `"seq":1`) || strings.Contains(body, `"seq":2`) {
		t.Fatalf("expected live-only SSE events without history replay, headers=%v body=%s", recorder.Header(), body)
	}
}

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", 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)
	retimestampLogBatchRequest(t, &next, time.Now().UTC().Add(time.Second))
	if _, err := core.IngestLogBatch(next.ToDomain()); err != nil {
		t.Fatalf("ingest next live batch: %v", err)
	}
	assertSSEEvent(t, reader, "log", `"seq":2`)
}

func TestLogEventsSSERelaysLiveBatchesWithoutSequenceGate(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events", nil).WithContext(ctx)
	streamWriter, streamReader := newSSEPipeResponseWriter()
	done := make(chan struct{})
	go func() {
		router.ServeHTTP(streamWriter, request)
		_ = streamWriter.Close()
		close(done)
	}()
	t.Cleanup(func() {
		cancel()
		_ = streamReader.Close()
		<-done
	})
	if status := <-streamWriter.status; status != http.StatusOK {
		t.Fatalf("unexpected SSE status: %d", status)
	}
	reader := bufio.NewReader(streamReader)
	assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
	assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
	assertSSEEvent(t, reader, "ready", `"streamCount":1`)

	first := validLogBatchRequest(t, hello.SessionToken, 4, 4)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", first), http.StatusOK)
	assertSSEEvent(t, reader, "log", `"seq":4`)

	second := validLogBatchRequest(t, hello.SessionToken, 1, 1)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/relay", second), http.StatusOK)
	assertSSEEvent(t, reader, "log", `"seq":1`)
}

func TestLogEventsSSESkipsBufferedBackfillAfterOpen(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	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", nil).WithContext(ctx)
	streamWriter, streamReader := newSSEPipeResponseWriter()
	done := make(chan struct{})
	go func() {
		router.ServeHTTP(streamWriter, request)
		_ = streamWriter.Close()
		close(done)
	}()
	t.Cleanup(func() {
		cancel()
		_ = streamReader.Close()
		<-done
	})
	if status := <-streamWriter.status; status != http.StatusOK {
		t.Fatalf("unexpected SSE status: %d", status)
	}
	reader := bufio.NewReader(streamReader)
	assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
	assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
	assertSSEEvent(t, reader, "ready", `"streamCount":1`)

	stale := validLogBatchRequest(t, hello.SessionToken, 2, 2)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", stale), http.StatusOK)
	fresh := validLogBatchRequest(t, hello.SessionToken, 3, 3)
	retimestampLogBatchRequest(t, &fresh, time.Now().UTC().Add(time.Second))
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", fresh), http.StatusOK)
	assertSSEEvent(t, reader, "log", `"seq":3`)
}

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")
	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"))

	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"))
	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"))
}

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", nil).WithContext(ctx)
	streamWriter, streamReader := newSSEPipeResponseWriter()
	done := make(chan struct{})
	go func() {
		router.ServeHTTP(streamWriter, request)
		_ = streamWriter.Close()
		close(done)
	}()
	t.Cleanup(func() {
		cancel()
		_ = streamReader.Close()
		<-done
	})
	if status := <-streamWriter.status; status != http.StatusOK {
		t.Fatalf("unexpected SSE status: %d", status)
	}
	reader := bufio.NewReader(streamReader)
	assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
	assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
	assertSSEEvent(t, reader, "ready", `"streamCount":1`)

	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"`)
	nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 21)
	nextLive.LogSessionID = "session-next"
	nextLive.SessionStartedAt = nextStartedAt
	retimestampLogBatchRequest(t, &nextLive, time.Now().UTC().Add(time.Second))
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", nextLive), http.StatusOK)
	assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`)
}

func TestLogEventsSSEOrdersSessionSwitchAndAdditionalStreamWithoutDuplicates(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", 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)
	}

	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"`)

	stderr := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stderr", "stderr", 1, 1, 21)
	stderr.LogSessionID = "session-next"
	stderr.SessionStartedAt = nextStartedAt
	retimestampLogBatchRequest(t, &stderr, time.Now().UTC().Add(time.Second))
	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
	retimestampLogBatchRequest(t, &nextLive, time.Now().UTC().Add(time.Second))
	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())
	cancel()
	req := httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx)
	rec := httptest.NewRecorder()
	router.ServeHTTP(rec, req)
	return rec
}

func TestLogIngestAPIDuplicateAndErrors(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 1)

	first := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, first, http.StatusOK)
	duplicate := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, duplicate, http.StatusOK)
	duplicateAck := decodeBody[dto.LogBatchIngestResponse](t, duplicate)
	if !duplicateAck.Duplicate {
		t.Fatalf("expected duplicate ack, got %+v", duplicateAck)
	}

	gap := validLogBatchRequest(t, hello.SessionToken, 3, 3)
	gapRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", gap)
	assertErrorResponse(t, gapRecorder, http.StatusBadRequest, errorCodeValidation)

	missingQuery := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: "missing", Limit: 1})
	assertErrorResponse(t, missingQuery, http.StatusNotFound, errorCodeNotFound)
}

func createLogIngestAPIFixtures(t *testing.T, router http.Handler) dto.RunControlHelloResponse {
	t.Helper()
	helloRequest := validRunControlHelloRequest()
	helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, "process.install", "process.start", "process.stop", "logs.read")
	helloRequest.CapabilityReport.Fingerprint = "cap-logs"
	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", 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
}

func validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) dto.LogBatchIngestRequest {
	return validLogBatchRequestForStream(t, sessionToken, "log-1", "stdout", firstSeq, lastSeq, 0)
}

func validLogBatchRequestForStream(t *testing.T, sessionToken string, streamID string, streamKey string, firstSeq uint64, lastSeq uint64, timestampOffset int) dto.LogBatchIngestRequest {
	t.Helper()
	entries := make([]dto.LogEntryBody, 0, lastSeq-firstSeq+1)
	domainEntries := make([]domain.LogEntry, 0, lastSeq-firstSeq+1)
	for seq := firstSeq; seq <= lastSeq; seq++ {
		entry := dto.LogEntryBody{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, timestampOffset+int(seq), 0, time.UTC), Level: "info", Line: "line"}
		entries = append(entries, entry)
		domainEntries = append(domainEntries, domain.LogEntry{Seq: entry.Seq, Timestamp: entry.Timestamp, Level: entry.Level, Line: entry.Line})
	}
	checksum, err := validator.LogEntriesChecksum(domainEntries)
	if err != nil {
		t.Fatalf("checksum entries: %v", err)
	}
	return dto.LogBatchIngestRequest{
		RunEndpointID:    "run-local",
		SessionToken:     sessionToken,
		LogStreamID:      streamID,
		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",
		Checksum:         checksum,
		Entries:          entries,
	}
}

func retimestampLogBatchRequest(t *testing.T, request *dto.LogBatchIngestRequest, first time.Time) {
	t.Helper()
	domainEntries := make([]domain.LogEntry, 0, len(request.Entries))
	for index := range request.Entries {
		request.Entries[index].Timestamp = first.Add(time.Duration(index) * time.Millisecond).UTC()
		domainEntries = append(domainEntries, domain.LogEntry{Seq: request.Entries[index].Seq, Timestamp: request.Entries[index].Timestamp, Level: request.Entries[index].Level, Line: request.Entries[index].Line, Fields: request.Entries[index].Fields, Redacted: request.Entries[index].Redacted})
	}
	checksum, err := validator.LogEntriesChecksum(domainEntries)
	if err != nil {
		t.Fatalf("checksum retimestamped entries: %v", err)
	}
	request.Checksum = checksum
}