refactor: extract TruncateRunes to internal/utility and fix integration build (#17938) (#17944)

internal/ingestion/pipeline/real_storage_integration_test.go

@@ -22,9 +22,9 @@ import (
 	componentpkg "ragflow/internal/ingestion/component"
 	_ "ragflow/internal/ingestion/component/chunker"
 	"ragflow/internal/server"
+	"ragflow/internal/server/config"
 	"ragflow/internal/storage"
 
-	"go.uber.org/zap"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
 	"gorm.io/gorm/logger"
@@ -46,7 +46,7 @@ func TestPipelineRun_TemplateGeneral_RealMySQLMinIO_OutputShape(t *testing.T) {
 		t.Fatalf("auto-migrate real mysql tables: %v", err)
 	}
 
-	realStorage, err := storage.NewMinioStorage(cfg.StorageEngine.Minio)
+	realStorage, err := storage.NewMinioStorage(cfg.GetMinioConfig())
 	if err != nil {
 		t.Fatalf("connect real minio: %v", err)
 	}
@@ -181,15 +181,17 @@ func TestPipelineRun_TemplateGeneral_RealMySQLMinIO_OutputShape(t *testing.T) {
 	}
 }
 
-func mustLoadRealIntegrationConfig(t *testing.T) *server.Config {
+func mustLoadRealIntegrationConfig(t *testing.T) *config.Config {
 	t.Helper()
-	server.SetLogger(zap.NewNop())
+	if err := common.InitLogger("info", common.FileOutput{}, ""); err != nil {
+		t.Fatalf("init logger: %v", err)
+	}
 	configPath := filepath.Join(repoRootFromPipelineTest(t), "conf", "service_conf.yaml")
 	if err := server.Init(configPath); err != nil {
 		t.Fatalf("init service config from %s: %v", configPath, err)
 	}
 	cfg := server.GetConfig()
-	if cfg == nil || cfg.Database.Host == "" || cfg.StorageEngine.Minio == nil || cfg.StorageEngine.Minio.Host == "" {
+	if cfg == nil || cfg.GetMySQLConfig().Host == "" || cfg.GetMinioConfig().Host == "" {
 		t.Fatal("real integration config is incomplete")
 	}
 	return cfg
@@ -300,15 +302,16 @@ func mustPrepareTokenizerOpenCC(t *testing.T, root string) {
 	mustSymlink(t, systemOpenCC, filepath.Join(root, "opencc"))
 }
 
-func mustOpenRealMySQL(t *testing.T, cfg *server.Config) *gorm.DB {
+func mustOpenRealMySQL(t *testing.T, cfg *config.Config) *gorm.DB {
 	t.Helper()
+	mc := cfg.GetMySQLConfig()
 	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=%s&parseTime=True&loc=Local",
-		cfg.Database.Username,
-		cfg.Database.Password,
-		cfg.Database.Host,
-		cfg.Database.Port,
-		cfg.Database.Database,
-		cfg.Database.Charset,
+		mc.User,
+		mc.Password,
+		mc.Host,
+		mc.Port,
+		mc.DatabaseName,
+		mc.Charset,
 	)
 	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
 		Logger: logger.Default.LogMode(logger.Silent),
@@ -386,7 +389,7 @@ func mustSeedRealPipelineDocument(
 	}).Error; err != nil {
 		t.Fatalf("create kb: %v", err)
 	}
-	if err := stg.Put(bucket, objectPath, []byte(content)); err != nil {
+	if err := stg.Put(context.Background(), bucket, objectPath, []byte(content)); err != nil {
 		t.Fatalf("put real minio object: %v", err)
 	}
 	if err := db.Create(&entity.File{
@@ -431,7 +434,7 @@ func cleanupRealPipelineDocument(db *gorm.DB, stg storage.Storage, tenantID, kbI
 	_ = db.Where("id = ?", fileID).Delete(&entity.File{}).Error
 	_ = db.Where("id = ?", kbID).Delete(&entity.Knowledgebase{}).Error
 	_ = db.Where("id = ?", tenantID).Delete(&entity.Tenant{}).Error
-	_ = stg.Remove(bucket, objectPath)
+	_ = stg.Remove(context.Background(), bucket, objectPath)
 }
 
 func strPtr(s string) *string {

internal/ingestion/pipeline/template_integration_test.go

@@ -48,6 +48,7 @@ import (
 type fixedEmbedder struct{}
 
 func (fixedEmbedder) MaxTokens() int { return 2048 }
+func (fixedEmbedder) BatchSize() int { return 16 }
 
 func (fixedEmbedder) Encode(ctx context.Context, texts []string) ([]componentpkg.EmbeddingResult, error) {
 	out := make([]componentpkg.EmbeddingResult, 0, len(texts))
@@ -911,7 +912,7 @@ func seedTemplateDocument(t *testing.T, stg storage.Storage, name, bucket, path,
 
 func seedTemplateDocumentBytes(t *testing.T, stg storage.Storage, name, bucket, path string, content []byte) string {
 	t.Helper()
-	if err := stg.Put(bucket, path, content); err != nil {
+	if err := stg.Put(context.Background(), bucket, path, content); err != nil {
 		t.Fatalf("seed storage: %v", err)
 	}
 	if registerTemplateDocumentRef == nil {

internal/ingestion/task/debug_log_sink.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"ragflow/internal/ingestion/pipeline"
+	"ragflow/internal/utility"
 )
 
 // DebugLogTTL is the Redis expiry for a debug-run log. It mirrors the Python
@@ -170,7 +171,7 @@ func (s *DebugLogSink) OnComponentProgress(_ context.Context, ev pipeline.Progre
 		message = "[ERROR] " + message
 	}
 	// Clamp the message so a runaway component cannot blow up the stored entry.
-	message = truncateRunes(message, maxMessageRunes)
+	message = utility.TruncateRunes(message, maxMessageRunes)
 
 	entry := debugTrace{
 		Progress:    progress,
@@ -291,19 +292,6 @@ func endOutputMessage(entries []debugLogEntry, runOutput map[string]any) string
 	return string(b)
 }
 
-// truncateRunes returns s truncated to at most max runes, preserving the original
-// bytes when shorter. Mirrors internal/service/agent_sessions.go's truncateRunes.
-func truncateRunes(s string, max int) string {
-	if max <= 0 {
-		return ""
-	}
-	r := []rune(s)
-	if len(r) <= max {
-		return s
-	}
-	return string(r[:max])
-}
-
 // trimToPayloadBudget collapses the middle of the log (keeping the first few
 // entries plus the END marker at the tail) until the marshaled size fits the
 // byte budget. The END marker is always the last element, so it is preserved.

internal/service/agent_sessions.go

@@ -591,7 +591,7 @@ func normalizeAgentTags(rawTags interface{}) (string, error) {
 	normalized := make([]string, 0, len(cleaned))
 	used := 0
 	for _, tag := range cleaned {
-		tag = truncateRunes(tag, agentTagMaxLen)
+		tag = utility.TruncateRunes(tag, agentTagMaxLen)
 		key := strings.ToLower(tag)
 		if _, ok := seen[key]; ok {
 			continue
@@ -612,14 +612,6 @@ func normalizeAgentTags(rawTags interface{}) (string, error) {
 	return strings.Join(normalized, ","), nil
 }
 
-func truncateRunes(value string, maxLen int) string {
-	runes := []rune(value)
-	if len(runes) <= maxLen {
-		return value
-	}
-	return string(runes[:maxLen])
-}
-
 // UpdateAgentTags normalises tags and persists them on a single canvas.
 func (s *AgentService) UpdateAgentTags(ctx context.Context, userID, canvasID string, tags interface{}) (bool, common.ErrorCode, error) {
 	ok, err := s.CheckCanvasAccess(ctx, userID, canvasID)

internal/service/nlp/datasetnav_integration_test.go

@@ -10,12 +10,11 @@ import (
 	"strings"
 	"testing"
 
+	"ragflow/internal/common"
 	"ragflow/internal/engine"
 	"ragflow/internal/engine/types"
 	"ragflow/internal/server"
 	"ragflow/internal/service/nav"
-
-	"go.uber.org/zap"
 )
 
 // repoRootOf walks up from the package directory to the repository root (the
@@ -72,12 +71,14 @@ func findNavRow(t *testing.T, tenantID, kbID, docID string) map[string]interface
 //
 // Run with: bash build.sh --test-integration ./internal/service/nlp/...
 func TestDatasetNav_AvailableIntZero_Isolation(t *testing.T) {
-	server.SetLogger(zap.NewNop())
+	if err := common.InitLogger("info", common.FileOutput{}, ""); err != nil {
+		t.Fatalf("init logger: %v", err)
+	}
 	configPath := filepath.Join(repoRootOf(t), "conf", "service_conf.yaml")
 	if err := server.Init(configPath); err != nil {
 		t.Fatalf("init service config: %v", err)
 	}
-	if err := engine.Init(); err != nil {
+	if err := engine.InitDocEngine(); err != nil {
 		t.Fatalf("init document engine: %v", err)
 	}
 	if engine.Get() == nil {

internal/storage/minio_test.go

@@ -20,6 +20,7 @@ package storage
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"log"
 	"ragflow/internal/utility"
@@ -27,19 +28,20 @@ import (
 	"time"
 
 	"ragflow/internal/server"
+	configpkg "ragflow/internal/server/config"
 )
 
 // getMinioConfig returns MinIO configuration for testing
 // Configuration can be loaded from environment variables or config file
-func getMinioConfig() (*server.MinioConfig, error) {
+func getMinioConfig() (configpkg.MinioConfig, error) {
 
 	// Initialize configuration
 	if err := server.Init(""); err != nil {
-		return nil, err
+		return configpkg.MinioConfig{}, err
 	}
 
 	// Try to get configuration from environment variables first
-	config := server.GetConfig().StorageEngine.Minio
+	config := server.GetConfig().GetMinioConfig()
 
 	log.Printf("MinioConfig: %+v", config)
 	return config, nil
@@ -87,14 +89,14 @@ func TestNewMinioStorage(t *testing.T) {
 		t.Error("Expected client to be non-nil")
 	}
 
-	if storage.config == nil {
+	if storage.config.Host == "" {
 		t.Error("Expected config to be non-nil")
 	}
 }
 
 func TestNewMinioStorage_InvalidConfig(t *testing.T) {
 	// Test with invalid host
-	config := &server.MinioConfig{
+	config := configpkg.MinioConfig{
 		Host:     "invalid-host:99999",
 		User:     "test",
 		Password: "test",
@@ -111,7 +113,7 @@ func TestNewMinioStorage_InvalidConfig(t *testing.T) {
 func TestMinioStorage_Health(t *testing.T) {
 	storage := newTestMinioStorage(t)
 
-	healthy := storage.Health()
+	healthy := storage.Health(context.Background())
 	// Health check should return true if connection is working
 	// Note: This depends on whether a default bucket is configured
 	t.Logf("Health check result: %v", healthy)
@@ -128,13 +130,13 @@ func TestMinioStorage_PutAndGet(t *testing.T) {
 	content := []byte("Hello, MinIO Test!")
 
 	// Test Put
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
 	// Test Get
-	retrieved, err := storage.Get(bucket, key)
+	retrieved, err := storage.Get(context.Background(), bucket, key)
 	if err != nil {
 		t.Fatalf("Failed to get object: %v", err)
 	}
@@ -144,7 +146,7 @@ func TestMinioStorage_PutAndGet(t *testing.T) {
 	}
 
 	// Cleanup
-	err = storage.Remove(bucket, key)
+	err = storage.Remove(context.Background(), bucket, key)
 	if err != nil {
 		t.Logf("Warning: failed to cleanup test object: %v", err)
 	}
@@ -157,19 +159,19 @@ func TestMinioStorage_Put_EmptyData(t *testing.T) {
 	key := "empty-file.txt"
 	content := []byte{}
 
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put empty object: %v", err)
 	}
 
 	// Verify object exists
-	exists := storage.ObjExist(bucket, key)
+	exists := storage.ObjExist(context.Background(), bucket, key)
 	if !exists {
 		t.Error("Expected empty object to exist")
 	}
 
 	// Cleanup
-	storage.Remove(bucket, key)
+	storage.Remove(context.Background(), bucket, key)
 }
 
 func TestMinioStorage_Put_LargeData(t *testing.T) {
@@ -183,12 +185,12 @@ func TestMinioStorage_Put_LargeData(t *testing.T) {
 		content[i] = byte(i % 256)
 	}
 
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put large object: %v", err)
 	}
 
-	retrieved, err := storage.Get(bucket, key)
+	retrieved, err := storage.Get(context.Background(), bucket, key)
 	if err != nil {
 		t.Fatalf("Failed to get large object: %v", err)
 	}
@@ -198,7 +200,7 @@ func TestMinioStorage_Put_LargeData(t *testing.T) {
 	}
 
 	// Cleanup
-	storage.Remove(bucket, key)
+	storage.Remove(context.Background(), bucket, key)
 }
 
 func TestMinioStorage_Get_NonExistent(t *testing.T) {
@@ -207,7 +209,7 @@ func TestMinioStorage_Get_NonExistent(t *testing.T) {
 	bucket := "test-bucket"
 	key := "non-existent-file.txt"
 
-	_, err := storage.Get(bucket, key)
+	_, err := storage.Get(context.Background(), bucket, key)
 	if err == nil {
 		t.Error("Expected error when getting non-existent object")
 	}
@@ -221,25 +223,25 @@ func TestMinioStorage_Remove(t *testing.T) {
 	content := []byte("Delete me")
 
 	// First, put an object
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
 	// Verify it exists
-	exists := storage.ObjExist(bucket, key)
+	exists := storage.ObjExist(context.Background(), bucket, key)
 	if !exists {
 		t.Fatal("Expected object to exist before removal")
 	}
 
 	// Remove it
-	err = storage.Remove(bucket, key)
+	err = storage.Remove(context.Background(), bucket, key)
 	if err != nil {
 		t.Fatalf("Failed to remove object: %v", err)
 	}
 
 	// Verify it's gone
-	exists = storage.ObjExist(bucket, key)
+	exists = storage.ObjExist(context.Background(), bucket, key)
 	if exists {
 		t.Error("Expected object to not exist after removal")
 	}
@@ -252,7 +254,7 @@ func TestMinioStorage_Remove_NonExistent(t *testing.T) {
 	key := "non-existent-file.txt"
 
 	// Removing a non-existent object should not error
-	err := storage.Remove(bucket, key)
+	err := storage.Remove(context.Background(), bucket, key)
 	if err != nil {
 		t.Logf("Remove non-existent object returned error (may be acceptable): %v", err)
 	}
@@ -266,25 +268,25 @@ func TestMinioStorage_ObjExist(t *testing.T) {
 	content := []byte("Test content")
 
 	// Check non-existent object
-	exists := storage.ObjExist(bucket, key)
+	exists := storage.ObjExist(context.Background(), bucket, key)
 	if exists {
 		t.Error("Expected non-existent object to return false")
 	}
 
 	// Create object
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
 	// Check existing object
-	exists = storage.ObjExist(bucket, key)
+	exists = storage.ObjExist(context.Background(), bucket, key)
 	if !exists {
 		t.Error("Expected existing object to return true")
 	}
 
 	// Cleanup
-	storage.Remove(bucket, key)
+	storage.Remove(context.Background(), bucket, key)
 }
 
 func TestMinioStorage_GetPresignedURL(t *testing.T) {
@@ -295,13 +297,13 @@ func TestMinioStorage_GetPresignedURL(t *testing.T) {
 	content := []byte("Presigned URL test content")
 
 	// Create object first
-	err := storage.Put(bucket, key, content)
+	err := storage.Put(context.Background(), bucket, key, content)
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
 	// Get presigned URL
-	url, err := storage.GetPresignedURL(bucket, key, 5*time.Minute)
+	url, err := storage.GetPresignedURL(context.Background(), bucket, key, 5*time.Minute)
 	if err != nil {
 		t.Fatalf("Failed to get presigned URL: %v", err)
 	}
@@ -316,7 +318,7 @@ func TestMinioStorage_GetPresignedURL(t *testing.T) {
 	}
 
 	// Cleanup
-	storage.Remove(bucket, key)
+	storage.Remove(context.Background(), bucket, key)
 }
 
 func TestMinioStorage_GetPresignedURL_NonExistent(t *testing.T) {
@@ -325,7 +327,7 @@ func TestMinioStorage_GetPresignedURL_NonExistent(t *testing.T) {
 	bucket := "test-bucket"
 	key := "non-existent-presigned.txt"
 
-	_, err := storage.GetPresignedURL(bucket, key, 5*time.Minute)
+	_, err := storage.GetPresignedURL(context.Background(), bucket, key, 5*time.Minute)
 	if err == nil {
 		t.Log("Note: Some MinIO versions may allow presigned URLs for non-existent objects")
 	}
@@ -337,25 +339,25 @@ func TestMinioStorage_BucketExists(t *testing.T) {
 	bucket := fmt.Sprintf("test-bucket-exists-%d", time.Now().Unix())
 
 	// Check non-existent bucket
-	exists := storage.BucketExists(bucket)
+	exists := storage.BucketExists(context.Background(), bucket)
 	if exists {
 		t.Error("Expected non-existent bucket to return false")
 	}
 
 	// Create bucket by putting an object
-	err := storage.Put(bucket, "test.txt", []byte("test"))
+	err := storage.Put(context.Background(), bucket, "test.txt", []byte("test"))
 	if err != nil {
 		t.Fatalf("Failed to create bucket: %v", err)
 	}
 
 	// Check existing bucket
-	exists = storage.BucketExists(bucket)
+	exists = storage.BucketExists(context.Background(), bucket)
 	if !exists {
 		t.Error("Expected existing bucket to return true")
 	}
 
 	// Cleanup
-	storage.RemoveBucket(bucket)
+	storage.RemoveBucket(context.Background(), bucket)
 }
 
 func TestMinioStorage_RemoveBucket(t *testing.T) {
@@ -364,30 +366,30 @@ func TestMinioStorage_RemoveBucket(t *testing.T) {
 	bucket := fmt.Sprintf("test-bucket-remove-%d", time.Now().Unix())
 
 	// Create bucket with some objects
-	err := storage.Put(bucket, "file1.txt", []byte("content1"))
+	err := storage.Put(context.Background(), bucket, "file1.txt", []byte("content1"))
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
-	err = storage.Put(bucket, "file2.txt", []byte("content2"))
+	err = storage.Put(context.Background(), bucket, "file2.txt", []byte("content2"))
 	if err != nil {
 		t.Fatalf("Failed to put object: %v", err)
 	}
 
 	// Verify bucket exists
-	exists := storage.BucketExists(bucket)
+	exists := storage.BucketExists(context.Background(), bucket)
 	if !exists {
 		t.Fatal("Expected bucket to exist before removal")
 	}
 
 	// Remove bucket
-	err = storage.RemoveBucket(bucket)
+	err = storage.RemoveBucket(context.Background(), bucket)
 	if err != nil {
 		t.Fatalf("Failed to remove bucket: %v", err)
 	}
 
 	// Verify bucket is gone
-	exists = storage.BucketExists(bucket)
+	exists = storage.BucketExists(context.Background(), bucket)
 	if exists {
 		t.Error("Expected bucket to not exist after removal")
 	}
@@ -403,25 +405,25 @@ func TestMinioStorage_Copy(t *testing.T) {
 	content := []byte("Content to copy")
 
 	// Create source object
-	err := storage.Put(srcBucket, srcKey, content)
+	err := storage.Put(context.Background(), srcBucket, srcKey, content)
 	if err != nil {
 		t.Fatalf("Failed to put source object: %v", err)
 	}
 
 	// Copy object
-	success := storage.Copy(srcBucket, srcKey, destBucket, destKey)
+	success := storage.Copy(context.Background(), srcBucket, srcKey, destBucket, destKey)
 	if !success {
 		t.Fatal("Failed to copy object")
 	}
 
 	// Verify destination exists
-	exists := storage.ObjExist(destBucket, destKey)
+	exists := storage.ObjExist(context.Background(), destBucket, destKey)
 	if !exists {
 		t.Error("Expected copied object to exist")
 	}
 
 	// Verify content matches
-	retrieved, err := storage.Get(destBucket, destKey)
+	retrieved, err := storage.Get(context.Background(), destBucket, destKey)
 	if err != nil {
 		t.Fatalf("Failed to get copied object: %v", err)
 	}
@@ -431,8 +433,8 @@ func TestMinioStorage_Copy(t *testing.T) {
 	}
 
 	// Cleanup
-	storage.Remove(srcBucket, srcKey)
-	storage.Remove(destBucket, destKey)
+	storage.Remove(context.Background(), srcBucket, srcKey)
+	storage.Remove(context.Background(), destBucket, destKey)
 }
 
 func TestMinioStorage_Copy_NonExistentSource(t *testing.T) {
@@ -443,16 +445,16 @@ func TestMinioStorage_Copy_NonExistentSource(t *testing.T) {
 	destBucket := "test-bucket-dest"
 	destKey := "should-not-exist.txt"
 
-	success := storage.Copy(srcBucket, srcKey, destBucket, destKey)
+	success := storage.Copy(context.Background(), srcBucket, srcKey, destBucket, destKey)
 	if success {
 		t.Error("Expected copy of non-existent object to fail")
 	}
 
 	// Verify destination does not exist
-	exists := storage.ObjExist(destBucket, destKey)
+	exists := storage.ObjExist(context.Background(), destBucket, destKey)
 	if exists {
 		t.Error("Expected destination object to not exist after failed copy")
-		storage.Remove(destBucket, destKey)
+		storage.Remove(context.Background(), destBucket, destKey)
 	}
 }
 
@@ -466,31 +468,31 @@ func TestMinioStorage_Move(t *testing.T) {
 	content := []byte("Content to move")
 
 	// Create source object
-	err := storage.Put(srcBucket, srcKey, content)
+	err := storage.Put(context.Background(), srcBucket, srcKey, content)
 	if err != nil {
 		t.Fatalf("Failed to put source object: %v", err)
 	}
 
 	// Move object
-	success := storage.Move(srcBucket, srcKey, destBucket, destKey)
+	success := storage.Move(context.Background(), srcBucket, srcKey, destBucket, destKey)
 	if !success {
 		t.Fatal("Failed to move object")
 	}
 
 	// Verify source is gone
-	exists := storage.ObjExist(srcBucket, srcKey)
+	exists := storage.ObjExist(context.Background(), srcBucket, srcKey)
 	if exists {
 		t.Error("Expected source object to not exist after move")
 	}
 
 	// Verify destination exists
-	exists = storage.ObjExist(destBucket, destKey)
+	exists = storage.ObjExist(context.Background(), destBucket, destKey)
 	if !exists {
 		t.Error("Expected moved object to exist")
 	}
 
 	// Verify content matches
-	retrieved, err := storage.Get(destBucket, destKey)
+	retrieved, err := storage.Get(context.Background(), destBucket, destKey)
 	if err != nil {
 		t.Fatalf("Failed to get moved object: %v", err)
 	}
@@ -500,7 +502,7 @@ func TestMinioStorage_Move(t *testing.T) {
 	}
 
 	// Cleanup
-	storage.Remove(destBucket, destKey)
+	storage.Remove(context.Background(), destBucket, destKey)
 }
 
 func TestMinioStorage_Move_NonExistentSource(t *testing.T) {
@@ -511,7 +513,7 @@ func TestMinioStorage_Move_NonExistentSource(t *testing.T) {
 	destBucket := "test-bucket-dest"
 	destKey := "should-not-exist.txt"
 
-	success := storage.Move(srcBucket, srcKey, destBucket, destKey)
+	success := storage.Move(context.Background(), srcBucket, srcKey, destBucket, destKey)
 	if success {
 		t.Error("Expected move of non-existent object to fail")
 	}
@@ -527,7 +529,7 @@ func TestMinioStorage_MultipleObjectsInBucket(t *testing.T) {
 	for i := 0; i < numObjects; i++ {
 		key := fmt.Sprintf("file-%d.txt", i)
 		content := []byte(fmt.Sprintf("Content %d", i))
-		err := storage.Put(bucket, key, content)
+		err := storage.Put(context.Background(), bucket, key, content)
 		if err != nil {
 			t.Fatalf("Failed to put object %d: %v", i, err)
 		}
@@ -536,7 +538,7 @@ func TestMinioStorage_MultipleObjectsInBucket(t *testing.T) {
 	// Verify all objects exist
 	for i := 0; i < numObjects; i++ {
 		key := fmt.Sprintf("file-%d.txt", i)
-		exists := storage.ObjExist(bucket, key)
+		exists := storage.ObjExist(context.Background(), bucket, key)
 		if !exists {
 			t.Errorf("Expected object %s to exist", key)
 		}
@@ -546,7 +548,7 @@ func TestMinioStorage_MultipleObjectsInBucket(t *testing.T) {
 	for i := 0; i < numObjects; i++ {
 		key := fmt.Sprintf("file-%d.txt", i)
 		expectedContent := []byte(fmt.Sprintf("Content %d", i))
-		retrieved, err := storage.Get(bucket, key)
+		retrieved, err := storage.Get(context.Background(), bucket, key)
 		if err != nil {
 			t.Errorf("Failed to get object %s: %v", key, err)
 			continue
@@ -557,7 +559,7 @@ func TestMinioStorage_MultipleObjectsInBucket(t *testing.T) {
 	}
 
 	// Cleanup - remove bucket with all objects
-	err := storage.RemoveBucket(bucket)
+	err := storage.RemoveBucket(context.Background(), bucket)
 	if err != nil {
 		t.Logf("Warning: failed to cleanup bucket: %v", err)
 	}
@@ -579,13 +581,13 @@ func TestMinioStorage_SpecialCharactersInKey(t *testing.T) {
 	for _, key := range specialKeys {
 		content := []byte(fmt.Sprintf("Content for %s", key))
 
-		err := storage.Put(bucket, key, content)
+		err := storage.Put(context.Background(), bucket, key, content)
 		if err != nil {
 			t.Errorf("Failed to put object with key '%s': %v", key, err)
 			continue
 		}
 
-		retrieved, err := storage.Get(bucket, key)
+		retrieved, err := storage.Get(context.Background(), bucket, key)
 		if err != nil {
 			t.Errorf("Failed to get object with key '%s': %v", key, err)
 			continue
@@ -596,7 +598,7 @@ func TestMinioStorage_SpecialCharactersInKey(t *testing.T) {
 		}
 
 		// Cleanup
-		storage.Remove(bucket, key)
+		storage.Remove(context.Background(), bucket, key)
 	}
 }
 
@@ -609,13 +611,13 @@ func TestMinioStorage_TenantID(t *testing.T) {
 	tenantID := "tenant-123"
 
 	// Put with tenant ID
-	err := storage.Put(bucket, key, content, tenantID)
+	err := storage.Put(context.Background(), bucket, key, content, tenantID)
 	if err != nil {
 		t.Fatalf("Failed to put object with tenant ID: %v", err)
 	}
 
 	// Get with tenant ID
-	retrieved, err := storage.Get(bucket, key, tenantID)
+	retrieved, err := storage.Get(context.Background(), bucket, key, tenantID)
 	if err != nil {
 		t.Fatalf("Failed to get object with tenant ID: %v", err)
 	}
@@ -625,11 +627,11 @@ func TestMinioStorage_TenantID(t *testing.T) {
 	}
 
 	// Check existence with tenant ID
-	exists := storage.ObjExist(bucket, key, tenantID)
+	exists := storage.ObjExist(context.Background(), bucket, key, tenantID)
 	if !exists {
 		t.Error("Expected object to exist with tenant ID")
 	}
 
 	// Cleanup
-	storage.Remove(bucket, key, tenantID)
+	storage.Remove(context.Background(), bucket, key, tenantID)
 }

internal/utility/truncate.go

@@ -0,0 +1,35 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package utility
+
+// TruncateRunes returns s truncated to at most max runes, preserving the
+// original bytes when shorter. Truncation counts runes (not bytes), so
+// multi-byte characters are never split. A non-positive max yields the empty
+// string.
+//
+// This is the single shared implementation previously duplicated in
+// internal/ingestion/task/debug_log_sink.go and internal/service/agent_sessions.go.
+func TruncateRunes(s string, max int) string {
+	if max <= 0 {
+		return ""
+	}
+	r := []rune(s)
+	if len(r) <= max {
+		return s
+	}
+	return string(r[:max])
+}

internal/utility/truncate_test.go

@@ -0,0 +1,47 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package utility
+
+import "testing"
+
+func TestTruncateRunes(t *testing.T) {
+	cases := []struct {
+		name string
+		in   string
+		max  int
+		want string
+	}{
+		{"empty", "", 10, ""},
+		{"ascii shorter than max", "hello", 10, "hello"},
+		{"ascii exactly max", "hello", 5, "hello"},
+		{"ascii truncated", "hello world", 5, "hello"},
+		// Multi-byte: each CJK rune is 3 bytes; ensure truncation counts RUNES,
+		// not bytes, so we keep exactly `max` characters, not `max` bytes.
+		{"unicode truncated by rune", "中文测试abc", 4, "中文测试"},
+		{"unicode shorter than max", "中文", 10, "中文"},
+		// Guard: non-positive max yields empty string (safe default).
+		{"max zero", "hello", 0, ""},
+		{"max negative", "hello", -3, ""},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := TruncateRunes(c.in, c.max); got != c.want {
+				t.Errorf("TruncateRunes(%q, %d) = %q, want %q", c.in, c.max, got, c.want)
+			}
+		})
+	}
+}