Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134package acp
// These tests force buffer overflow and control the replay, response, and live// event ordering that network timing cannot reproduce reliably.import ( "bytes" "context" "encoding/json" "errors" "strconv" "strings" "testing" "time"
"albedo/cli/internal/daemon")
type replayBackend struct { Backend history func(int64) ([]daemon.StreamEvent, int64, error) historyContext func(context.Context, int64) ([]daemon.StreamEvent, int64, error)}
func (b replayBackend) History(ctx context.Context, _ string, cursor daemon.ACPHistoryCursor, cutoff int64) ([]daemon.StreamEvent, daemon.ACPHistoryCursor, error) { var events []daemon.StreamEvent var position int64 var err error if b.historyContext != nil { events, position, err = b.historyContext(ctx, cursor.Position) } else { events, position, err = b.history(cursor.Position) } return events, daemon.ACPHistoryCursor{Position: position, Token: strconv.FormatInt(position, 10), Done: position >= cutoff}, err}
func TestLoadOrdersReplayResponseAndLive(t *testing.T) { s, backend := testSession(t) s.turn = "turn" s.buffering = true s.historyCutoff = 2 s.adapter.backend = replayBackend{history: func(after int64) ([]daemon.StreamEvent, int64, error) { if after == 0 { s.mu.Lock() err := s.consumeLocked(daemon.StreamEvent{Type: daemon.EventText, TurnID: "turn", MessageID: "later", Text: "live"}) s.mu.Unlock() if err != nil { t.Fatal(err) } } return []daemon.StreamEvent{{Type: daemon.EventMessage, EntryID: string(rune('a' + after)), Text: "same"}}, after + 1, nil }} if err := s.load(request{ID: json.RawMessage(`1`)}); err != nil { t.Fatal(err) } var messages []map[string]any for range 4 { var message map[string]any if err := json.Unmarshal(<-backend.written, &message); err != nil { t.Fatal(err) } messages = append(messages, message) } if messages[2]["id"] != float64(1) { t.Fatalf("response interleaved: %#v", messages) } for _, index := range []int{0, 1, 3} { if messages[index]["method"] != "session/update" { t.Fatal(messages) } } if s.buffering || len(s.buffered) > 0 { t.Fatal("buffer retained") }}
func TestLoadRejectsOverflowAndCursorReset(t *testing.T) { for _, event := range []daemon.StreamEvent{{Type: daemon.EventText, Text: strings.Repeat("x", maxUnsent)}, {Type: daemon.EventReset}} { s, _ := testSession(t) s.buffering = true s.ready = make(chan struct{}) if err := s.consumeLocked(event); err == nil { t.Fatal("unsafe load accepted") } if err := s.load(request{ID: json.RawMessage(`1`)}); err == nil { t.Fatal("load succeeded after failure") } }}
func TestLoadLargeDurableMessageDoesNotOverflowWriter(t *testing.T) { s, _ := testSession(t) var output bytes.Buffer s.adapter.writer = newWriter(&output) go s.adapter.writer.run(s.ctx) s.buffering = true s.historyCutoff = 1 text := strings.Repeat("\"", maxUnsent+1) s.adapter.backend = replayBackend{history: func(int64) ([]daemon.StreamEvent, int64, error) { return []daemon.StreamEvent{{Type: daemon.EventMessage, EntryID: "durable", Text: text}}, 1, nil }} if err := s.load(request{ID: json.RawMessage(`1`)}); err != nil { t.Fatal(err) } <-s.adapter.writer.flush() if s.adapter.writer.failure() != nil || !strings.Contains(output.String(), `"id":1`) { t.Fatal("large replay failed") }}
func TestLoadOverflowCancelsOutstandingHistoryRead(t *testing.T) { s, _ := testSession(t) s.buffering = true s.historyCutoff = 1 s.adapter.backend = replayBackend{historyContext: func(ctx context.Context, _ int64) ([]daemon.StreamEvent, int64, error) { s.mu.Lock() err := s.consumeLocked(daemon.StreamEvent{Type: daemon.EventText, Text: strings.Repeat("x", maxUnsent)}) s.mu.Unlock() if err == nil { t.Error("overflow accepted") } select { case <-ctx.Done(): return nil, 0, ctx.Err() case <-time.After(time.Second): t.Error("history read not canceled") return nil, 0, errors.New("read timeout") } }} var failure *rpcError if err := s.load(request{ID: json.RawMessage(`1`)}); !errors.As(err, &failure) || failure.Data["reason"] != "load_buffer_overflow" { t.Fatal(err) }}