init

spool/README.md

@@ -0,0 +1,19 @@
+# run/spool
+
+Local durable queues live here.
+
+Required spool areas:
+
+- `logs`: unacknowledged log segments.
+- `jobs`: accepted job journal for duplicate detection and reconciliation.
+- `artifacts`: incomplete artifact transfer state.
+
+Spool pressure must be visible in run capacity reports.
+
+## Channel Isolation
+
+- Log and artifact retry state are stored in separate spool areas and are acknowledged independently.
+- Acknowledging a log batch must not scan, remove, or block on artifact chunks.
+- Acknowledging an artifact chunk must not scan, remove, or block on log batches.
+- Control heartbeat and job ack/result payloads remain metadata-only; they must never carry spool file paths, artifact chunks, log entries, raw credentials, direct sockets, or large inline bodies.
+- Artifact transfer backlog is lower priority than log flush, job lifecycle calls, and control heartbeat.

spool/artifact_queue.go

@@ -0,0 +1,183 @@
+package spool
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"browser.local/run/protocol"
+)
+
+type ArtifactQueue struct {
+	dir string
+}
+
+func NewArtifactQueue(dir string) (ArtifactQueue, error) {
+	if strings.TrimSpace(dir) == "" {
+		return ArtifactQueue{}, fmt.Errorf("spool directory is required")
+	}
+	artifactDir := filepath.Join(dir, "artifacts")
+	if err := os.MkdirAll(artifactDir, 0o755); err != nil {
+		return ArtifactQueue{}, fmt.Errorf("create artifact queue: %w", err)
+	}
+	return ArtifactQueue{dir: artifactDir}, nil
+}
+
+func (queue ArtifactQueue) Enqueue(chunk protocol.ArtifactChunkUploadRequest) error {
+	path := queue.chunkPath(chunk)
+	if existing, err := readArtifactChunk(path); err == nil {
+		if existing.TransferID == chunk.TransferID && existing.ArtifactID == chunk.ArtifactID && existing.ChunkIndex == chunk.ChunkIndex && existing.Checksum == chunk.Checksum {
+			return nil
+		}
+		return fmt.Errorf("artifact queue chunk conflicts with committed chunk")
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+	tmp := path + ".tmp"
+	file, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
+	if err != nil {
+		return fmt.Errorf("open artifact queue chunk: %w", err)
+	}
+	encodeErr := json.NewEncoder(file).Encode(chunk)
+	closeErr := file.Close()
+	if encodeErr != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("encode artifact queue chunk: %w", encodeErr)
+	}
+	if closeErr != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("close artifact queue chunk: %w", closeErr)
+	}
+	if err := syncFile(tmp); err != nil {
+		_ = os.Remove(tmp)
+		return err
+	}
+	if err := os.Rename(tmp, path); err != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("commit artifact queue chunk: %w", err)
+	}
+	return nil
+}
+
+type ArtifactChunkClient interface {
+	UploadArtifactChunk(context.Context, protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error)
+}
+
+func (queue ArtifactQueue) Flush(ctx context.Context, client ArtifactChunkClient) (int, error) {
+	if client == nil {
+		return 0, fmt.Errorf("artifact chunk client is required")
+	}
+	pending, err := queue.Pending()
+	if err != nil {
+		return 0, err
+	}
+	acknowledged := 0
+	for _, chunk := range pending {
+		if err := ctx.Err(); err != nil {
+			return acknowledged, err
+		}
+		response, err := client.UploadArtifactChunk(ctx, chunk)
+		if err != nil {
+			return acknowledged, err
+		}
+		if !response.Accepted || response.TransferID != chunk.TransferID || response.ArtifactID != chunk.ArtifactID || response.ChunkIndex != chunk.ChunkIndex {
+			return acknowledged, fmt.Errorf("platform artifact acknowledgement does not match pending chunk")
+		}
+		if err := queue.Ack(response); err != nil {
+			return acknowledged, err
+		}
+		acknowledged++
+	}
+	return acknowledged, nil
+}
+
+func readArtifactChunk(path string) (protocol.ArtifactChunkUploadRequest, error) {
+	file, err := os.Open(path)
+	if err != nil {
+		return protocol.ArtifactChunkUploadRequest{}, err
+	}
+	defer file.Close()
+	var chunk protocol.ArtifactChunkUploadRequest
+	if err := json.NewDecoder(file).Decode(&chunk); err != nil {
+		return protocol.ArtifactChunkUploadRequest{}, fmt.Errorf("decode artifact queue chunk: %w", err)
+	}
+	return chunk, nil
+}
+
+func (queue ArtifactQueue) Pending() ([]protocol.ArtifactChunkUploadRequest, error) {
+	entries, err := os.ReadDir(queue.dir)
+	if err != nil {
+		return nil, fmt.Errorf("read artifact queue: %w", err)
+	}
+	paths := make([]string, 0, len(entries))
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
+			continue
+		}
+		paths = append(paths, filepath.Join(queue.dir, entry.Name()))
+	}
+	sort.Strings(paths)
+	chunks := make([]protocol.ArtifactChunkUploadRequest, 0, len(paths))
+	for _, path := range paths {
+		file, err := os.Open(path)
+		if err != nil {
+			return nil, fmt.Errorf("open artifact queue chunk: %w", err)
+		}
+		var chunk protocol.ArtifactChunkUploadRequest
+		decodeErr := json.NewDecoder(file).Decode(&chunk)
+		closeErr := file.Close()
+		if decodeErr != nil {
+			return nil, fmt.Errorf("decode artifact queue chunk: %w", decodeErr)
+		}
+		if closeErr != nil {
+			return nil, fmt.Errorf("close artifact queue chunk: %w", closeErr)
+		}
+		chunks = append(chunks, chunk)
+	}
+	return chunks, nil
+}
+
+func (queue ArtifactQueue) Ack(response protocol.ArtifactChunkUploadResponse) error {
+	if !response.Accepted {
+		return nil
+	}
+	entries, err := os.ReadDir(queue.dir)
+	if err != nil {
+		return fmt.Errorf("read artifact queue: %w", err)
+	}
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
+			continue
+		}
+		path := filepath.Join(queue.dir, entry.Name())
+		file, err := os.Open(path)
+		if err != nil {
+			return fmt.Errorf("open artifact queue chunk: %w", err)
+		}
+		var chunk protocol.ArtifactChunkUploadRequest
+		decodeErr := json.NewDecoder(file).Decode(&chunk)
+		closeErr := file.Close()
+		if decodeErr != nil {
+			return fmt.Errorf("decode artifact queue chunk: %w", decodeErr)
+		}
+		if closeErr != nil {
+			return fmt.Errorf("close artifact queue chunk: %w", closeErr)
+		}
+		if chunk.TransferID == response.TransferID && chunk.ArtifactID == response.ArtifactID && chunk.ChunkIndex == response.ChunkIndex {
+			if err := os.Remove(path); err != nil {
+				return fmt.Errorf("remove acknowledged artifact queue chunk: %w", err)
+			}
+		}
+	}
+	return nil
+}
+
+func (queue ArtifactQueue) chunkPath(chunk protocol.ArtifactChunkUploadRequest) string {
+	transferID := sanitizeSegmentName(chunk.TransferID)
+	artifactID := sanitizeSegmentName(chunk.ArtifactID)
+	return filepath.Join(queue.dir, fmt.Sprintf("%s-%s-%020d.json", transferID, artifactID, chunk.ChunkIndex))
+}

