Merge branch 'main' into tyler/project-reduce-ci-costs

Author: tylerslaton

apps/dojo/e2e/featurePages/ToolBaseGenUIPage.ts

@@ -125,19 +125,19 @@ export class ToolBaseGenUIPage {
     return mainHaikuContent;
   }
 
-  async checkHaikuDisplay(page: Page): Promise<void> {
-    const chatHaikuContent = await this.extractChatHaikuContent(page);
-
-    await page.waitForTimeout(3000);
-
-    // Check that the haiku exists somewhere in the carousel
+  private async carouselIncludesHaiku(
+    page: Page,
+    chatHaikuContent: string,
+  ): Promise<boolean> {
     const carousel = page.locator('[data-testid="haiku-carousel"]');
-    await carousel.waitFor({ state: "visible", timeout: 10000 });
+
+    if (!(await carousel.isVisible())) {
+      return false;
+    }
 
     const allCarouselCards = carousel.locator('[data-testid="haiku-card"]');
     const cardCount = await allCarouselCards.count();
 
-    let foundMatch = false;
     for (let i = 0; i < cardCount; i++) {
       const card = allCarouselCards.nth(i);
       const lines = card.locator('[data-testid="haiku-japanese-line"]');
@@ -151,11 +151,21 @@ export class ToolBaseGenUIPage {
 
       const cardContent = cardLines.join("").replace(/\s/g, "");
       if (cardContent === chatHaikuContent) {
-        foundMatch = true;
-        break;
+        return true;
       }
     }
 
-    expect(foundMatch).toBe(true);
+    return false;
+  }
+
+  async checkHaikuDisplay(page: Page): Promise<void> {
+    const chatHaikuContent = await this.extractChatHaikuContent(page);
+
+    await expect
+      .poll(
+        async () => this.carouselIncludesHaiku(page, chatHaikuContent),
+        { timeout: 20000, intervals: [500, 1000, 2000] },
+      )
+      .toBe(true);
   }
 }

integrations/langgraph/python/ag_ui_langgraph/agent.py

@@ -4,7 +4,13 @@
 import inspect
 
 from langgraph.graph.state import CompiledStateGraph
-from langchain.schema import BaseMessage, SystemMessage
+
+try:
+    from langchain.schema import BaseMessage, SystemMessage
+except ImportError:
+    # Langchain >= 1.0.0
+    from langchain_core.messages import BaseMessage, SystemMessage
+    
 from langchain_core.runnables import RunnableConfig, ensure_config
 from langchain_core.messages import HumanMessage
 from langgraph.types import Command

sdks/community/go/pkg/encoding/sse/writer.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"net/http"
 	"strings"
 	"sync"
 	"time"
@@ -67,6 +68,9 @@ func (w *SSEWriter) WriteBytes(ctx context.Context, writer io.Writer, event []by
 			return fmt.Errorf("SSE flush failed: %w", err)
 		}
 	}
+	if flusher, ok := writer.(flusherWithoutError); ok {
+		flusher.Flush()
+	}
 	return nil
 }
 
@@ -116,6 +120,9 @@ func (w *SSEWriter) WriteEventWithType(ctx context.Context, writer io.Writer, ev
 			return fmt.Errorf("SSE flush failed: %w", err)
 		}
 	}
+	if flusher, ok := writer.(flusherWithoutError); ok {
+		flusher.Flush()
+	}
 
 	return nil
 }
@@ -187,6 +194,10 @@ type flusher interface {
 	Flush() error
 }
 
+// flusherWithoutError is a type alias for http.Flusher.
+// It is used to flush the writer without returning an error.
+type flusherWithoutError = http.Flusher
+
 // CustomEvent is a simple implementation of events.Event for error and custom events
 type CustomEvent struct {
 	events.BaseEvent

sdks/community/go/pkg/encoding/sse/writer_test.go

@@ -68,6 +68,15 @@ func (fw *flushWriter) Flush() error {
 	return fw.flushError
 }
 
+type httpFlushWriter struct {
+	bytes.Buffer
+	flushCalled bool
+}
+
+func (fw *httpFlushWriter) Flush() {
+	fw.flushCalled = true
+}
+
 func TestNewSSEWriter(t *testing.T) {
 	writer := NewSSEWriter()
 	if writer == nil {
@@ -451,6 +460,39 @@ func TestSSEWriter_Flushing(t *testing.T) {
 	}
 }
 
+func TestSSEWriter_HTTPFlusherFallback(t *testing.T) {
+	ctx := context.Background()
+	writer := NewSSEWriter()
+
+	t.Run("WriteEvent", func(t *testing.T) {
+		fw := &httpFlushWriter{}
+		event := &mockEvent{
+			BaseEvent: events.BaseEvent{
+				EventType: events.EventTypeCustom,
+			},
+		}
+
+		if err := writer.WriteEvent(ctx, fw, event); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		if !fw.flushCalled {
+			t.Error("expected fallback flusher to be called")
+		}
+	})
+
+	t.Run("WriteBytes", func(t *testing.T) {
+		fw := &httpFlushWriter{}
+		if err := writer.WriteBytes(ctx, fw, []byte(`{"test":"data"}`)); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		if !fw.flushCalled {
+			t.Error("expected fallback flusher to be called")
+		}
+	})
+}
+
 func TestCustomEvent(t *testing.T) {
 	t.Run("Data operations", func(t *testing.T) {
 		event := &CustomEvent{}