okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca2026-10-03

// A controlled peer holds snapshots across stream updates and delivers replies
// late. Real provider timing cannot reliably create these response orderings.
package tui

import (
	"albedo/cli/internal/testwire"
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"albedo/cli/internal/daemon"
	tea "charm.land/bubbletea/v2"
	"github.com/charmbracelet/x/ansi"
)

func TestAgentsRefreshDiscardsDirtyAndPreviousSubscriptionReplies(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
	defer cancel()
	feeds := []chan string{make(chan string, 4), make(chan string, 4), make(chan string, 4)}
	failureRelease, dirtyRelease := make(chan struct{}), make(chan struct{})
	var streams, snapshots, seeds atomic.Int32
	changed := make(chan struct{}, 16)
	notify := func() {
		select {
		case changed <- struct{}{}:
		default:
		}
	}
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch {
		case r.URL.Path == "/sessions" && r.Header.Get("Accept") == "text/event-stream":
			number := streams.Add(1)
			notify()
			if number > int32(len(feeds)) {
				t.Error("unexpected extra subscription")
				return
			}
			w.Header().Set("Content-Type", "text/event-stream")
			testwire.InitialStream(w, "agents")
			w.(http.Flusher).Flush()
			for {
				select {
				case batch := <-feeds[number-1]:
					_, _ = fmt.Fprintf(w, "data: %s\n\n", batch)
					w.(http.Flusher).Flush()
					if strings.Contains(batch, "overflow") {
						return
					}
				case <-r.Context().Done():
					return
				case <-ctx.Done():
					return
				}
			}
		case r.URL.Path == "/sessions":
			number := snapshots.Add(1)
			notify()
			if streams.Load() == 0 {
				t.Error("snapshot preceded subscription readiness")
			}
			name := "LIVE_TREE"
			switch number {
			case 1:
				select {
				case <-failureRelease:
				case <-ctx.Done():
					return
				}
				w.WriteHeader(http.StatusServiceUnavailable)
				_, _ = fmt.Fprint(w, `{"detail":"snapshot refused"}`)
				return
			case 2:
				select {
				case <-dirtyRelease:
				case <-ctx.Done():
					return
				}
				name = "DIRTY_TREE"
			case 4:
				name = "OLD_SUBSCRIPTION_TREE"
			case 5:
				name = "FINAL_TREE"
			}
			session := protocolSession("lead", generationA, 0)
			session["name"] = name
			_ = json.NewEncoder(w).Encode(map[string]any{"items": []any{session}, "next": nil, "family": map[string]any{"root_id": "lead", "revision": "family-a"}})
		case r.URL.Path == "/sessions/lead":
			session := protocolSession("lead", generationA, 0)
			if r.URL.Query().Has("tail") {
				preview := "OLD_SUBSCRIPTION_SEED"
				if seeds.Add(1) > 1 {
					preview = "FINAL_SEED"
				}
				session["preview"].(map[string]any)["transcript_count"] = 1
				session["history"].(map[string]any)["items"] = []any{protocolEntry("answer", "assistant", preview, 1)}
			}
			_ = json.NewEncoder(w).Encode(session)
		default:
			t.Errorf("unexpected request: %s", r.URL.Path)
			w.WriteHeader(http.StatusNotFound)
		}
	}))
	defer server.Close()
	defer cancel()
	connection := daemon.NewConnection(daemon.ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil)
	defer connection.HTTPClient().CloseIdleConnections()
	model := NewAgentsViewModel(connection, "lead")
	defer func() { model.Close() }()
	model.SetSize(120, 30)
	inbox := make(chan tea.Msg, 16)
	var run func(tea.Cmd)
	run = func(cmd tea.Cmd) {
		if cmd == nil {
			return
		}
		go func() {
			message := cmd()
			if batch, ok := message.(tea.BatchMsg); ok {
				for _, child := range batch {
					run(child)
				}
				return
			}
			switch message.(type) {
			case agentsSnapshotMsg, agentsEventsMsg, agentsStreamClosedMsg, agentsSnapshotRetryMsg, agentsSeedMsg, agentsSeedErrMsg:
				select {
				case inbox <- message:
				case <-ctx.Done():
				}
			}
		}()
	}
	update := func(message tea.Msg) { var cmd tea.Cmd; model, cmd = model.Update(message); run(cmd) }
	view := func() string { return ansi.Strip(model.View()) }
	pump := func(done func() bool) {
		t.Helper()
		for !done() {
			select {
			case message := <-inbox:
				update(message)
			case <-changed:
			case <-ctx.Done():
				t.Fatalf("refresh did not settle:\n%s", view())
			}
		}
	}
	take := func(matches func(tea.Msg) bool) tea.Msg {
		t.Helper()
		for {
			select {
			case message := <-inbox:
				if matches(message) {
					return message
				}
				update(message)
			case <-ctx.Done():
				t.Fatalf("expected response never arrived:\n%s", view())
				return nil
			}
		}
	}
	run(model.Init())
	pump(func() bool { return snapshots.Load() == 1 })
	feeds[0] <- `{"events":[{"type":"invalidate","data":{"urls":["/sessions"],"session_ids":["lead"],"scope_dirty":true}}]}`
	// The event must be consumed before releasing the failed snapshot.
	takeEvent := take(func(message tea.Msg) bool { batch, ok := message.(agentsEventsMsg); return ok && len(batch.Events) > 0 })
	update(takeEvent)
	close(failureRelease)
	pump(func() bool { return strings.Contains(view(), "snapshot refused") })
	pump(func() bool { return snapshots.Load() == 2 })
	feeds[0] <- `{"events":[{"type":"invalidate","data":{"urls":["/sessions"],"session_ids":["lead"],"scope_dirty":true}}]}`
	update(take(func(message tea.Msg) bool { batch, ok := message.(agentsEventsMsg); return ok && len(batch.Events) > 0 }))
	close(dirtyRelease)
	dirty := take(func(message tea.Msg) bool {
		snapshot, ok := message.(agentsSnapshotMsg)
		return ok && snapshot.Err == nil
	})
	update(dirty)
	if strings.Contains(view(), "DIRTY_TREE") {
		t.Fatal("dirty snapshot replaced the live graph")
	}
	pump(func() bool { return strings.Contains(view(), "LIVE_TREE") })
	oldSeed := take(func(message tea.Msg) bool { _, ok := message.(agentsSeedMsg); return ok })
	feeds[0] <- `{"events":[{"type":"overflow","data":{}}]}`
	oldSnapshot := take(func(message tea.Msg) bool {
		snapshot, ok := message.(agentsSnapshotMsg)
		return ok && snapshot.Err == nil
	})
	feeds[1] <- `{"events":[{"type":"overflow","data":{}}]}`
	pump(func() bool { return strings.Contains(view(), "FINAL_SEED") })
	update(oldSnapshot)
	update(oldSeed)
	if strings.Contains(view(), "OLD_SUBSCRIPTION") || !strings.Contains(view(), "FINAL_TREE") || !strings.Contains(view(), "FINAL_SEED") {
		t.Fatalf("late success restored stale membership or history:\n%s", view())
	}
	if streams.Load() != 3 || snapshots.Load() != 5 {
		t.Fatalf("recovery made unexpected requests: streams=%d snapshots=%d", streams.Load(), snapshots.Load())
	}
}