feat(events): follow mode via ring sequence cursor, plus demo-watch console

cmd/marvel/main.go

@@ -381,6 +381,7 @@ func eventsCmd() *cobra.Command {
 	var n int
 	var workspace, team, role, session, kind string
 	var warningsOnly bool
+	var follow bool
 	cmd := &cobra.Command{
 		Use:   "events",
 		Short: "List recent session/team state-transition events",
@@ -409,60 +410,110 @@ Examples:
   marvel events --kind context.limit-unresolved  # why a CTX% cell is blank
   marvel events --kind admission.refused     # spawns a team budget refused
   marvel events --warnings                   # only warning-severity events
+  marvel events --follow                     # live tail; poll the ring every second
   marvel --cluster desk events               # remote daemon via mrvl://`,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			params := map[string]any{"n": n}
-			if workspace != "" {
-				params["workspace"] = workspace
-			}
-			if team != "" {
-				params["team"] = team
-			}
-			if role != "" {
-				params["role"] = role
+			buildParams := func(sinceSeq uint64) json.RawMessage {
+				params := map[string]any{"n": n}
+				if sinceSeq > 0 {
+					// Cursor requests return every new event, not a tail.
+					params["n"] = 0
+					params["since_seq"] = sinceSeq
+				}
+				if workspace != "" {
+					params["workspace"] = workspace
+				}
+				if team != "" {
+					params["team"] = team
+				}
+				if role != "" {
+					params["role"] = role
+				}
+				if session != "" {
+					params["session"] = session
+				}
+				if kind != "" {
+					params["kind"] = kind
+				}
+				if warningsOnly {
+					params["min_severity"] = "warning"
+				}
+				raw, _ := json.Marshal(params)
+				return raw
 			}
-			if session != "" {
-				params["session"] = session
+			fetch := func(sinceSeq uint64) ([]events.Event, error) {
+				resp, err := send(daemon.Request{Method: "events", Params: buildParams(sinceSeq)})
+				if err != nil {
+					return nil, err
+				}
+				if resp.Error != "" {
+					return nil, fmt.Errorf("%s", resp.Error)
+				}
+				var result struct {
+					Events []events.Event `json:"events"`
+				}
+				if err := json.Unmarshal(resp.Result, &result); err != nil {
+					return nil, fmt.Errorf("parse events: %w", err)
+				}
+				return result.Events, nil
 			}
-			if kind != "" {
-				params["kind"] = kind
+			printBatch := func(evs []events.Event, header bool) {
+				tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
+				if header {
+					_, _ = fmt.Fprintln(tw, "TIME\tSEV\tKIND\tSESSION\tMESSAGE")
+				}
+				for _, ev := range evs {
+					sev := string(ev.Severity)
+					if sev == "" {
+						sev = "info"
+					}
+					sessRef := ev.Session
+					if sessRef == "" && ev.Team != "" {
+						sessRef = ev.Workspace + "/" + ev.Team
+					}
+					_, _ = fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\n",
+						ev.Timestamp.Format("15:04:05"), sev, ev.Kind, sessRef, ev.Message)
+				}
+				_ = tw.Flush()
 			}
-			if warningsOnly {
-				params["min_severity"] = "warning"
+			maxSeq := func(evs []events.Event, cur uint64) uint64 {
+				for _, ev := range evs {
+					if ev.Seq > cur {
+						cur = ev.Seq
+					}
+				}
+				return cur
 			}
-			raw, _ := json.Marshal(params)
-			resp, err := send(daemon.Request{Method: "events", Params: raw})
+
+			evs, err := fetch(0)
 			if err != nil {
 				return err
 			}
-			if resp.Error != "" {
-				return fmt.Errorf("%s", resp.Error)
-			}
-			var result struct {
-				Events []events.Event `json:"events"`
-			}
-			if err := json.Unmarshal(resp.Result, &result); err != nil {
-				return fmt.Errorf("parse events: %w", err)
-			}
-			if len(result.Events) == 0 {
+			if len(evs) == 0 && !follow {
 				fmt.Println("no events")
 				return nil
 			}
-			tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
-			_, _ = fmt.Fprintln(tw, "TIME\tSEV\tKIND\tSESSION\tMESSAGE")
-			for _, ev := range result.Events {
-				sev := string(ev.Severity)
-				if sev == "" {
-					sev = "info"
+			printBatch(evs, true)
+			if !follow {
+				return nil
+			}
+			// Follow mode: poll the ring with a Seq cursor so each event
+			// prints exactly once, in order, until interrupted. The ring
+			// assigns Seq monotonically, so a cursor survives ring
+			// wraparound (missed events are simply gone, never repeated).
+			cursor := maxSeq(evs, 0)
+			for {
+				time.Sleep(time.Second)
+				batch, err := fetch(cursor)
+				if err != nil {
+					return err
 				}
-				sessRef := ev.Session
-				if sessRef == "" && ev.Team != "" {
-					sessRef = ev.Workspace + "/" + ev.Team
+				if len(batch) == 0 {
+					continue
 				}
-				_, _ = fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\n",
-					ev.Timestamp.Format("15:04:05"), sev, ev.Kind, sessRef, ev.Message)
+				printBatch(batch, false)
+				cursor = maxSeq(batch, cursor)
 			}
-			return tw.Flush()
 		},
 	}
 	cmd.Flags().IntVarP(&n, "lines", "n", 100, "number of events to return (0 = all buffered)")
@@ -472,6 +523,7 @@ Examples:
 	cmd.Flags().StringVar(&session, "session", "", "filter by session key (workspace/name)")
 	cmd.Flags().StringVar(&kind, "kind", "", "filter by event kind (e.g. session.crashed, health.failed, agent.tool.call)")
 	cmd.Flags().BoolVar(&warningsOnly, "warnings", false, "show only warning-severity events")
+	cmd.Flags().BoolVarP(&follow, "follow", "f", false, "poll for new events every second until interrupted")
 	return cmd
 }
 

docs/demo.md

@@ -35,6 +35,20 @@ rm -f ~/.marvel/state/marvel.bolt   # forget persisted resources
 `marvel events` reads a bounded in-memory ring, so it is empty on a fresh
 daemon and fills as the act runs.
 
+## Watching live
+
+The acts read better watched than replayed. Two tools:
+
+- `marvel events --follow` (`-f`) tails the ring: it prints the current
+  tail, then polls once a second and prints each new event exactly once,
+  in order, using a ring-assigned sequence cursor. All the usual filters
+  compose with it (`--workspace`, `--kind`, `--warnings`).
+- `just demo-watch` builds a four-pane tmux operator console (driver
+  shell, live session table, live event tail, daemon log poll). Attach
+  with `tmux attach -t marvel-watch`, drive the beats from the top-left
+  pane, and watch states flip in real time. The panes poll until a
+  daemon appears, so start it in whichever order you like.
+
 ---
 
 ## Act 1 — Recover

internal/daemon/daemon.go

@@ -609,6 +609,9 @@ type eventsParams struct {
 	Session     string `json:"session,omitempty"`
 	Kind        string `json:"kind,omitempty"`
 	MinSeverity string `json:"min_severity,omitempty"` // "" or "warning"
+	// SinceSeq returns only events with Seq strictly greater than this
+	// value — the follow-mode resume cursor. Zero means no cursor.
+	SinceSeq uint64 `json:"since_seq,omitempty"`
 }
 
 type eventsResult struct {
@@ -629,6 +632,7 @@ func (d *Daemon) handleEvents(params json.RawMessage) Response {
 		Session:     p.Session,
 		Kind:        events.Kind(p.Kind),
 		MinSeverity: events.Severity(p.MinSeverity),
+		SinceSeq:    p.SinceSeq,
 	}
 	snap := d.events.Snapshot(f, p.N)
 	data, err := json.Marshal(eventsResult{Events: snap})

internal/events/events.go

@@ -97,6 +97,12 @@ const (
 
 // Event is one structured state-transition record.
 type Event struct {
+	// Seq is a ring-assigned monotonic sequence number, starting at 1.
+	// It exists so poll-based consumers (`marvel events --follow`) can
+	// resume from the last event they saw instead of deduplicating on
+	// timestamps, which are not unique. Producers never set it; the
+	// ring assigns it under its own lock at Emit time.
+	Seq       uint64    `json:"seq,omitempty"`
 	Timestamp time.Time `json:"ts"`
 	Kind      Kind      `json:"kind"`
 	Severity  Severity  `json:"severity"`
@@ -147,6 +153,7 @@ type Ring struct {
 	buf      []Event
 	head     int // index of the oldest event when len(buf) == capacity
 	full     bool
+	nextSeq  uint64 // next Seq to assign; monotonic for the ring's lifetime
 }
 
 // DefaultCapacity is the ring size used when NewRing is called with
@@ -164,6 +171,7 @@ func NewRing(capacity int) *Ring {
 	return &Ring{
 		capacity: capacity,
 		buf:      make([]Event, 0, capacity),
+		nextSeq:  1,
 	}
 }
 
@@ -178,6 +186,8 @@ func (r *Ring) Emit(ev Event) {
 	if ev.Severity == "" {
 		ev.Severity = SeverityInfo
 	}
+	ev.Seq = r.nextSeq
+	r.nextSeq++
 	if !r.full {
 		r.buf = append(r.buf, ev)
 		if len(r.buf) == r.capacity {
@@ -199,6 +209,10 @@ type Filter struct {
 	Session     string
 	Kind        Kind
 	MinSeverity Severity
+	// SinceSeq, when nonzero, matches only events with Seq strictly
+	// greater than this value — the resume cursor for follow-mode
+	// polling.
+	SinceSeq uint64
 }
 
 // Snapshot returns up to `n` most recent events matching f, oldest-first
@@ -261,5 +275,8 @@ func matches(ev Event, f Filter) bool {
 	if f.MinSeverity == SeverityWarning && ev.Severity != SeverityWarning {
 		return false
 	}
+	if f.SinceSeq > 0 && ev.Seq <= f.SinceSeq {
+		return false
+	}
 	return true
 }

internal/events/events_test.go

@@ -128,3 +128,37 @@ func TestRingConcurrentEmit(t *testing.T) {
 		t.Fatal("expected some events after concurrent emit")
 	}
 }
+
+func TestRingAssignsMonotonicSeq(t *testing.T) {
+	r := NewRing(3)
+	for i := 0; i < 5; i++ {
+		r.Emit(Event{Kind: KindSessionCreated, Message: string(rune('a' + i))})
+	}
+	snap := r.Snapshot(Filter{}, 0)
+	// Survivors after wraparound are c,d,e with seqs 3,4,5 — Seq is
+	// assigned per Emit and never reused, so it survives overwrites.
+	want := uint64(3)
+	for _, ev := range snap {
+		if ev.Seq != want {
+			t.Fatalf("Seq=%d, want %d (message %q)", ev.Seq, want, ev.Message)
+		}
+		want++
+	}
+}
+
+func TestRingFilterSinceSeq(t *testing.T) {
+	r := NewRing(10)
+	for i := 0; i < 5; i++ {
+		r.Emit(Event{Kind: KindSessionCreated, Message: string(rune('a' + i))})
+	}
+	snap := r.Snapshot(Filter{SinceSeq: 3}, 0)
+	if len(snap) != 2 {
+		t.Fatalf("Snapshot len=%d, want 2 (strictly greater than cursor)", len(snap))
+	}
+	if snap[0].Seq != 4 || snap[1].Seq != 5 {
+		t.Fatalf("expected seqs 4,5 — got %d,%d", snap[0].Seq, snap[1].Seq)
+	}
+	if got := r.Snapshot(Filter{SinceSeq: 5}, 0); len(got) != 0 {
+		t.Fatalf("cursor at newest should return nothing, got %d", len(got))
+	}
+}

justfile

@@ -183,3 +183,24 @@ demo-all:
     @echo "  just demo-act3    # Control plane"
     @echo ""
     @echo "Between acts: just stop && rm -f ~/.marvel/state/marvel.bolt && just start-bg"
+
+# Operator console for watching a demo live: four panes in one tmux
+# session — driver shell, live session table, live event tail, daemon
+# log poll. Attach from your own terminal: tmux attach -t marvel-watch
+demo-watch: build
+    #!/usr/bin/env bash
+    set -euo pipefail
+    tmux kill-session -t marvel-watch 2>/dev/null || true
+    tmux new-session -d -s marvel-watch -x 220 -y 55
+    P0=$(tmux display -p -t marvel-watch '#{pane_id}')
+    P1=$(tmux split-window -t "$P0" -v -P -F '#{pane_id}')
+    P2=$(tmux split-window -t "$P0" -h -P -F '#{pane_id}')
+    P3=$(tmux split-window -t "$P1" -h -P -F '#{pane_id}')
+    tmux send-keys -t "$P0" 'clear; echo "DRIVER — run demo beats here (runbook: docs/demo.md). Daemon: ./bin/marvel daemon &"' C-m
+    tmux send-keys -t "$P2" 'while :; do ./bin/marvel get sessions -w 1 2>/dev/null; sleep 2; clear; done' C-m
+    tmux send-keys -t "$P1" 'while :; do ./bin/marvel events --follow 2>/dev/null; sleep 2; done' C-m
+    tmux send-keys -t "$P3" 'while :; do clear; ./bin/marvel daemon logs -n 14 2>/dev/null || echo "(daemon not up yet)"; sleep 2; done' C-m
+    tmux select-pane -t "$P0"
+    echo "Operator console ready: tmux attach -t marvel-watch"
+    echo "  top-left  DRIVER          top-right  sessions (live)"
+    echo "  bot-left  events --follow bot-right  daemon logs"