spool/artifact_queue_test.go

@@ -0,0 +1,75 @@
+package spool
+
+import (
+	"testing"
+
+	"browser.local/run/protocol"
+)
+
+func TestArtifactQueueRetainsPendingAndRemovesAcknowledgedChunk(t *testing.T) {
+	queue, err := NewArtifactQueue(t.TempDir())
+	if err != nil {
+		t.Fatalf("new artifact queue: %v", err)
+	}
+	first := validQueuedArtifactChunk(0)
+	second := validQueuedArtifactChunk(1)
+	if err := queue.Enqueue(first); err != nil {
+		t.Fatalf("enqueue first: %v", err)
+	}
+	if err := queue.Enqueue(second); err != nil {
+		t.Fatalf("enqueue second: %v", err)
+	}
+	pending, err := queue.Pending()
+	if err != nil {
+		t.Fatalf("pending before ack: %v", err)
+	}
+	if len(pending) != 2 {
+		t.Fatalf("expected two pending chunks, got %+v", pending)
+	}
+
+	if err := queue.Ack(protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: "transfer-1", ArtifactID: "artifact-1", ChunkIndex: 0}); err != nil {
+		t.Fatalf("ack first: %v", err)
+	}
+	pending, err = queue.Pending()
+	if err != nil {
+		t.Fatalf("pending after ack: %v", err)
+	}
+	if len(pending) != 1 || pending[0].ChunkIndex != 1 {
+		t.Fatalf("expected second chunk pending, got %+v", pending)
+	}
+}
+
+func TestArtifactQueueRetainsChunkWhenAckDoesNotMatch(t *testing.T) {
+	queue, err := NewArtifactQueue(t.TempDir())
+	if err != nil {
+		t.Fatalf("new artifact queue: %v", err)
+	}
+	if err := queue.Enqueue(validQueuedArtifactChunk(0)); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if err := queue.Ack(protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: "transfer-1", ArtifactID: "artifact-1", ChunkIndex: 1}); err != nil {
+		t.Fatalf("ack mismatch: %v", err)
+	}
+	pending, err := queue.Pending()
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(pending) != 1 || pending[0].ChunkIndex != 0 {
+		t.Fatalf("expected original chunk pending, got %+v", pending)
+	}
+}
+
+func validQueuedArtifactChunk(index int) protocol.ArtifactChunkUploadRequest {
+	payload := []byte{byte(index), byte(index + 1)}
+	return protocol.ArtifactChunkUploadRequest{
+		RunEndpointID: "run-local",
+		SessionToken:  "session-token",
+		TransferID:    "transfer-1",
+		ArtifactID:    "artifact-1",
+		ChunkIndex:    index,
+		Offset:        int64(index * len(payload)),
+		SizeBytes:     len(payload),
+		Checksum:      "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+		Payload:       payload,
+	}
+}

spool/channel_isolation_test.go

