fix(remote): share the follow cursor between remote logs and the fleet dashboard

cmd/spinloop/remote_logs.go

@@ -22,12 +22,6 @@ var logsFetchFn = remote.FetchLogs
 // on it.
 var logsFollowInterval = 5 * time.Second
 
-// logsFollowOverlap is how far back of the last event a follow re-asks from.
-// The shipping agent's delivery lag means an event can land with a timestamp
-// slightly behind one already returned, so each poll deliberately re-reads a
-// little and suppresses what it has already printed by event id.
-const logsFollowOverlap = 10 * time.Second
-
 // cmdRemoteLogs prints the logs an environment's instances shipped to
 // CloudWatch. It reads the durable store rather than the instance, so a boot
 // that failed and an instance that has since terminated are both still
@@ -193,8 +187,7 @@ func followLogs(cfg remote.Config, q remote.LogQuery, format string) error {
 // caller so it can be driven directly.
 func followLogsLoop(ctx context.Context, cfg remote.Config, q remote.LogQuery,
 	format string, w io.Writer) error {
-	printed := map[string]time.Time{}
-	newest := time.Time{}
+	cursor := remote.NewFollowCursor(remote.FollowOverlap)
 	// Labelling is decided across the whole session, not per batch: a poll that
 	// happened to return one instance's lines must not drop the prefix the
 	// previous poll's lines carried. Once a second origin appears the output
@@ -214,17 +207,9 @@ func followLogsLoop(ctx context.Context, cfg remote.Config, q remote.LogQuery,
 		if first && res.Omitted > 0 && format != "json" {
 			fmt.Fprintf(w, "... %d earlier events omitted (raise --limit to see more)\n", res.Omitted)
 		}
-		fresh := make([]remote.LogEvent, 0, len(res.Events))
-		for _, e := range res.Events {
-			if _, seen := printed[e.ID]; seen {
-				continue
-			}
-			printed[e.ID] = e.Timestamp
+		fresh := cursor.Advance(res.Events)
+		for _, e := range fresh {
 			origins[e.Source+"/"+e.Instance] = true
-			fresh = append(fresh, e)
-			if e.Timestamp.After(newest) {
-				newest = e.Timestamp
-			}
 		}
 		if len(fresh) > 0 {
 			if format == "json" {
@@ -235,15 +220,8 @@ func followLogsLoop(ctx context.Context, cfg remote.Config, q remote.LogQuery,
 				writeLogsText(w, fresh, len(origins) > 1)
 			}
 		}
-		// Only the overlap window can produce a duplicate, so ids older than
-		// it are forgotten and the map cannot grow with the session.
-		for id, ts := range printed {
-			if ts.Before(newest.Add(-2 * logsFollowOverlap)) {
-				delete(printed, id)
-			}
-		}
-		if !newest.IsZero() {
-			q.Start = newest.Add(-logsFollowOverlap)
+		if start := cursor.Start(); !start.IsZero() {
+			q.Start = start
 		}
 		select {
 		case <-ctx.Done():

internal/fleet/remote_node.go

@@ -23,6 +23,12 @@ import (
 type remoteNode struct {
 	name string
 	cfg  remote.Config
+	// logs holds the position a follow of this node's engine log has reached.
+	// It is the same cursor `spinloop remote logs -f` uses, and for the same
+	// reason: CloudWatch has no resumable read position of its own, so a poll
+	// re-asks a little behind the newest event already seen and this
+	// suppresses what the overlap re-reads, by event id.
+	logs *remote.FollowCursor
 }
 
 // NewRemoteNode builds the live node for a named remote environment. The config
@@ -35,7 +41,7 @@ func NewRemoteNode(name string, cfg remote.Config) (Node, error) {
 			"remote environment %q is not fully configured: start_url, stop_url and region are all required",
 			name)
 	}
-	return &remoteNode{name: name, cfg: cfg}, nil
+	return &remoteNode{name: name, cfg: cfg, logs: remote.NewFollowCursor(remote.FollowOverlap)}, nil
 }
 
 func (n *remoteNode) Name() string { return n.name }
@@ -96,18 +102,28 @@ func (n *remoteNode) Stop(ctx context.Context) (daemon.StatusResponse, error) {
 const remoteEngineTail = 1000
 
 func (n *remoteNode) Logs(ctx context.Context, offset int64, limit int) (daemon.LogsResponse, error) {
-	// The offset is accepted to satisfy the node contract, but remote logs are a
-	// tail of the log store, not a position to resume from, so it is not a cursor.
-	_ = offset
+	// daemon.TailLog means a fresh open of the view: start the cursor over,
+	// so this open shows its own tail rather than having it suppressed as
+	// already seen by whatever this node last followed.
+	if offset == daemon.TailLog {
+		n.logs.Reset()
+	}
+	start := n.logs.Start()
 	res, err := remote.FetchLogs(ctx, n.cfg, remote.LogQuery{
 		Environment: n.cfg.Environment,
 		Source:      remote.LogSourceEngine,
 		Limit:       remoteEngineTail,
+		Start:       start,
 	})
 	if err != nil {
 		return daemon.LogsResponse{}, err
 	}
-	return logsFromRemote(res), nil
+	fresh := n.logs.Advance(res.Events)
+	// Missing only on the read that had no lower bound yet finding nothing:
+	// that is genuinely no log, ever. A later poll with nothing new is a
+	// quiet log, not a missing one.
+	missing := start.IsZero() && len(res.Events) == 0
+	return logsFromRemote(fresh, missing), nil
 }
 
 // statusFromRemote maps the control plane's status reply onto the node's status.
@@ -142,22 +158,31 @@ func statsFromRemote(resp remote.StatsResponse) metrics.Stats {
 	}
 }
 
-// logsFromRemote maps a fetched log tail onto the node's log reply. Events arrive
-// oldest first, so the content reads top to bottom. An empty tail is reported as
-// a missing log rather than an empty one: that is the state a reader can act on
-// (the engine has not run here, or has sent nothing).
-func logsFromRemote(res remote.LogResult) daemon.LogsResponse {
-	if len(res.Events) == 0 {
-		return daemon.LogsResponse{Missing: true}
+// logsFromRemote maps one poll's fresh events — the ones remoteNode.Logs's
+// FollowCursor has not already returned — onto the node's log reply. Events
+// arrive oldest first, so the content reads top to bottom. missing reports
+// that this was a from-the-beginning read that found nothing: the state a
+// reader can act on (the engine has not run here, or has sent nothing),
+// distinct from a later poll simply having nothing new to add. NextOffset
+// only needs to say "not a fresh open" to the next call — the real position
+// lives in the node's own cursor — so it carries the newest event shown for
+// whoever finds that useful to see, and 0 has the same effect when there is
+// none.
+func logsFromRemote(fresh []remote.LogEvent, missing bool) daemon.LogsResponse {
+	if len(fresh) == 0 {
+		return daemon.LogsResponse{Missing: missing}
 	}
-	msgs := make([]string, len(res.Events))
-	for i, e := range res.Events {
+	msgs := make([]string, len(fresh))
+	for i, e := range fresh {
 		msgs[i] = e.Message
 	}
-	content := strings.Join(msgs, "\n")
+	// Each event is already one complete, discrete line — unlike a byte-stream
+	// tail, there is never a trailing partial line to leave unterminated — so
+	// the join ends in a newline the same way a local engine log's lines do.
+	content := strings.Join(msgs, "\n") + "\n"
 	return daemon.LogsResponse{
 		Content:    content,
-		NextOffset: int64(len(content)),
+		NextOffset: fresh[len(fresh)-1].Timestamp.UnixMilli(),
 		Size:       int64(len(content)),
 	}
 }

internal/fleet/remote_node_test.go

@@ -2,13 +2,16 @@ package fleet
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/spinloop-ai/spinloop/internal/daemon"
 	"github.com/spinloop-ai/spinloop/internal/metrics"
@@ -89,18 +92,106 @@ func TestStatsFromRemote(t *testing.T) {
 }
 
 func TestLogsFromRemote(t *testing.T) {
-	if got := logsFromRemote(remote.LogResult{}); !got.Missing {
-		t.Errorf("an empty tail should be reported as a missing log, got %+v", got)
+	if got := logsFromRemote(nil, true); !got.Missing {
+		t.Errorf("no fresh events on a from-the-beginning read should be reported missing, got %+v", got)
 	}
-	got := logsFromRemote(remote.LogResult{Events: []remote.LogEvent{
-		{Message: "loading model"}, {Message: "server ready"},
-	}})
-	want := "loading model\nserver ready"
+	// No fresh events on a later poll — the common steady state once a follow
+	// is caught up — must not be reported missing: the log is not gone, it is
+	// just quiet right now.
+	if got := logsFromRemote(nil, false); got.Missing {
+		t.Errorf("no fresh events on a later poll should not be reported missing, got %+v", got)
+	}
+
+	t1 := time.UnixMilli(1000)
+	t2 := time.UnixMilli(2000)
+	got := logsFromRemote([]remote.LogEvent{
+		{Message: "loading model", Timestamp: t1}, {Message: "server ready", Timestamp: t2},
+	}, false)
+	want := "loading model\nserver ready\n"
 	if got.Content != want {
 		t.Errorf("content = %q, want %q", got.Content, want)
 	}
-	if got.NextOffset != int64(len(want)) || got.Size != int64(len(want)) {
-		t.Errorf("offset/size = %d/%d, want %d", got.NextOffset, got.Size, len(want))
+	if got.NextOffset != t2.UnixMilli() {
+		t.Errorf("nextOffset = %d, want the newest event's millisecond %d", got.NextOffset, t2.UnixMilli())
+	}
+	if got.Size != int64(len(want)) {
+		t.Errorf("size = %d, want %d", got.Size, len(want))
+	}
+}
+
+// A remote node's log follow uses the exact same cursor as `spinloop remote
+// logs -f` (remote.FollowCursor, seeded with remote.FollowOverlap) — not a
+// lookalike reimplementation — so a second poll bounds its query behind the
+// newest event already shown by the same overlap window, and does not show
+// that event again. The bug this guards is the tail being replayed in full
+// on every poll regardless of what was already shown.
+func TestRemoteNodeLogsSharesTheFollowCursorWithRemoteLogsCommand(t *testing.T) {
+	stubAWSCreds(t)
+	eventMs := time.Date(2026, 8, 9, 11, 30, 0, 0, time.UTC).UnixMilli()
+
+	var mu sync.Mutex
+	served := map[string]bool{}
+	var starts []*int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		var in struct {
+			LogGroupName string `json:"logGroupName"`
+			StartTime    *int64 `json:"startTime"`
+		}
+		json.Unmarshal(body, &in)
+
+		mu.Lock()
+		starts = append(starts, in.StartTime)
+		first := !served[in.LogGroupName]
+		served[in.LogGroupName] = true
+		mu.Unlock()
+
+		w.Header().Set("Content-Type", "application/x-amz-json-1.1")
+		if first {
+			// Each group ships the same one event, the first time it is asked —
+			// the fixture's stand-in for "the engine's log group has this".
+			fmt.Fprintf(w, `{"events":[{"eventId":"e1","timestamp":%d,"logStreamName":"env-1/i-1","message":"hello"}]}`, eventMs)
+			return
+		}
+		w.Write([]byte(`{"events":[]}`))
+	}))
+	t.Cleanup(srv.Close)
+	t.Setenv("AWS_ENDPOINT_URL_CLOUDWATCH_LOGS", srv.URL)
+	cfg := remote.Config{StartURL: "http://x", StopURL: "http://x", Environment: "env-1", Region: "us-east-1"}
+	node, err := NewRemoteNode("env", cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp1, err := node.Logs(context.Background(), daemon.TailLog, 0)
+	if err != nil {
+		t.Fatalf("tail poll: %v", err)
+	}
+	if !strings.Contains(resp1.Content, "hello") {
+		t.Fatalf("the tail poll should show the event, got %+v", resp1)
+	}
+	for _, s := range starts {
+		if s != nil {
+			t.Errorf("the tail poll (offset TailLog) sent a start bound: %v", *s)
+		}
+	}
+
+	starts = nil
+	resp2, err := node.Logs(context.Background(), 0, 0)
+	if err != nil {
+		t.Fatalf("follow-up poll: %v", err)
+	}
+	if resp2.Content != "" || resp2.Missing {
+		t.Errorf("the follow-up poll should show nothing new and not report missing, got %+v", resp2)
+	}
+	wantStart := eventMs - remote.FollowOverlap.Milliseconds()
+	for _, s := range starts {
+		if s == nil {
+			t.Fatal("the follow-up poll sent no start bound")
+		}
+		if *s != wantStart {
+			t.Errorf("start bound = %d, want the shared overlap behind the last event (%d)", *s, wantStart)
+		}
 	}
 }
 

internal/remote/follow.go

@@ -0,0 +1,85 @@
+package remote
+
+import (
+	"sync"
+	"time"
+)
+
+// FollowOverlap is how far back of the newest event already seen a follow
+// re-asks from by default. The shipping agent's delivery lag means an event
+// can land with a timestamp slightly behind one already returned, so a poll
+// deliberately re-reads a little; FollowCursor suppresses what it has
+// already returned, by event id. Both `spinloop remote logs -f` and a fleet
+// node's own log follow share this constant, so a fleet-dashboard poll and a
+// standalone follow tolerate the same shipping lag.
+const FollowOverlap = 10 * time.Second
+
+// FollowCursor turns repeated LogResult polls into a stream of events shown
+// exactly once. CloudWatch offers no resumable read position — Start is a
+// lower bound on a query, not a cursor a store can resume from — so a follow
+// has to re-ask a little behind the newest event it has already seen, to
+// catch anything the shipping agent delivered late, and then suppress by
+// event id whatever that overlap re-reads. One cursor holds that state for
+// the life of one follow, whether that follow is `spinloop remote logs -f`
+// polling in a loop or a fleet node answering repeated Logs calls.
+type FollowCursor struct {
+	overlap time.Duration
+
+	mu      sync.Mutex
+	printed map[string]time.Time
+	newest  time.Time
+}
+
+// NewFollowCursor starts an empty cursor that re-asks overlap behind the
+// newest event seen on every poll.
+func NewFollowCursor(overlap time.Duration) *FollowCursor {
+	return &FollowCursor{overlap: overlap, printed: map[string]time.Time{}}
+}
+
+// Advance filters events down to the ones this cursor has not already
+// returned, in the order given, and folds them into its state so a later
+// poll that re-reads the overlap does not return them again.
+func (c *FollowCursor) Advance(events []LogEvent) []LogEvent {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	fresh := make([]LogEvent, 0, len(events))
+	for _, e := range events {
+		if _, seen := c.printed[e.ID]; seen {
+			continue
+		}
+		c.printed[e.ID] = e.Timestamp
+		fresh = append(fresh, e)
+		if e.Timestamp.After(c.newest) {
+			c.newest = e.Timestamp
+		}
+	}
+	// Only ids within the overlap window can ever be re-read, so anything
+	// older is forgotten and the map cannot grow across a long follow.
+	for id, ts := range c.printed {
+		if ts.Before(c.newest.Add(-2 * c.overlap)) {
+			delete(c.printed, id)
+		}
+	}
+	return fresh
+}
+
+// Start is the query bound the next poll should use: zero until an event has
+// been seen, then the overlap window behind the newest one.
+func (c *FollowCursor) Start() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.newest.IsZero() {
+		return time.Time{}
+	}
+	return c.newest.Add(-c.overlap)
+}
+
+// Reset drops everything the cursor has seen, as a follow restarted from the
+// tail rather than continued needs: a fresh open should show its own tail
+// again, not have it suppressed as already seen by a previous session.
+func (c *FollowCursor) Reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.printed = map[string]time.Time{}
+	c.newest = time.Time{}
+}