Fix Windows embedding runtime self-heal

internal/config/logs.go

@@ -7,9 +7,11 @@ import (
 )
 
 const (
-	LogsDir                = "logs"
-	ServerLogFileName      = "csghub-lite.log"
-	LlamaServerLogFileName = "llama-server.log"
+	LogsDir                    = "logs"
+	ServerLogFileName          = "csghub-lite.log"
+	LlamaServerLogFileName     = "llama-server.log"
+	EmbeddingWorkerLogFileName = "embedding-worker.log"
+	DiffusersWorkerLogFileName = "diffusers-worker.log"
 
 	LogStderrEnv          = "CSGHUB_LITE_LOG_STDERR"
 	DisableFileLoggingEnv = "CSGHUB_LITE_DISABLE_FILE_LOGGING"
@@ -31,6 +33,14 @@ func LlamaServerLogPath() (string, error) {
 	return logPath(LlamaServerLogFileName)
 }
 
+func EmbeddingWorkerLogPath() (string, error) {
+	return logPath(EmbeddingWorkerLogFileName)
+}
+
+func DiffusersWorkerLogPath() (string, error) {
+	return logPath(DiffusersWorkerLogFileName)
+}
+
 func LogStderrEnabled() bool {
 	return strings.TrimSpace(os.Getenv(LogStderrEnv)) != "0"
 }

internal/config/logs_test.go

@@ -34,4 +34,20 @@ func TestLogPaths(t *testing.T) {
 	if want := filepath.Join(logDir, LlamaServerLogFileName); llamaLogPath != want {
 		t.Fatalf("LlamaServerLogPath() = %q, want %q", llamaLogPath, want)
 	}
+
+	embeddingLogPath, err := EmbeddingWorkerLogPath()
+	if err != nil {
+		t.Fatalf("EmbeddingWorkerLogPath() error: %v", err)
+	}
+	if want := filepath.Join(logDir, EmbeddingWorkerLogFileName); embeddingLogPath != want {
+		t.Fatalf("EmbeddingWorkerLogPath() = %q, want %q", embeddingLogPath, want)
+	}
+
+	diffusersLogPath, err := DiffusersWorkerLogPath()
+	if err != nil {
+		t.Fatalf("DiffusersWorkerLogPath() error: %v", err)
+	}
+	if want := filepath.Join(logDir, DiffusersWorkerLogFileName); diffusersLogPath != want {
+		t.Fatalf("DiffusersWorkerLogPath() = %q, want %q", diffusersLogPath, want)
+	}
 }

internal/embedding/worker.go

@@ -7,17 +7,20 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"net/http"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/opencsgs/csglite/internal/config"
 	"github.com/opencsgs/csglite/internal/imagegen"
 	"github.com/opencsgs/csglite/internal/inference"
+	"github.com/opencsgs/csglite/internal/logutil"
 )
 
 //go:embed worker/embedding_worker.py