@@ -0,0 +1,85 @@
+package spool
+
+import (
+	"testing"
+
+	"browser.local/run/protocol"
+)
+
+func TestLogSpoolAckIsIndependentFromArtifactBacklog(t *testing.T) {
+	root := t.TempDir()
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	artifactQueue, err := NewArtifactQueue(root)
+	if err != nil {
+		t.Fatalf("new artifact queue: %v", err)
+	}
+
+	if err := logSpool.Enqueue(validSpoolLogBatch(1, 1)); err != nil {
+		t.Fatalf("enqueue log: %v", err)
+	}
+	for i := 0; i < 3; i++ {
+		if err := artifactQueue.Enqueue(validQueuedArtifactChunk(i)); err != nil {
+			t.Fatalf("enqueue artifact chunk %d: %v", i, err)
+		}
+	}
+
+	if err := logSpool.Ack(protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
+		t.Fatalf("ack log batch: %v", err)
+	}
+	logs, err := logSpool.Pending()
+	if err != nil {
+		t.Fatalf("pending logs: %v", err)
+	}
+	if len(logs) != 0 {
+		t.Fatalf("expected log batch removed despite artifact backlog, got %+v", logs)
+	}
+	chunks, err := artifactQueue.Pending()
+	if err != nil {
+		t.Fatalf("pending artifacts: %v", err)
+	}
+	if len(chunks) != 3 {
+		t.Fatalf("artifact backlog should remain independent, got %+v", chunks)
+	}
+}
+
+func TestArtifactAckIsIndependentFromLogBacklog(t *testing.T) {
+	root := t.TempDir()
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	artifactQueue, err := NewArtifactQueue(root)
+	if err != nil {
+		t.Fatalf("new artifact queue: %v", err)
+	}
+
+	for _, batch := range []protocol.LogBatchIngestRequest{validSpoolLogBatch(1, 1), validSpoolLogBatch(2, 2)} {
+		if err := logSpool.Enqueue(batch); err != nil {
+			t.Fatalf("enqueue log batch: %v", err)
+		}
+	}
+	if err := artifactQueue.Enqueue(validQueuedArtifactChunk(0)); err != nil {
+		t.Fatalf("enqueue artifact chunk: %v", err)
+	}
+
+	if err := artifactQueue.Ack(protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: "transfer-1", ArtifactID: "artifact-1", ChunkIndex: 0}); err != nil {
+		t.Fatalf("ack artifact chunk: %v", err)
+	}
+	chunks, err := artifactQueue.Pending()
+	if err != nil {
+		t.Fatalf("pending artifacts: %v", err)
+	}
+	if len(chunks) != 0 {
+		t.Fatalf("expected artifact chunk removed despite log backlog, got %+v", chunks)
+	}
+	logs, err := logSpool.Pending()
+	if err != nil {
+		t.Fatalf("pending logs: %v", err)
+	}
+	if len(logs) != 2 {
+		t.Fatalf("log backlog should remain independent, got %+v", logs)
+	}
+}

spool/flush_test.go

@@ -0,0 +1,36 @@
+package spool
+
+import (
+	"context"
+	"testing"
+
+	"browser.local/run/protocol"
+)
+
+type fakeLogBatchClient struct {
+	accepted bool
+}
+
+func (client fakeLogBatchClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
+	return protocol.LogBatchIngestResponse{Accepted: client.accepted, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
+}
+
+func TestLogSpoolFlushRetainsRejectedBatchForRetry(t *testing.T) {
+	spool, err := NewLogSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("new spool: %v", err)
+	}
+	if err := spool.Enqueue(validSpoolLogBatch(1, 1)); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if _, err := spool.Flush(context.Background(), fakeLogBatchClient{accepted: false}); err == nil {
+		t.Fatal("expected rejected flush")
+	}
+	pending, err := spool.Pending()
+	if err != nil || len(pending) != 1 {
+		t.Fatalf("expected batch retained after rejection, pending=%+v err=%v", pending, err)
+	}
+	if count, err := spool.Flush(context.Background(), fakeLogBatchClient{accepted: true}); err != nil || count != 1 {
+		t.Fatalf("expected retry success, count=%d err=%v", count, err)
+	}
+}

spool/log_spool.go

