test(stream): cover replay across restarts
Make old cursors meet new actors. Check durable resets, keepalives, failed reads, and older-history navigation.
12345678910111213%% Replace a real session actor and wait until its old lifetime has ended.-module(albedo_stream_replay_probe).-export([restart/1]).
restart(Id) -> {some, Session} = 'albedo@daemon@session':live(Id), {ok, Pid} = 'gleam@erlang@process':subject_owner(Session), Monitor = monitor(process, Pid), exit(Pid, kill), receive {'DOWN', Monitor, process, Pid, killed} -> <<"actor_stopped">> after 10000 -> error(actor_still_alive) end.123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177"""A replay cursor cannot skip durable events from a replacement session actor."""
import jsonimport socketimport sqlite3import subprocessimport unittest
from harness import Albedo, Provider, ROOT, exclusive, text
def restart_actor(app, session): support = ROOT / "test/e2e/albedo_stream_replay_probe.erl" subprocess.run( ["erlc", "-Werror", "-o", str(app.root), str(support)], check=True, timeout=30, ) cookie = (app.home / "inspect.cookie").read_text().strip() node = f"albedo_{app.daemon._pid}@{socket.gethostname().split('.')[0]}" beam = app.root / "albedo_stream_replay_probe.beam" expression = ( f"Node = '{node}', " f"{{ok, Binary}} = file:read_file({json.dumps(str(beam))}), " "{module, albedo_stream_replay_probe} = " "rpc:call(Node, code, load_binary, " '[albedo_stream_replay_probe, "probe.erl", Binary]), ' '<<"actor_stopped">> = rpc:call(Node, albedo_stream_replay_probe, ' f"restart, [<<{json.dumps(session)}>>]), " 'io:put_chars("actor_stopped"), halt().' ) result = subprocess.run( [ "erl", "+S", "2:2", "-sname", "stream_replay_probe", "-setcookie", cookie, "-noshell", "-eval", expression, ], capture_output=True, text=True, check=True, timeout=30, ) if result.stdout != "actor_stopped": raise AssertionError(result.stdout + result.stderr)
class StreamReplayTest(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _: text("answer")) self.addCleanup(self.provider.close)
def turn(self, app, session, prompt): app.prompt(session, prompt).close() app.idle(session)
def assert_reset_history(self, page, prompts): self.assertEqual(page["events"][0]["type"], "reset") self.assertEqual( [event["text"] for event in page["events"] if event["type"] == "user"], prompts, )
@exclusive def test_old_actor_cursor_inside_replacement_window_resets_durable_history(self): def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1"
with Albedo(self.provider, prepare=prepare) as app: session = app.session() self.turn(app, session, "before actor restart") old = app.stream_page(session) restart_actor(app, session) self.turn(app, session, "after actor restart") current = app.stream_page(session) self.assertGreaterEqual(current["cursor"], old["cursor"]) replay = app.stream_page(session, old) self.assert_reset_history( replay, ["before actor restart", "after actor restart"] ) self.assertNotEqual(replay["generation"], old["generation"]) self.assertEqual(replay["generation"], current["generation"])
@exclusive def test_old_daemon_cursor_resets_durable_history(self): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "before daemon restart") old = app.stream_page(session) app.restart(crash=True) self.turn(app, session, "after daemon restart") current = app.stream_page(session) self.assertGreaterEqual(current["cursor"], old["cursor"]) replay = app.stream_page(session, old) self.assert_reset_history( replay, ["before daemon restart", "after daemon restart"] ) self.assertNotEqual(replay["generation"], old["generation"])
def test_same_generation_replays_only_subsequent_events_in_order(self): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "already consumed") old = app.stream_page(session) self.turn(app, session, "next prompt") self.turn(app, session, "last prompt") replay = app.stream_page(session, old) self.assertEqual(replay["generation"], old["generation"]) self.assertGreater(replay["cursor"], old["cursor"]) self.assertNotIn("reset", [event["type"] for event in replay["events"]]) self.assertEqual( [ event["text"] for event in replay["events"] if event["type"] == "user" ], ["next prompt", "last prompt"], ) keepalive = app.stream_page(session, replay) self.assertEqual(keepalive["events"], []) self.assertEqual(keepalive["generation"], replay["generation"]) self.assertEqual(keepalive["cursor"], replay["cursor"])
@exclusive def test_failed_transcript_read_emits_failure_without_replacement_cursor(self): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "committed history") old = app.stream_page(session) with sqlite3.connect(app.home / "albedo.sqlite") as database: database.execute( "ALTER TABLE transcript RENAME TO unavailable_transcript" ) with app.api( f"/sessions/{session}/stream?after_generation=other&after_seq={old['cursor']}&tail=2" ) as response: lines = list(response) self.assertIn(b"event: error\n", lines) batches = [ json.loads(line[6:]) for line in lines if line.startswith(b"data: ") ] self.assertEqual(len(batches), 1) self.assertIn("could not load transcript", batches[0]["error"]) self.assertNotIn("generation", batches[0]) self.assertNotIn("cursor", batches[0])
def test_evicted_cursor_resets_in_same_generation_and_initial_tail_pages_history( self, ): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "oldest prompt") old = app.stream_page(session) # A long streamed answer evicts the count-limited replay window. self.provider.script = lambda _: text("long answer " * 600) self.turn(app, session, "newest prompt") replay = app.stream_page(session, old) self.assertEqual(replay["generation"], old["generation"]) self.assert_reset_history(replay, ["oldest prompt", "newest prompt"]) initial = app.stream_page(session, tail=2) self.assert_reset_history(initial, ["newest prompt"]) reset = initial["events"][0] self.assertTrue(reset["more"]) with app.api( f"/sessions/{session}/history?rows=2&before={reset['before']}" ) as response: older = json.load(response) self.assertEqual( [event["text"] for event in older["events"] if event["type"] == "user"], ["oldest prompt"], )25 unmodified lines26272829303132333430313233343536373839407 unmodified lines4849505152535452555657585910 unmodified lines7071727371747576777879808182838485865 unmodified lines92939495969798991001011021031041051069710710810911011178 unmodified lines19019119219318419419519619719885 unmodified lines28428528628728828927929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134225 unmodified lines tea "charm.land/bubbletea/v2")
type streamFaultProxy struct { connection *daemon.Connection mu sync.Mutex cursors []string firstCursor int64 reconnected chan struct{} connection *daemon.Connection mu sync.Mutex cursors []string generations []string firstCursor int64 firstGeneration string reconnected chan struct{}}
func faultSessionStream(t *testing.T, session string, fault string) *streamFaultProxy { t.Helper()7 unmodified lines server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/sessions/"+session+"/stream" { proxy.mu.Lock() proxy.cursors = append(proxy.cursors, r.URL.Query().Get("after_seq")) proxy.generations = append(proxy.generations, r.URL.Query().Get("after_generation")) count := len(proxy.cursors) proxy.mu.Unlock() if fault == "eof" { if fault == "eof" || fault == "after-history" { if count == 1 { outgoing := r.Clone(r.Context()) outgoing.URL.Scheme, outgoing.URL.Host = destination.Scheme, destination.Host outgoing.RequestURI = ""10 unmodified lines line := scanner.Text() _, _ = fmt.Fprintln(w, line) if strings.HasPrefix(line, "data: ") { var envelope struct { Cursor int64 `json:"cursor"` Cursor int64 `json:"cursor"` Generation string `json:"generation"` } if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &envelope); err != nil { t.Error(err) } proxy.mu.Lock() proxy.firstCursor = envelope.Cursor proxy.firstGeneration = envelope.Generation proxy.mu.Unlock() } if line == "" { return5 unmodified lines t.Error("daemon stream ended before a complete SSE frame") } return } if fault == "after-history" && count == 2 { w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: "+`{"generation":"invalid-replacement","cursor":10,"events":[{"type":"text","text":"poison must stay hidden"}]}`+"\n\n") return } select { case proxy.reconnected <- struct{}{}: default: } } else if count == 1 || fault == "repeated" { w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: "+`{"cursor":10,"events":[{"type":"text","text":"poison must stay hidden"},{"type":"message","role":"assistant","text":false}]}`+"\n\n") _, _ = io.WriteString(w, "data: "+`{"generation":"fault","cursor":10,"events":[{"type":"text","text":"poison must stay hidden"},{"type":"message","role":"assistant","text":false}]}`+"\n\n") return } } forward.ServeHTTP(w, r)78 unmodified lines if !recoverySeen || strings.Contains(view, "poison must stay hidden") || strings.Count(view, "durable stream recovery") != 2 { t.Fatalf("reset lost or duplicated durable transcript:\n%s", view) } requests := proxy.requests() if len(requests) != 2 || requests[1] != "-1" { if len(requests) != 2 || requests[1] != "" { t.Fatalf("protocol recovery did not request exactly one durable reset: %v", requests) }}
85 unmodified lines t.Fatal(err) } proxy.mu.Lock() requests, first := append([]string(nil), proxy.cursors...), proxy.firstCursor generations, firstGeneration := append([]string(nil), proxy.generations...), proxy.firstGeneration proxy.mu.Unlock() if len(requests) != 2 || requests[1] != fmt.Sprint(first) { if len(requests) != 2 || requests[1] != fmt.Sprint(first) || generations[1] != firstGeneration || firstGeneration == "" { t.Fatalf("EOF lost consumed cursor: requests=%v, cursor=%d", requests, first) } view := driver.View() if strings.Count(view, "before EOF") != 2 || strings.Count(view, "after EOF") != 2 { t.Fatalf("EOF replay lost or duplicated rows:\n%s", view) }}
func TestTUIResetReplacesHistoryAndPreservesOlderNavigation(t *testing.T) { profile := providerRoute(t, echoReply) session := daemonSession(t, newSession(t, t.TempDir())) client := daemon.NewChatClient(conn(t), session.ID) const turns = 62 for index := range turns { if _, err := client.Send(t.Context(), fmt.Sprintf("history row %03d", index), nil); err != nil { t.Fatal(err) } waitIdle(t, session.ID, profile, index+1) } proxy := faultSessionStream(t, session.ID, "after-history") driver := driveTUIWithConnection(t, &session, proxy.connection) t.Cleanup(driver.App.Chat.Close) driver.Update(tea.WindowSizeMsg{Width: 100, Height: 50}) recoverySeen := false pumpStreamUntil(t, driver, func(msg tea.Msg) bool { if result, ok := msg.(tui.ChatStreamResultMsg); ok && result.Recovering { recoverySeen = true } event, ok := msg.(tui.ChatStreamEventMsg) return recoverySeen && ok && event.Event.Type == daemon.EventMessage && strings.Contains(event.Event.Text, "history row 061") }) if requests := proxy.requests(); len(requests) != 3 || requests[2] != "" { t.Fatalf("generation failure did not recover with a fresh subscription: %v", requests) } seen := map[string]int{} for _, entry := range driver.App.Chat.History.Entries() { seen[entry.Text]++ } if seen["history row 061"] != 1 || seen["history row 000"] != 0 || strings.Contains(driver.View(), "poison must stay hidden") { t.Fatalf("reset duplicated history or lost the tail boundary: %v", seen) } for range 4 { driver.Dispatch(tea.KeyPressMsg{Code: tea.KeyHome, Mod: tea.ModCtrl}) } seen = map[string]int{} for _, entry := range driver.App.Chat.History.Entries() { seen[entry.Text]++ } if seen["history row 000"] != 1 || seen["history row 061"] != 1 { t.Fatalf("older navigation after reset lost or duplicated rows: %v", seen) }}