@@ -31,6 +34,8 @@ type PythonEngine struct {
 	exitCh    chan error
 	port      int
 	client    *http.Client
+	logBuf    *logutil.TailWriter
+	logFile   *os.File
 }
 
 func NewPythonEngine(ctx context.Context, modelName, modelDir string, runtimeManager *imagegen.RuntimeManager) (*PythonEngine, error) {
@@ -59,9 +64,27 @@ func NewPythonEngine(ctx context.Context, modelName, modelDir string, runtimeMan
 	storageRoot := storageRootFromModelDir(modelDir)
 	cmd := exec.Command(runtimeManager.PythonPath(), workerPath, "--model-dir", modelDir, "--model-name", modelName, "--port", strconv.Itoa(port), "--storage-root", storageRoot, "--temp-dir", tempDir)
 	cmd.Env = withTempDir(os.Environ(), tempDir)
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
+	logBuf := logutil.NewTailWriter(64 * 1024)
+	stdout := io.Writer(logBuf)
+	stderr := io.Writer(logBuf)
+	var logFile *os.File
+	if config.FileLoggingEnabled() {
+		if path, err := config.EmbeddingWorkerLogPath(); err != nil {
+			log.Printf("warning: could not resolve embedding worker log path: %v", err)
+		} else if file, err := logutil.OpenAppendFile(path); err != nil {
+			log.Printf("warning: could not open embedding worker log file %s: %v", path, err)
+		} else {
+			logFile = file
+			stdout = io.MultiWriter(stdout, file)
+			stderr = io.MultiWriter(stderr, file)
+		}
+	}
+	cmd.Stdout = stdout
+	cmd.Stderr = stderr
 	if err := cmd.Start(); err != nil {
+		if logFile != nil {
+			_ = logFile.Close()
+		}
 		return nil, fmt.Errorf("starting embedding worker: %w", err)
 	}
 	exitCh := make(chan error, 1)
@@ -77,6 +100,8 @@ func NewPythonEngine(ctx context.Context, modelName, modelDir string, runtimeMan
 		exitCh:    exitCh,
 		port:      port,
 		client:    &http.Client{Timeout: 30 * time.Minute},
+		logBuf:    logBuf,
+		logFile:   logFile,
 	}
 	if err := engine.waitReady(ctx); err != nil {
 		_ = engine.Close()
@@ -127,6 +152,10 @@ func (e *PythonEngine) Close() error {
 		case <-time.After(5 * time.Second):
 		}
 	}
+	if e.logFile != nil {
+		_ = e.logFile.Close()
+		e.logFile = nil
+	}
 	return nil
 }
 
@@ -142,9 +171,9 @@ func (e *PythonEngine) waitReady(ctx context.Context) error {
 			return ctx.Err()
 		case err := <-e.exitCh:
 			if err != nil {
-				return fmt.Errorf("embedding worker exited before becoming ready: %w", err)
+				return fmt.Errorf("%s", e.workerStartError("embedding worker exited before becoming ready: "+err.Error()))
 			}
-			return fmt.Errorf("embedding worker exited before becoming ready")
+			return fmt.Errorf("%s", e.workerStartError("embedding worker exited before becoming ready"))
 		default:
 		}
 		resp, err := e.client.Get(e.url("/health"))
@@ -156,13 +185,22 @@ func (e *PythonEngine) waitReady(ctx context.Context) error {
 		}
 		time.Sleep(500 * time.Millisecond)
 	}
-	return fmt.Errorf("timeout waiting for embedding worker")
+	return fmt.Errorf("%s", e.workerStartError("timeout waiting for embedding worker"))
 }
 
 func (e *PythonEngine) url(path string) string {
 	return fmt.Sprintf("http://127.0.0.1:%d%s", e.port, path)
 }
 
+func (e *PythonEngine) workerStartError(msg string) string {
+	if e.logBuf != nil {
+		if tail := strings.TrimSpace(e.logBuf.String()); tail != "" {
+			msg += "\n\nembedding worker output:\n" + tail
+		}
+	}
+	return msg
+}
+
 func writeEmbeddingWorkerScript(runtimeDir string) error {
 	if err := os.MkdirAll(runtimeDir, 0o755); err != nil {
 		return err

internal/embedding/worker_test.go

@@ -0,0 +1,40 @@
+package embedding
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/opencsgs/csglite/internal/logutil"
+)
+
+func TestWaitReadyIncludesWorkerOutputTail(t *testing.T) {
+	logBuf := logutil.NewTailWriter(64 * 1024)
+	if _, err := logBuf.Write([]byte("Traceback: libtorchaudio.pyd failed to load")); err != nil {
+		t.Fatal(err)
+	}
+	exitCh := make(chan error, 1)
+	exitCh <- errors.New("exit status 1")
+	close(exitCh)
+
+	engine := &PythonEngine{
+		exitCh: exitCh,
+		port:   1,
+		client: &http.Client{Timeout: time.Millisecond},
+		logBuf: logBuf,
+	}
+	err := engine.waitReady(context.Background())
+	if err == nil {
+		t.Fatal("waitReady returned nil")
+	}
+	msg := err.Error()
+	if !strings.Contains(msg, "embedding worker exited before becoming ready") {
+		t.Fatalf("waitReady error missing exit context: %q", msg)
+	}
+	if !strings.Contains(msg, "Traceback: libtorchaudio.pyd failed to load") {
+		t.Fatalf("waitReady error missing worker output tail: %q", msg)
+	}
+}