@@ -0,0 +1,560 @@
+package spool
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+
+	"browser.local/run/protocol"
+)
+
+type LogSpool struct {
+	dir        string
+	statePath  string
+	mu         *sync.Mutex
+	watermarks map[string]logStreamWatermark
+	inflight   map[string]bool
+}
+
+const maxAggregatedLogEntries = 128
+
+type logStreamWatermark struct {
+	Allocated    uint64 `json:"allocated"`
+	Acknowledged uint64 `json:"acknowledged"`
+	SourceOffset int64  `json:"sourceOffset,omitempty"`
+}
+type logSpoolState struct {
+	Streams map[string]logStreamWatermark `json:"streams"`
+}
+
+type LogSourceCursor struct {
+	StartOffset int64 `json:"startOffset"`
+	EndOffset   int64 `json:"endOffset"`
+}
+
+type durableLogBatch struct {
+	protocol.LogBatchIngestRequest
+	SourceCursor *LogSourceCursor `json:"_sourceCursor,omitempty"`
+}
+
+type pendingLogSegment struct {
+	path  string
+	batch durableLogBatch
+}
+
+func NewLogSpool(dir string) (LogSpool, error) {
+	if strings.TrimSpace(dir) == "" {
+		return LogSpool{}, fmt.Errorf("spool directory is required")
+	}
+	logDir := filepath.Join(dir, "logs")
+	if err := os.MkdirAll(logDir, 0o755); err != nil {
+		return LogSpool{}, fmt.Errorf("create log spool: %w", err)
+	}
+	spool := LogSpool{dir: logDir, statePath: filepath.Join(dir, "log-watermarks.json"), mu: &sync.Mutex{}, watermarks: map[string]logStreamWatermark{}, inflight: map[string]bool{}}
+	if err := spool.loadWatermarks(); err != nil {
+		return LogSpool{}, err
+	}
+	if err := spool.restorePendingWatermarks(); err != nil {
+		return LogSpool{}, err
+	}
+	return spool, nil
+}
+
+// restorePendingWatermarks supports upgrades from spools created before
+// per-stream watermarks existed. Pending durable segments are still an
+// allocation fact and must win over a remote progress lookup.
+func (spool *LogSpool) restorePendingWatermarks() error {
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	pending, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		return err
+	}
+	changed := false
+	for _, segment := range pending {
+		batch := segment.batch
+		watermark := spool.watermarks[batch.LogStreamID]
+		if batch.LastSeq > watermark.Allocated {
+			watermark.Allocated = batch.LastSeq
+			changed = true
+		}
+		if batch.SourceCursor != nil && batch.SourceCursor.EndOffset > watermark.SourceOffset {
+			watermark.SourceOffset = batch.SourceCursor.EndOffset
+			changed = true
+		}
+		spool.watermarks[batch.LogStreamID] = watermark
+	}
+	if changed {
+		return spool.persistWatermarksLocked()
+	}
+	return nil
+}
+
+func (spool LogSpool) NextSequence(ctx context.Context, streamID string, recover func(context.Context, string) (uint64, error)) (uint64, error) {
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	watermark, known := spool.watermarks[streamID]
+	if !known && recover != nil {
+		latest, err := recover(ctx, streamID)
+		if err != nil {
+			return 0, err
+		}
+		watermark = logStreamWatermark{Allocated: latest, Acknowledged: latest}
+	}
+	watermark.Allocated++
+	spool.watermarks[streamID] = watermark
+	if err := spool.persistWatermarksLocked(); err != nil {
+		return 0, err
+	}
+	return watermark.Allocated, nil
+}
+
+func (spool LogSpool) Enqueue(batch protocol.LogBatchIngestRequest) error {
+	return spool.enqueue(batch, nil)
+}
+
+// EnqueueAggregated extends the newest compatible durable segment so callers
+// do not create one upload request for every process-output line.
+func (spool LogSpool) EnqueueAggregated(batch protocol.LogBatchIngestRequest, checksum func([]protocol.LogEntry) (string, error)) error {
+	return spool.enqueue(batch, checksum)
+}
+
+// EnqueueNextAggregated allocates the next sequence and commits its batch
+// under one spool lock. The durable segment is the source of truth; startup
+// restores its watermark if the separate watermark snapshot was interrupted.
+func (spool LogSpool) EnqueueNextAggregated(ctx context.Context, batch protocol.LogBatchIngestRequest, cursor *LogSourceCursor, recover func(context.Context, string) (uint64, error), checksum func([]protocol.LogEntry) (string, error)) (uint64, bool, error) {
+	if len(batch.Entries) != 1 {
+		return 0, false, fmt.Errorf("next aggregated log batch requires exactly one entry")
+	}
+	if cursor != nil && (cursor.StartOffset < 0 || cursor.EndOffset <= cursor.StartOffset) {
+		return 0, false, fmt.Errorf("source cursor range is invalid")
+	}
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	watermark, known := spool.watermarks[batch.LogStreamID]
+	if !known && recover != nil {
+		latest, err := recover(ctx, batch.LogStreamID)
+		if err != nil {
+			return 0, false, err
+		}
+		watermark = logStreamWatermark{Allocated: latest, Acknowledged: latest}
+	}
+	if cursor != nil && cursor.EndOffset <= watermark.SourceOffset {
+		return watermark.Allocated, false, nil
+	}
+	sequence := watermark.Allocated + 1
+	batch.FirstSeq = sequence
+	batch.LastSeq = sequence
+	batch.Entries[0].Seq = sequence
+	var err error
+	batch.Checksum, err = checksum(batch.Entries)
+	if err != nil {
+		return 0, false, err
+	}
+	if err := spool.writeAggregatedLocked(durableLogBatch{LogBatchIngestRequest: batch, SourceCursor: cursor}, checksum); err != nil {
+		return 0, false, err
+	}
+	watermark.Allocated = sequence
+	if cursor != nil {
+		watermark.SourceOffset = cursor.EndOffset
+	}
+	spool.watermarks[batch.LogStreamID] = watermark
+	// The batch is already synced and durable. A later startup reconstructs
+	// this watermark from pending segments if this snapshot cannot be written.
+	_ = spool.persistWatermarksLocked()
+	return sequence, true, nil
+}
+
+func (spool LogSpool) enqueue(batch protocol.LogBatchIngestRequest, checksum func([]protocol.LogEntry) (string, error)) error {
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	watermark := spool.watermarks[batch.LogStreamID]
+	if batch.LastSeq > watermark.Allocated {
+		watermark.Allocated = batch.LastSeq
+		spool.watermarks[batch.LogStreamID] = watermark
+		if err := spool.persistWatermarksLocked(); err != nil {
+			return err
+		}
+	}
+	return spool.writeAggregatedLocked(durableLogBatch{LogBatchIngestRequest: batch}, checksum)
+}
+
+func (spool LogSpool) writeAggregatedLocked(batch durableLogBatch, checksum func([]protocol.LogEntry) (string, error)) error {
+	if checksum != nil && len(batch.Entries) == 1 {
+		if merged, previous, ok, err := spool.mergeLatest(batch, checksum); err != nil {
+			return err
+		} else if ok {
+			return spool.replaceBatch(previous.path, merged)
+		}
+	}
+	return spool.writeBatch(batch)
+}
+
+func (spool LogSpool) mergeLatest(batch durableLogBatch, checksum func([]protocol.LogEntry) (string, error)) (durableLogBatch, pendingLogSegment, bool, error) {
+	pending, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		return durableLogBatch{}, pendingLogSegment{}, false, err
+	}
+	for index := len(pending) - 1; index >= 0; index-- {
+		previous := pending[index]
+		if spool.inflight[previous.path] || !compatibleLogBatch(previous.batch, batch) || previous.batch.LastSeq+1 != batch.FirstSeq || len(previous.batch.Entries)+len(batch.Entries) > maxAggregatedLogEntries || !compatibleSourceCursor(previous.batch.SourceCursor, batch.SourceCursor) {
+			continue
+		}
+		merged := previous.batch
+		merged.LastSeq = batch.LastSeq
+		merged.Entries = append(append([]protocol.LogEntry(nil), previous.batch.Entries...), batch.Entries...)
+		if merged.SourceCursor != nil {
+			cursor := *merged.SourceCursor
+			cursor.EndOffset = batch.SourceCursor.EndOffset
+			merged.SourceCursor = &cursor
+		}
+		merged.Checksum, err = checksum(merged.Entries)
+		if err != nil {
+			return durableLogBatch{}, pendingLogSegment{}, false, err
+		}
+		return merged, previous, true, nil
+	}
+	return durableLogBatch{}, pendingLogSegment{}, false, nil
+}
+
+func compatibleLogBatch(previous durableLogBatch, next durableLogBatch) bool {
+	return previous.LogStreamID == next.LogStreamID && previous.ServerInstanceID == next.ServerInstanceID && previous.StreamKey == next.StreamKey && previous.Source == next.Source && previous.Compression == next.Compression && previous.LogSessionID == next.LogSessionID && previous.SessionStartedAt.Equal(next.SessionStartedAt)
+}
+
+func compatibleSourceCursor(previous *LogSourceCursor, next *LogSourceCursor) bool {
+	if previous == nil || next == nil {
+		return previous == nil && next == nil
+	}
+	return next.StartOffset >= previous.EndOffset
+}
+
+func (spool LogSpool) writeBatch(batch durableLogBatch) error {
+	path := spool.batchPath(batch)
+	if existing, err := readDurableLogBatch(path); err == nil {
+		if existing.LogStreamID == batch.LogStreamID && existing.FirstSeq == batch.FirstSeq && existing.LastSeq == batch.LastSeq && existing.Checksum == batch.Checksum {
+			return nil
+		}
+		return fmt.Errorf("log spool segment conflicts with committed batch")
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+	return spool.writeBatchAt(path, batch)
+}
+
+func (spool LogSpool) replaceBatch(path string, batch durableLogBatch) error {
+	if strings.TrimSpace(path) == "" {
+		return fmt.Errorf("log spool segment path is required")
+	}
+	return spool.writeBatchAt(path, batch)
+}
+
+func (spool LogSpool) writeBatchAt(path string, batch durableLogBatch) error {
+	tmp := path + ".tmp"
+	file, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
+	if err != nil {
+		return fmt.Errorf("open log spool segment: %w", err)
+	}
+	encodeErr := json.NewEncoder(file).Encode(batch)
+	closeErr := file.Close()
+	if encodeErr != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("encode log spool segment: %w", encodeErr)
+	}
+	if closeErr != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("close log spool segment: %w", closeErr)
+	}
+	if err := syncFile(tmp); err != nil {
+		_ = os.Remove(tmp)
+		return err
+	}
+	if err := os.Rename(tmp, path); err != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("commit log spool segment: %w", err)
+	}
+	return nil
+}
+
+func (spool LogSpool) Pending() ([]protocol.LogBatchIngestRequest, error) {
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	segments, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		return nil, err
+	}
+	batches := make([]protocol.LogBatchIngestRequest, 0, len(segments))
+	for _, segment := range segments {
+		batches = append(batches, segment.batch.LogBatchIngestRequest)
+	}
+	return batches, nil
+}
+
+func (spool LogSpool) pendingSegmentsLocked() ([]pendingLogSegment, error) {
+	entries, err := os.ReadDir(spool.dir)
+	if err != nil {
+		return nil, fmt.Errorf("read log spool: %w", err)
+	}
+	paths := make([]string, 0, len(entries))
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
+			continue
+		}
+		paths = append(paths, filepath.Join(spool.dir, entry.Name()))
+	}
+	sort.Strings(paths)
+	segments := make([]pendingLogSegment, 0, len(paths))
+	for _, path := range paths {
+		batch, err := readDurableLogBatch(path)
+		if err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			return nil, err
+		}
+		segments = append(segments, pendingLogSegment{path: path, batch: batch})
+	}
+	return segments, nil
+}
+
+func (spool LogSpool) Ack(response protocol.LogBatchIngestResponse) error {
+	if response.AcceptedFrom == 0 || response.AcceptedTo < response.AcceptedFrom {
+		return nil
+	}
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	watermark := spool.watermarks[response.LogStreamID]
+	if response.AcceptedTo > watermark.Acknowledged {
+		watermark.Acknowledged = response.AcceptedTo
+	}
+	if watermark.Allocated < watermark.Acknowledged {
+		watermark.Allocated = watermark.Acknowledged
+	}
+	spool.watermarks[response.LogStreamID] = watermark
+	if err := spool.persistWatermarksLocked(); err != nil {
+		return err
+	}
+	segments, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		return err
+	}
+	for _, segment := range segments {
+		batch := segment.batch
+		if batch.LogStreamID == response.LogStreamID && batch.FirstSeq >= response.AcceptedFrom && batch.LastSeq <= response.AcceptedTo {
+			if err := os.Remove(segment.path); err != nil && !os.IsNotExist(err) {
+				return fmt.Errorf("remove acknowledged log spool segment: %w", err)
+			}
+		}
+	}
+	return nil
+}
+
+func (spool *LogSpool) loadWatermarks() error {
+	body, err := os.ReadFile(spool.statePath)
+	if os.IsNotExist(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("read log watermarks: %w", err)
+	}
+	var state logSpoolState
+	if err := json.Unmarshal(body, &state); err != nil {
+		return fmt.Errorf("decode log watermarks: %w", err)
+	}
+	if state.Streams != nil {
+		spool.watermarks = state.Streams
+	}
+	return nil
+}
+
+func (spool LogSpool) persistWatermarksLocked() error {
+	body, err := json.Marshal(logSpoolState{Streams: spool.watermarks})
+	if err != nil {
+		return err
+	}
+	temporary := spool.statePath + ".tmp"
+	if err := os.WriteFile(temporary, body, 0o600); err != nil {
+		return fmt.Errorf("write log watermarks: %w", err)
+	}
+	if err := syncFile(temporary); err != nil {
+		_ = os.Remove(temporary)
+		return err
+	}
+	if err := os.Rename(temporary, spool.statePath); err != nil {
+		_ = os.Remove(temporary)
+		return fmt.Errorf("commit log watermarks: %w", err)
+	}
+	return nil
+}
+
+type LogBatchClient interface {
+	IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error)
+}
+
+type PermanentLogBatchError struct {
+	Reason string
+	Err    error
+}
+
+func (err PermanentLogBatchError) Error() string {
+	if err.Err == nil {
+		return "permanent log batch rejection: " + err.Reason
+	}
+	return "permanent log batch rejection: " + err.Reason + ": " + err.Err.Error()
+}
+
+func (err PermanentLogBatchError) Unwrap() error {
+	return err.Err
+}
+
+func PermanentLogBatchRejection(reason string, err error) error {
+	return PermanentLogBatchError{Reason: sanitizeSegmentName(reason), Err: err}
+}
+
+func (spool LogSpool) Flush(ctx context.Context, client LogBatchClient) (int, error) {
+	if client == nil {
+		return 0, fmt.Errorf("log batch client is required")
+	}
+	spool.mu.Lock()
+	pending, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		spool.mu.Unlock()
+		return 0, err
+	}
+	for _, segment := range pending {
+		spool.inflight[segment.path] = true
+	}
+	spool.mu.Unlock()
+	defer func() {
+		spool.mu.Lock()
+		defer spool.mu.Unlock()
+		for _, segment := range pending {
+			delete(spool.inflight, segment.path)
+		}
+	}()
+	acknowledged := 0
+	for _, segment := range pending {
+		batch := segment.batch.LogBatchIngestRequest
+		if err := ctx.Err(); err != nil {
+			return acknowledged, err
+		}
+		response, err := client.IngestLogBatch(ctx, batch)
+		if err != nil {
+			var permanent PermanentLogBatchError
+			if errors.As(err, &permanent) {
+				if rejectErr := spool.Reject(batch, permanent.Reason); rejectErr != nil {
+					return acknowledged, rejectErr
+				}
+				acknowledged++
+				continue
+			}
+			return acknowledged, err
+		}
+		if !response.Accepted || response.LogStreamID != batch.LogStreamID || response.AcceptedFrom > batch.FirstSeq || response.AcceptedTo < batch.LastSeq {
+			return acknowledged, fmt.Errorf("platform log acknowledgement does not cover pending batch")
+		}
+		if err := spool.Ack(response); err != nil {
+			return acknowledged, err
+		}
+		acknowledged++
+	}
+	return acknowledged, nil
+}
+
+func (spool LogSpool) Reject(batch protocol.LogBatchIngestRequest, reason string) error {
+	spool.mu.Lock()
+	defer spool.mu.Unlock()
+	segments, err := spool.pendingSegmentsLocked()
+	if err != nil {
+		return err
+	}
+	path := ""
+	for _, segment := range segments {
+		current := segment.batch
+		if current.LogStreamID == batch.LogStreamID && current.FirstSeq == batch.FirstSeq && current.LastSeq == batch.LastSeq && current.Checksum == batch.Checksum {
+			path = segment.path
+			break
+		}
+	}
+	if path == "" {
+		return nil
+	}
+	rejectedDir := filepath.Join(filepath.Dir(spool.dir), "logs-rejected")
+	if err := os.MkdirAll(rejectedDir, 0o755); err != nil {
+		return fmt.Errorf("create rejected log spool directory: %w", err)
+	}
+	rejectedPath := filepath.Join(rejectedDir, fmt.Sprintf("%s.%s", filepath.Base(path), sanitizeSegmentName(reason)))
+	if err := os.Rename(path, rejectedPath); err != nil {
+		return fmt.Errorf("move rejected log spool segment: %w", err)
+	}
+	return nil
+}
+
+func readDurableLogBatch(path string) (durableLogBatch, error) {
+	file, err := os.Open(path)
+	if err != nil {
+		return durableLogBatch{}, err
+	}
+	defer file.Close()
+	var batch durableLogBatch
+	if err := json.NewDecoder(file).Decode(&batch); err != nil {
+		return durableLogBatch{}, fmt.Errorf("decode log spool segment: %w", err)
+	}
+	return batch, nil
+}
+
+func syncFile(path string) error {
+	file, err := os.OpenFile(path, os.O_RDWR, 0)
+	if err != nil {
+		return fmt.Errorf("open spool file for sync: %w", err)
+	}
+	defer file.Close()
+	if err := file.Sync(); err != nil {
+		return fmt.Errorf("sync spool file: %w", err)
+	}
+	return nil
+}
+
+func (spool LogSpool) batchPath(batch durableLogBatch) string {
+	streamID := sanitizeSegmentName(batch.LogStreamID)
+	return filepath.Join(spool.dir, fmt.Sprintf("%s-%020d-%020d.json", streamID, batch.FirstSeq, batch.LastSeq))
+}
+
+func (spool LogSpool) watermarkPath(streamID string) string {
+	return filepath.Join(filepath.Dir(spool.dir), "log-watermarks", sanitizeSegmentName(streamID)+".json")
+}
+
+func (spool LogSpool) readWatermark(streamID string) (uint64, error) {
+	file, err := os.Open(spool.watermarkPath(streamID))
+	if err != nil {
+		return 0, err
+	}
+	defer file.Close()
+	var watermark struct {
+		LatestSeq uint64 `json:"latestSeq"`
+	}
+	if err := json.NewDecoder(file).Decode(&watermark); err != nil {
+		return 0, fmt.Errorf("decode log watermark: %w", err)
+	}
+	return watermark.LatestSeq, nil
+}
+
+func sanitizeSegmentName(value string) string {
+	var builder strings.Builder
+	for _, r := range value {
+		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' {
+			builder.WriteRune(r)
+			continue
+		}
+		builder.WriteByte('_')
+	}
+	if builder.Len() == 0 {
+		return "stream"
+	}
+	return builder.String()
+}

