diff --git a/cmd/marvel/main.go b/cmd/marvel/main.go index 7209145..41afa66 100644 --- a/cmd/marvel/main.go +++ b/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 } diff --git a/docs/demo.md b/docs/demo.md index 4f602d8..1c4ce9a 100644 --- a/docs/demo.md +++ b/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 diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 2008a05..bba8dd6 100644 --- a/internal/daemon/daemon.go +++ b/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}) diff --git a/internal/events/events.go b/internal/events/events.go index 0e129c6..cb4abab 100644 --- a/internal/events/events.go +++ b/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 } diff --git a/internal/events/events_test.go b/internal/events/events_test.go index 4e3240f..40c876e 100644 --- a/internal/events/events_test.go +++ b/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)) + } +} diff --git a/justfile b/justfile index 458b3bd..3ca41eb 100644 --- a/justfile +++ b/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"