add the subscription store and polling loop

internal/notifications holds the per-session subscription set and the
poller behind review notifications (ADR 2.2). The store maps change
numbers to cursors -- the last-seen updated timestamp -- and advances a
cursor in one atomic compare-and-move, so a poll result racing an
unsubscribe cannot resurrect the subscription.

The poller batches every subscribed change into one change:A OR
change:B query per tick and skips the network entirely when nothing is
subscribed. Failures are logged to stderr and retried next tick: a
background loop has no caller to return errors to, and a transient
Gerrit failure must not end the session's subscriptions. Emission goes
through the Emitter seam; the channel-backed implementation follows in
a separate change.

The tracer payload is the fact of movement -- a bare review_activity
element; richer activity extraction arrives with the deltas phase.

Refs: #51

Author: xobotyi

internal/notifications/poller.go

@@ -0,0 +1,122 @@
+package notifications
+
+import (
+	"context"
+	"log/slog"
+	"strconv"
+	"strings"
+	"time"
+
+	gerrit "github.com/andygrunwald/go-gerrit"
+
+	"dev.gaijin.team/go/go-gerrit-mcp/internal/gerritclient"
+	"dev.gaijin.team/go/go-gerrit-mcp/internal/llmxml"
+)
+
+// Emitter delivers one rendered notification into the agent's session.
+// Content is a rendered llmxml payload; meta carries routing context that
+// becomes tag attributes on the injected block, so keys must be limited to
+// letters, digits, and underscores.
+type Emitter interface {
+	Emit(ctx context.Context, content string, meta map[string]string) error
+}
+
+// Poller periodically queries Gerrit for movement on subscribed changes and
+// hands detected activity to the emitter. Failures are logged and retried on
+// the next tick — a background loop has no caller to return errors to, and a
+// transient Gerrit failure must not end the session's subscriptions.
+type Poller struct {
+	store    *Store
+	client   *gerritclient.Client
+	emitter  Emitter
+	interval time.Duration
+	lgr      *slog.Logger
+}
+
+// NewPoller assembles a poller over the given subscription store. It does not
+// start anything; the caller runs [Poller.Run] on its own goroutine.
+func NewPoller(
+	store *Store, client *gerritclient.Client, emitter Emitter, interval time.Duration, lgr *slog.Logger,
+) *Poller {
+	return &Poller{
+		store:    store,
+		client:   client,
+		emitter:  emitter,
+		interval: interval,
+		lgr:      lgr,
+	}
+}
+
+// Run polls at the configured interval until ctx is cancelled. It blocks;
+// the caller owns the goroutine.
+func (p *Poller) Run(ctx context.Context) {
+	ticker := time.NewTicker(p.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			p.tick(ctx)
+		}
+	}
+}
+
+// tick runs one poll cycle: a single batched query over the subscription
+// snapshot, cursor advancement, and emission for every change that moved.
+// An empty snapshot skips the cycle without touching the network.
+func (p *Poller) tick(ctx context.Context) {
+	changes := p.store.Changes()
+	if len(changes) == 0 {
+		return
+	}
+
+	res, err := p.client.QueryChanges(ctx, batchQuery(changes), len(changes), 0)
+	if err != nil {
+		p.lgr.Error("review notifications poll", "error", err)
+
+		return
+	}
+
+	for i := range res.Changes {
+		ci := &res.Changes[i]
+
+		if !p.store.Advance(ci.Number, ci.Updated.Time) {
+			continue
+		}
+
+		p.emit(ctx, ci)
+	}
+}
+
+func (p *Poller) emit(ctx context.Context, ci *gerrit.ChangeInfo) {
+	meta := map[string]string{"change": strconv.Itoa(ci.Number)}
+
+	if err := p.emitter.Emit(ctx, renderActivity(ci), meta); err != nil {
+		p.lgr.Error("review notification emit", "change", ci.Number, "error", err)
+	}
+}
+
+// batchQuery composes one change:A OR change:B query over the snapshot, so a
+// tick costs a single request regardless of subscription count.
+func batchQuery(changes []int) string {
+	clauses := make([]string, len(changes))
+	for i, n := range changes {
+		clauses[i] = "change:" + strconv.Itoa(n)
+	}
+
+	return strings.Join(clauses, " OR ")
+}
+
+// renderActivity renders the tracer-bullet payload: the fact of movement on a
+// subscribed change. Activity details (messages, votes, comment threads)
+// arrive with the deltas phase.
+func renderActivity(ci *gerrit.ChangeInfo) string {
+	return llmxml.NewElement("review_activity",
+		llmxml.Attr("change", ci.Number),
+		llmxml.Attr("project", ci.Project),
+		llmxml.Attr("status", ci.Status),
+		llmxml.Attr("updated", ci.Updated.UTC().Format(time.RFC3339)),
+	).String()
+}