spool/log_spool_test.go

@@ -0,0 +1,290 @@
+package spool
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"browser.local/run/protocol"
+)
+
+func TestLogSpoolAggregatesContiguousEntriesForOneStream(t *testing.T) {
+	logSpool, err := NewLogSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
+	for sequence := uint64(1); sequence <= 3; sequence++ {
+		batch := validSpoolLogBatch(sequence, sequence)
+		if err := logSpool.EnqueueAggregated(batch, checksum); err != nil {
+			t.Fatalf("enqueue sequence %d: %v", sequence, err)
+		}
+	}
+	pending, err := logSpool.Pending()
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].LastSeq != 3 || len(pending[0].Entries) != 3 || pending[0].Checksum != "sha256:3" {
+		t.Fatalf("expected one aggregated batch, got %+v", pending)
+	}
+	if err := logSpool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 3}); err != nil {
+		t.Fatalf("ack aggregated batch: %v", err)
+	}
+	pending, err = logSpool.Pending()
+	if err != nil || len(pending) != 0 {
+		t.Fatalf("expected acknowledged aggregation removed, pending=%+v err=%v", pending, err)
+	}
+}
+
+func TestLogSpoolRetainsPendingAndRemovesAcknowledgedBatch(t *testing.T) {
+	spool, err := NewLogSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	first := validSpoolLogBatch(1, 2)
+	second := validSpoolLogBatch(3, 3)
+	if err := spool.Enqueue(first); err != nil {
+		t.Fatalf("enqueue first: %v", err)
+	}
+	if err := spool.Enqueue(second); err != nil {
+		t.Fatalf("enqueue second: %v", err)
+	}
+	pending, err := spool.Pending()
+	if err != nil {
+		t.Fatalf("pending before ack: %v", err)
+	}
+	if len(pending) != 2 {
+		t.Fatalf("expected two pending batches, got %+v", pending)
+	}
+
+	if err := spool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 2}); err != nil {
+		t.Fatalf("ack first: %v", err)
+	}
+	pending, err = spool.Pending()
+	if err != nil {
+		t.Fatalf("pending after ack: %v", err)
+	}
+	if len(pending) != 1 || pending[0].FirstSeq != 3 {
+		t.Fatalf("expected second batch pending, got %+v", pending)
+	}
+}
+
+func TestLogSpoolRetainsBatchWhenAckDoesNotCoverRange(t *testing.T) {
+	spool, err := NewLogSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	if err := spool.Enqueue(validSpoolLogBatch(1, 2)); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if err := spool.Ack(protocol.LogBatchIngestResponse{LogStreamID: "log-1", AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
+		t.Fatalf("partial ack: %v", err)
+	}
+	pending, err := spool.Pending()
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(pending) != 1 {
+		t.Fatalf("expected batch to remain pending, got %+v", pending)
+	}
+}
+
+func TestLogSpoolQuarantinesPermanentRejectedBatch(t *testing.T) {
+	root := t.TempDir()
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	if err := logSpool.Enqueue(validSpoolLogBatch(5, 5)); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	flushed, err := logSpool.Flush(context.Background(), permanentRejectLogBatchClient{})
+	if err != nil {
+		t.Fatalf("flush permanent rejection: %v", err)
+	}
+	if flushed != 1 {
+		t.Fatalf("expected one quarantined batch, got %d", flushed)
+	}
+	pending, err := logSpool.Pending()
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("expected no pending batches, got %+v", pending)
+	}
+	rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
+	if err != nil {
+		t.Fatalf("read rejected dir: %v", err)
+	}
+	if len(rejected) != 1 || !filepath.IsLocal(rejected[0].Name()) {
+		t.Fatalf("expected one local rejected file, got %+v", rejected)
+	}
+}
+
+func TestLogSpoolRestoresPendingAllocationWithoutWatermark(t *testing.T) {
+	root := t.TempDir()
+	first, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new first spool: %v", err)
+	}
+	batch := validSpoolLogBatch(9, 9)
+	batch.LogStreamID = "run.endpoint.server.stdout"
+	if err := first.Enqueue(batch); err != nil {
+		t.Fatalf("enqueue pending batch: %v", err)
+	}
+	if err := os.Remove(filepath.Join(root, "log-watermarks.json")); err != nil {
+		t.Fatalf("remove watermark state: %v", err)
+	}
+	restarted, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("restart spool: %v", err)
+	}
+	called := false
+	sequence, err := restarted.NextSequence(context.Background(), batch.LogStreamID, func(context.Context, string) (uint64, error) {
+		called = true
+		return 3, nil
+	})
+	if err != nil {
+		t.Fatalf("allocate after restart: %v", err)
+	}
+	if called || sequence != 10 {
+		t.Fatalf("expected pending watermark to allocate 10 without remote recovery, got sequence=%d remoteCalled=%t", sequence, called)
+	}
+}
+
+func TestLogSpoolSourceCursorDeduplicatesAcknowledgedReplayAfterRestart(t *testing.T) {
+	root := t.TempDir()
+	checksum := func(entries []protocol.LogEntry) (string, error) {
+		return fmt.Sprintf("sha256:%d:%s", len(entries), entries[len(entries)-1].Line), nil
+	}
+	first, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	batch := validSpoolLogBatch(0, 0)
+	batch.FirstSeq = 0
+	batch.LastSeq = 0
+	batch.Entries = []protocol.LogEntry{{Timestamp: time.Now().UTC(), Line: "first line"}}
+	sequence, appended, err := first.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 0, EndOffset: 11}, nil, checksum)
+	if err != nil || !appended || sequence != 1 {
+		t.Fatalf("append first cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
+	}
+	if err := first.Ack(protocol.LogBatchIngestResponse{LogStreamID: batch.LogStreamID, AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
+		t.Fatalf("ack first cursor: %v", err)
+	}
+	restarted, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("restart log spool: %v", err)
+	}
+	sequence, appended, err = restarted.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 0, EndOffset: 11}, nil, checksum)
+	if err != nil || appended || sequence != 1 {
+		t.Fatalf("deduplicate acknowledged cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
+	}
+	batch.Entries[0].Line = "second line"
+	sequence, appended, err = restarted.EnqueueNextAggregated(context.Background(), batch, &LogSourceCursor{StartOffset: 11, EndOffset: 23}, nil, checksum)
+	if err != nil || !appended || sequence != 2 {
+		t.Fatalf("append next cursor: sequence=%d appended=%t err=%v", sequence, appended, err)
+	}
+	pending, err := restarted.Pending()
+	if err != nil || len(pending) != 1 || pending[0].FirstSeq != 2 || pending[0].Entries[0].Line != "second line" {
+		t.Fatalf("unexpected pending cursor batches: pending=%+v err=%v", pending, err)
+	}
+}
+
+func TestLogSpoolAggregatedSegmentIsReplacedInPlace(t *testing.T) {
+	root := t.TempDir()
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
+	if err := logSpool.EnqueueAggregated(validSpoolLogBatch(1, 1), checksum); err != nil {
+		t.Fatalf("enqueue first segment: %v", err)
+	}
+	before, err := os.ReadDir(filepath.Join(root, "logs"))
+	if err != nil || len(before) != 1 {
+		t.Fatalf("read first segment: entries=%+v err=%v", before, err)
+	}
+	if err := logSpool.EnqueueAggregated(validSpoolLogBatch(2, 2), checksum); err != nil {
+		t.Fatalf("aggregate second segment: %v", err)
+	}
+	after, err := os.ReadDir(filepath.Join(root, "logs"))
+	if err != nil || len(after) != 1 || after[0].Name() != before[0].Name() {
+		t.Fatalf("aggregation did not replace one stable path: before=%+v after=%+v err=%v", before, after, err)
+	}
+	pending, err := logSpool.Pending()
+	if err != nil || len(pending) != 1 || pending[0].FirstSeq != 1 || pending[0].LastSeq != 2 {
+		t.Fatalf("unexpected aggregate after replacement: pending=%+v err=%v", pending, err)
+	}
+}
+
+func TestLogSpoolDoesNotExtendInflightAggregate(t *testing.T) {
+	logSpool, err := NewLogSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("new log spool: %v", err)
+	}
+	checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil }
+	if err := logSpool.EnqueueAggregated(validSpoolLogBatch(1, 1), checksum); err != nil {
+		t.Fatalf("enqueue first segment: %v", err)
+	}
+	client := &blockingLogBatchClient{started: make(chan struct{}), release: make(chan struct{})}
+	done := make(chan error, 1)
+	go func() {
+		_, err := logSpool.Flush(context.Background(), client)
+		done <- err
+	}()
+	<-client.started
+	if err := logSpool.EnqueueAggregated(validSpoolLogBatch(2, 2), checksum); err != nil {
+		t.Fatalf("enqueue while first segment is inflight: %v", err)
+	}
+	close(client.release)
+	if err := <-done; err != nil {
+		t.Fatalf("flush inflight segment: %v", err)
+	}
+	pending, err := logSpool.Pending()
+	if err != nil || len(pending) != 1 || pending[0].FirstSeq != 2 || pending[0].LastSeq != 2 {
+		t.Fatalf("inflight segment was extended or next segment lost: pending=%+v err=%v", pending, err)
+	}
+}
+
+type blockingLogBatchClient struct {
+	once    sync.Once
+	started chan struct{}
+	release chan struct{}
+}
+
+func (client *blockingLogBatchClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
+	client.once.Do(func() { close(client.started) })
+	<-client.release
+	return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
+}
+
+type permanentRejectLogBatchClient struct{}
+
+func (permanentRejectLogBatchClient) IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
+	return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_sequence_gap", nil)
+}
+
+func validSpoolLogBatch(firstSeq uint64, lastSeq uint64) protocol.LogBatchIngestRequest {
+	entries := make([]protocol.LogEntry, 0, lastSeq-firstSeq+1)
+	for seq := firstSeq; seq <= lastSeq; seq++ {
+		entries = append(entries, protocol.LogEntry{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, int(seq), 0, time.UTC), Level: "info", Line: "line"})
+	}
+	return protocol.LogBatchIngestRequest{
+		RunEndpointID:    "run-local",
+		SessionToken:     "session-token",
+		LogStreamID:      "log-1",
+		ServerInstanceID: "server-1",
+		StreamKey:        "stdout",
+		Source:           "process",
+		FirstSeq:         firstSeq,
+		LastSeq:          lastSeq,
+		Compression:      "none",
+		Checksum:         "sha256:test",
+		Entries:          entries,
+	}
+}