Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217// Stream preview state across partial tool arguments, detach, and snapshot boundaries is race-prone and hard to force with a live daemon.package daemon
import ( "context" "encoding/json" "fmt" "net" "net/http" "net/http/httptest" "strings" "sync/atomic" "testing")
func formatPage(cursor int, events []any) string { b, _ := json.Marshal(map[string]any{ "cursor": cursor, "events": events, }) return fmt.Sprintf("data: %s\n\n", string(b))}
func TestCompletedCallPreviewsReleaseState(t *testing.T) { aData, _ := json.Marshal(map[string]any{"code": "read('first.py')"}) bData, _ := json.Marshal(map[string]any{"code": "read('second.py')"}) a := string(aData) b := string(bData)
var requestCount atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { count := requestCount.Add(1) w.Header().Set("Content-Type", "text/event-stream") if count == 1 { _, _ = w.Write([]byte(formatPage(1, []any{ map[string]any{"type": "arguments_delta", "name": "python", "callId": "a", "text": a}, map[string]any{"type": "arguments_delta", "name": "python", "callId": "b", "text": b[:15]}, map[string]any{"type": "tool", "callId": "a", "name": "python", "args": a, "result": "done"}, }))) } else { _, _ = w.Write([]byte(formatPage(2, []any{ map[string]any{"type": "arguments_delta", "name": "python", "callId": "b", "text": b[15:]}, map[string]any{"type": "tool_progress", "progress": map[string]any{"callId": "b", "name": "python", "phase": "running"}}, map[string]any{"type": "tool", "callId": "b", "name": "python", "args": b, "result": "done"}, }))) } })) defer server.Close()
conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") client := NewChatClient(conn, "session")
var events []StreamEvent _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { events = append(events, event) return nil }) _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { events = append(events, event) return nil })
hasSecondPreview := false for _, ev := range events { if ev.Type == EventToolProgress && ev.Progress != nil && ev.Progress.Code != nil { if ev.Progress.Code.Text == "read('second.py')" { hasSecondPreview = true break } } } if !hasSecondPreview { t.Fatal("expected second.py progress preview") }
runningIdx := -1 for i, ev := range events { if ev.Type == EventToolProgress && ev.Progress != nil && ev.Progress.Phase == "running" { runningIdx = i break } } if runningIdx <= 0 { t.Fatal("expected running progress event preceded by clear") } if events[runningIdx-1].Type != EventToolProgress || events[runningIdx-1].Progress != nil { t.Fatalf("expected nil progress before running, got: %+v", events[runningIdx-1]) }
if events[len(events)-1].Type != EventTool { t.Fatalf("expected last event tool, got: %+v", events[len(events)-1]) }}
func TestUnfinishedArgumentsBounded(t *testing.T) { var requestedURLs []string var serverCallCount atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requestedURLs = append(requestedURLs, r.URL.String()) count := serverCallCount.Add(1) w.Header().Set("Content-Type", "text/event-stream") if count <= 33 { _, _ = w.Write([]byte(formatPage(int(count), []any{ map[string]any{"type": "arguments_delta", "name": "python", "callId": fmt.Sprintf("call-%d", count), "text": `{"code":"`}, }))) } else { _, _ = w.Write([]byte(formatPage(int(count), []any{ map[string]any{"type": "reset"}, map[string]any{"type": "message", "role": "assistant", "text": "recovered"}, }))) } })) defer server.Close()
conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") client := NewChatClient(conn, "session")
for range 32 { _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { return nil }) }
err := client.Stream(context.Background(), 0, func(event StreamEvent) error { return nil }) if err == nil { t.Fatalf("expected error on call count limit, got: %v", err) }
var events []StreamEvent _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { events = append(events, event) return nil })
lastURL := requestedURLs[len(requestedURLs)-1] if !strings.HasSuffix(lastURL, "after_seq=-1") { t.Fatalf("expected reconnect cursor reset to after_seq=-1, got: %s", lastURL) } if len(events) == 0 || events[len(events)-1].Text != "recovered" { t.Fatalf("expected recovered event, got: %+v", events) }}
func TestDetachingDropsArgumentPreviews(t *testing.T) { var requestedURLs []string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requestedURLs = append(requestedURLs, r.URL.String()) code := "old()" if len(requestedURLs) > 1 { code = "new()" } raw, _ := json.Marshal(map[string]any{"code": code}) w.Header().Set("Content-Type", "text/event-stream") _, _ = w.Write([]byte(formatPage(len(requestedURLs), []any{ map[string]any{"type": "arguments_delta", "name": "python", "callId": "same", "text": string(raw)}, }))) })) defer server.Close()
conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") client := NewChatClient(conn, "session")
ctx, cancel := context.WithCancel(context.Background()) _ = client.Stream(ctx, 0, func(event StreamEvent) error { if event.Type == EventToolProgress && event.Progress != nil { cancel() } return nil })
var events []StreamEvent _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { events = append(events, event) return nil })
if len(requestedURLs) < 2 || !strings.HasSuffix(requestedURLs[1], "after_seq=-1") { t.Fatalf("expected reconnect cursor reset to after_seq=-1, got: %+v", requestedURLs) } hasNew := false for _, ev := range events { if ev.Type == EventToolProgress && ev.Progress != nil && ev.Progress.Code != nil { if ev.Progress.Code.Text == "new()" { hasNew = true break } } } if !hasNew { t.Fatal("expected new() tool progress") }}
func TestSnapshotEventsAreMarkedReplayed(t *testing.T) { pages := formatPage(2, []any{ map[string]any{"type": "reset"}, map[string]any{"type": "thinking", "text": "earlier"}, }) + formatPage(3, []any{ map[string]any{"type": "thinking", "text": "now"}, }) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") _, _ = w.Write([]byte(pages)) })) defer server.Close() conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") client := NewChatClient(conn, "session") replayed := map[string]bool{} _ = client.Stream(context.Background(), 0, func(event StreamEvent) error { if event.Type == EventThinking { replayed[event.Text] = event.Replayed } return nil }) if !replayed["earlier"] || replayed["now"] { t.Fatalf("only the snapshot after a reset is history: %v", replayed) }}