internal/notifications/poller_internal_test.go

@@ -0,0 +1,272 @@
+package notifications
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"dev.gaijin.team/go/golib/e"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"dev.gaijin.team/go/go-gerrit-mcp/internal/config"
+	"dev.gaijin.team/go/go-gerrit-mcp/internal/gerritclient"
+)
+
+const selfJSON = ")]}'\n" + `{"_account_id":42,"name":"Review Bot","username":"bot"}`
+
+// changeJSON renders one change query result; updated is a Gerrit-format
+// timestamp such as "2026-07-01 10:00:00.000000000".
+func changeJSON(updated string) string {
+	return ")]}'\n" + `[{"_number":123,"project":"core","status":"NEW","updated":"` + updated + `"}]`
+}
+
+type emitCall struct {
+	content string
+	meta    map[string]string
+}
+
+// fakeEmitter records emissions; a non-nil err makes every Emit fail.
+type fakeEmitter struct {
+	mu    sync.Mutex `exhaustruct:"optional"`
+	calls []emitCall `exhaustruct:"optional"`
+	err   error      `exhaustruct:"optional"`
+}
+
+func (f *fakeEmitter) Emit(_ context.Context, content string, meta map[string]string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	f.calls = append(f.calls, emitCall{content: content, meta: meta})
+
+	return f.err
+}
+
+func (f *fakeEmitter) recorded() []emitCall {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	return append([]emitCall(nil), f.calls...)
+}
+
+// gerritStub serves the credential-validation endpoint plus change queries,
+// counting query requests and answering them via the swappable handler.
+type gerritStub struct {
+	queries atomic.Int64                     `exhaustruct:"optional"`
+	handler atomic.Pointer[http.HandlerFunc] `exhaustruct:"optional"`
+}
+
+func (g *gerritStub) setHandler(h http.HandlerFunc) {
+	g.handler.Store(&h)
+}
+
+func (g *gerritStub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Path == "/a/accounts/self" {
+		_, _ = w.Write([]byte(selfJSON))
+
+		return
+	}
+
+	g.queries.Add(1)
+	(*g.handler.Load())(w, r)
+}
+
+// pollerFixture bundles a poller with the stub Gerrit, the recording
+// emitter, and the captured log output behind it.
+type pollerFixture struct {
+	poller  *Poller
+	stub    *gerritStub
+	emitter *fakeEmitter
+	logs    *bytes.Buffer
+}
+
+// newTestPoller wires a poller against a stub Gerrit; the interval only
+// matters for Run-based tests.
+func newTestPoller(t *testing.T, interval time.Duration) *pollerFixture {
+	t.Helper()
+
+	stub := &gerritStub{}
+	stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte(")]}'\n[]"))
+	})
+
+	srv := httptest.NewServer(stub)
+	t.Cleanup(srv.Close)
+
+	client, err := gerritclient.New(t.Context(), &config.Config{
+		GerritURL:                       srv.URL,
+		Username:                        "bot",
+		Token:                           "s3cret",
+		Groups:                          []config.Group{config.GroupRead},
+		IncludeTools:                    nil,
+		ExcludeTools:                    nil,
+		Projects:                        nil,
+		AllowForeignChanges:             false,
+		ReviewNotifications:             true,
+		ReviewNotificationsPollInterval: interval,
+	})
+	require.NoError(t, err)
+
+	emitter := &fakeEmitter{}
+	logs := &bytes.Buffer{}
+	lgr := slog.New(slog.NewTextHandler(logs, nil))
+
+	return &pollerFixture{
+		poller:  NewPoller(NewStore(), client, emitter, interval, lgr),
+		stub:    stub,
+		emitter: emitter,
+		logs:    logs,
+	}
+}
+
+func Test_Poller_Tick(t *testing.T) {
+	t.Parallel()
+
+	base := time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC)
+
+	t.Run("movement emits once and advances the cursor", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(changeJSON("2026-07-01 11:00:00.000000000")))
+		})
+
+		f.poller.store.Add(123, base)
+
+		f.poller.tick(t.Context())
+		f.poller.tick(t.Context())
+
+		calls := f.emitter.recorded()
+		require.Len(t, calls, 1, "second tick over the same state must be silent")
+		assert.Equal(t,
+			`<review_activity change="123" project="core" status="NEW" updated="2026-07-01T11:00:00Z"/>`,
+			calls[0].content)
+		assert.Equal(t, map[string]string{"change": "123"}, calls[0].meta)
+	})
+
+	t.Run("no movement emits nothing", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(changeJSON("2026-07-01 10:00:00.000000000")))
+		})
+
+		f.poller.store.Add(123, base)
+
+		f.poller.tick(t.Context())
+
+		assert.Empty(t, f.emitter.recorded())
+	})
+
+	t.Run("empty subscription set skips the query", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+
+		f.poller.tick(t.Context())
+
+		assert.Zero(t, f.stub.queries.Load(), "no subscriptions must mean no network request")
+		assert.Empty(t, f.emitter.recorded())
+	})
+
+	t.Run("query failure is logged and the next tick recovers", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusInternalServerError)
+		})
+
+		f.poller.store.Add(123, base)
+
+		f.poller.tick(t.Context())
+
+		assert.Empty(t, f.emitter.recorded())
+		assert.Contains(t, f.logs.String(), "review notifications poll")
+
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(changeJSON("2026-07-01 11:00:00.000000000")))
+		})
+
+		f.poller.tick(t.Context())
+
+		assert.Len(t, f.emitter.recorded(), 1, "poll failure must not end the subscription")
+	})
+
+	t.Run("emit failure is logged and does not stop the tick", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(changeJSON("2026-07-01 11:00:00.000000000")))
+		})
+
+		f.emitter.err = e.New("session gone")
+
+		f.poller.store.Add(123, base)
+
+		f.poller.tick(t.Context())
+
+		assert.Len(t, f.emitter.recorded(), 1)
+		assert.Contains(t, f.logs.String(), "review notification emit")
+	})
+
+	t.Run("result for an unsubscribed change is ignored", func(t *testing.T) {
+		t.Parallel()
+
+		f := newTestPoller(t, time.Minute)
+		f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(changeJSON("2026-07-01 11:00:00.000000000")))
+		})
+
+		f.poller.store.Add(456, base)
+
+		f.poller.tick(t.Context())
+
+		assert.Empty(t, f.emitter.recorded())
+	})
+}
+
+func Test_Poller_Run(t *testing.T) {
+	t.Parallel()
+
+	const pollDeadline = 10 * time.Second
+
+	base := time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC)
+
+	f := newTestPoller(t, 5*time.Millisecond)
+	f.stub.setHandler(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte(changeJSON("2026-07-01 11:00:00.000000000")))
+	})
+
+	f.poller.store.Add(123, base)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		f.poller.Run(ctx)
+	}()
+
+	require.Eventually(t, func() bool {
+		return len(f.emitter.recorded()) > 0
+	}, pollDeadline, time.Millisecond, "ticker must fire and emit")
+
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(pollDeadline):
+		t.Fatal("Run did not stop on context cancellation")
+	}
+}