Add runtime restore and temp asset cleanup

apps/api/internal/runner/helpers.go

@@ -11,8 +11,18 @@ func stringFromMap(values map[string]any, key string) string {
 }
 
 func stringFromAny(value any) string {
-	text, _ := value.(string)
-	return strings.TrimSpace(text)
+	switch typed := value.(type) {
+	case string:
+		return strings.TrimSpace(typed)
+	case map[string]any:
+		if text := stringFromAny(typed["url"]); text != "" {
+			return text
+		}
+		if ref, ok := typed["assetRef"].(map[string]any); ok {
+			return stringFromAny(ref["url"])
+		}
+	}
+	return ""
 }
 
 func boolFromMap(values map[string]any, key string) bool {

apps/api/internal/runner/request_assets.go

@@ -0,0 +1,249 @@
+package runner
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+func (s *Service) restoreTaskRequestReferences(ctx context.Context, task store.GatewayTask) (map[string]any, error) {
+	body := cloneMap(task.Request)
+	if body["messages"] != nil || body["messageRefs"] == nil || s.store == nil {
+		return body, nil
+	}
+	refs, err := s.store.ListTaskConversationMessages(ctx, task.ID)
+	if err != nil {
+		return nil, err
+	}
+	if len(refs) == 0 {
+		return body, nil
+	}
+	messages := make([]any, 0, len(refs))
+	for _, ref := range refs {
+		messages = append(messages, ref.Message)
+	}
+	body["messages"] = messages
+	return body, nil
+}
+
+func (s *Service) slimTaskRequestSnapshot(task store.GatewayTask, body map[string]any) map[string]any {
+	out := cloneMap(body)
+	messageRefs := task.Request["messageRefs"]
+	if messageRefs == nil {
+		return out
+	}
+	delete(out, "messages")
+	out["messageRefs"] = messageRefs
+	for _, key := range []string{"conversationId", "conversationRecordId", "newMessageCount"} {
+		if value := task.Request[key]; value != nil {
+			out[key] = value
+		}
+	}
+	return out
+}
+
+func (s *Service) slimParameterPreprocessingLog(task store.GatewayTask, log parameterPreprocessingLog) parameterPreprocessingLog {
+	log.Input = s.slimTaskRequestSnapshot(task, log.Input)
+	log.Output = s.slimTaskRequestSnapshot(task, log.Output)
+	return log
+}
+
+func (s *Service) hydrateProviderRequestAssets(ctx context.Context, body map[string]any) (map[string]any, error) {
+	value, err := s.hydrateProviderRequestAssetValue(ctx, body, nil)
+	if err != nil {
+		return nil, err
+	}
+	out, _ := value.(map[string]any)
+	if out == nil {
+		return map[string]any{}, nil
+	}
+	return out, nil
+}
+
+func (s *Service) hydrateProviderRequestAssetValue(ctx context.Context, value any, path []string) (any, error) {
+	switch typed := value.(type) {
+	case map[string]any:
+		if ref, ok := typed["assetRef"].(map[string]any); ok {
+			return s.hydrateProviderRequestAssetRef(ctx, ref, path)
+		}
+		next := make(map[string]any, len(typed))
+		for key, item := range typed {
+			hydrated, err := s.hydrateProviderRequestAssetValue(ctx, item, append(path, key))
+			if err != nil {
+				return nil, err
+			}
+			next[key] = hydrated
+		}
+		return next, nil
+	case []any:
+		next := make([]any, 0, len(typed))
+		for index, item := range typed {
+			hydrated, err := s.hydrateProviderRequestAssetValue(ctx, item, append(path, fmt.Sprintf("[%d]", index)))
+			if err != nil {
+				return nil, err
+			}
+			next = append(next, hydrated)
+		}
+		return next, nil
+	default:
+		return value, nil
+	}
+}
+
+func (s *Service) hydrateProviderRequestAssetRef(ctx context.Context, ref map[string]any, path []string) (any, error) {
+	asset, err := s.resolveRequestAsset(ctx, ref)
+	if err != nil {
+		return nil, err
+	}
+	if providerFieldNeedsBase64(path) {
+		payload, err := s.readRequestAssetBytes(ctx, asset)
+		if err != nil {
+			return nil, err
+		}
+		return base64.StdEncoding.EncodeToString(payload), nil
+	}
+	if strings.TrimSpace(asset.URL) == "" {
+		return nil, requestAssetExpiredError(asset)
+	}
+	return asset.URL, nil
+}
+
+func (s *Service) resolveRequestAsset(ctx context.Context, ref map[string]any) (store.RequestAsset, error) {
+	sha := stringFromAny(ref["sha256"])
+	contentType := stringFromAny(ref["contentType"])
+	asset := store.RequestAsset{
+		SHA256:          sha,
+		ContentType:     contentType,
+		URL:             stringFromAny(ref["url"]),
+		StorageProvider: stringFromAny(ref["storageProvider"]),
+	}
+	if size := floatFromAny(ref["size"]); size > 0 {
+		asset.ByteSize = int64(size)
+	}
+	if expiresAt := stringFromAny(ref["expiresAt"]); expiresAt != "" {
+		if parsed, err := time.Parse(time.RFC3339, expiresAt); err == nil {
+			asset.ExpiresAt = &parsed
+		}
+	}
+	if s.store != nil && sha != "" && contentType != "" {
+		if stored, ok, err := s.store.FindRequestAsset(ctx, sha, contentType); err != nil && !store.IsUndefinedDatabaseObject(err) {
+			return store.RequestAsset{}, err
+		} else if ok {
+			asset = stored
+		}
+	}
+	if requestAssetIsExpired(asset, time.Now()) {
+		return store.RequestAsset{}, requestAssetExpiredError(asset)
+	}
+	return asset, nil
+}
+
+func (s *Service) readRequestAssetBytes(ctx context.Context, asset store.RequestAsset) ([]byte, error) {
+	if requestAssetIsExpired(asset, time.Now()) {
+		return nil, requestAssetExpiredError(asset)
+	}
+	if strings.TrimSpace(asset.LocalPath) != "" {
+		payload, err := os.ReadFile(asset.LocalPath)
+		if err != nil {
+			return nil, requestAssetExpiredError(asset)
+		}
+		return payload, nil
+	}
+	if localPath := s.localPathFromRequestAssetURL(asset.URL); localPath != "" {
+		payload, err := os.ReadFile(localPath)
+		if err != nil {
+			return nil, requestAssetExpiredError(asset)
+		}
+		return payload, nil
+	}
+	if strings.HasPrefix(asset.URL, "http://") || strings.HasPrefix(asset.URL, "https://") {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, asset.URL, nil)
+		if err != nil {
+			return nil, err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return nil, &clients.ClientError{Code: "request_asset_fetch_failed", Message: err.Error(), Retryable: true}
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+			return nil, &clients.ClientError{Code: "request_asset_fetch_failed", Message: resp.Status, StatusCode: resp.StatusCode, Retryable: clients.HTTPRetryable(resp.StatusCode)}
+		}
+		payload, err := io.ReadAll(io.LimitReader(resp.Body, 256<<20))
+		if err != nil {
+			return nil, &clients.ClientError{Code: "request_asset_fetch_failed", Message: err.Error(), Retryable: true}
+		}
+		return payload, nil
+	}
+	return nil, requestAssetExpiredError(asset)
+}
+
+func (s *Service) localPathFromRequestAssetURL(value string) string {
+	raw := strings.TrimSpace(value)
+	if raw == "" {
+		return ""
+	}
+	pathValue := raw
+	if parsed, err := url.Parse(raw); err == nil && parsed.Path != "" {
+		pathValue = parsed.Path
+	}
+	const uploadedPrefix = "/static/uploaded/"
+	if !strings.HasPrefix(pathValue, uploadedPrefix) {
+		return ""
+	}
+	fileName := filepath.Base(strings.TrimPrefix(pathValue, uploadedPrefix))
+	if !strings.HasPrefix(fileName, "gateway-request-asset-") {
+		return ""
+	}
+	storageDir := strings.TrimSpace(s.cfg.LocalUploadedStorageDir)
+	if storageDir == "" {
+		storageDir = config.DefaultLocalUploadedStorageDir
+	}
+	return filepath.Join(storageDir, fileName)
+}
+
+func providerFieldNeedsBase64(path []string) bool {
+	if len(path) == 0 {
+		return false
+	}
+	key := strings.ToLower(strings.Trim(path[len(path)-1], "[]"))
+	parent := ""
+	if len(path) > 1 {
+		parent = strings.ToLower(strings.Trim(path[len(path)-2], "[]"))
+	}
+	return key == "b64_json" ||
+		key == "base64" ||
+		key == "b64" ||
+		strings.Contains(key, "base64") ||
+		strings.Contains(key, "_b64") ||
+		(parent == "input_audio" && key == "data")
+}
+
+func requestAssetIsExpired(asset store.RequestAsset, now time.Time) bool {
+	if asset.ExpiredAt != nil {
+		return true
+	}
+	if asset.ExpiresAt != nil && !asset.ExpiresAt.After(now) {
+		return true
+	}
+	return false
+}
+
+func requestAssetExpiredError(asset store.RequestAsset) error {
+	message := "request asset is expired or unavailable"
+	if asset.SHA256 != "" {
+		message = "request asset is expired or unavailable: " + asset.SHA256
+	}
+	return &clients.ClientError{Code: "request_asset_expired", Message: message, Retryable: false}
+}

apps/api/internal/runner/request_assets_test.go

@@ -0,0 +1,101 @@
+package runner
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+func TestHydrateProviderRequestAssetsConvertsStrictBase64Field(t *testing.T) {
+	storageDir := t.TempDir()
+	fileName := "gateway-request-asset-test.png"
+	if err := os.WriteFile(filepath.Join(storageDir, fileName), []byte("image bytes"), 0o644); err != nil {
+		t.Fatalf("write request asset: %v", err)
+	}
+	service := &Service{cfg: config.Config{LocalUploadedStorageDir: storageDir}}
+	body := map[string]any{
+		"model": "demo",
+		"b64_json": map[string]any{
+			"assetRef": map[string]any{
+				"sha256":          "sha-test",
+				"contentType":     "image/png",
+				"url":             "/static/uploaded/" + fileName,
+				"storageProvider": "local_static",
+			},
+			"url": "/static/uploaded/" + fileName,
+		},
+	}
+
+	hydrated, err := service.hydrateProviderRequestAssets(context.Background(), body)
+	if err != nil {
+		t.Fatalf("hydrate request assets: %v", err)
+	}
+	if got, want := stringFromAny(hydrated["b64_json"]), base64.StdEncoding.EncodeToString([]byte("image bytes")); got != want {
+		t.Fatalf("unexpected hydrated base64: got %q want %q", got, want)
+	}
+}
+
+func TestHydrateProviderRequestAssetsReturnsExpiredError(t *testing.T) {
+	expiredAt := time.Now().Add(-time.Minute).UTC().Format(time.RFC3339)
+	service := &Service{}
+	body := map[string]any{
+		"image_url": map[string]any{
+			"assetRef": map[string]any{
+				"sha256":          "sha-expired",
+				"contentType":     "image/png",
+				"url":             "/static/uploaded/gateway-request-asset-expired.png",
+				"storageProvider": "local_static",
+				"expiresAt":       expiredAt,
+			},
+			"url": "/static/uploaded/gateway-request-asset-expired.png",
+		},
+	}
+
+	_, err := service.hydrateProviderRequestAssets(context.Background(), body)
+	if err == nil {
+		t.Fatal("expected expired request asset error")
+	}
+	var clientErr *clients.ClientError
+	if !errors.As(err, &clientErr) || clientErr.Code != "request_asset_expired" {
+		t.Fatalf("expected request_asset_expired, got %T %v", err, err)
+	}
+}
+
+func TestStringFromAnyReadsRequestAssetWrapperURL(t *testing.T) {
+	wrapper := map[string]any{
+		"assetRef": map[string]any{"url": "https://cdn.example/request.png"},
+	}
+	if got := stringFromAny(wrapper); got != "https://cdn.example/request.png" {
+		t.Fatalf("expected wrapper URL, got %q", got)
+	}
+}
+
+func TestSlimTaskRequestSnapshotKeepsMessageRefs(t *testing.T) {
+	service := &Service{}
+	task := store.GatewayTask{Request: map[string]any{
+		"conversationId":  "conv-1",
+		"messageRefs":     []any{map[string]any{"messageId": "msg-1", "position": 0}},
+		"newMessageCount": 1,
+	}}
+	body := map[string]any{
+		"model":    "demo",
+		"messages": []any{map[string]any{"role": "user", "content": "hello"}},
+	}
+
+	snapshot := service.slimTaskRequestSnapshot(task, body)
+
+	if snapshot["messages"] != nil {
+		t.Fatalf("snapshot should not persist restored messages: %+v", snapshot)
+	}
+	if snapshot["messageRefs"] == nil || snapshot["newMessageCount"] != 1 {
+		t.Fatalf("snapshot should keep message refs and new count: %+v", snapshot)
+	}
+}

apps/api/internal/runner/service.go

@@ -87,9 +87,13 @@ func (s *Service) ExecuteStream(ctx context.Context, task store.GatewayTask, use
 
 func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta) (Result, error) {
 	executeStartedAt := time.Now()
-	body := normalizeRequest(task.Kind, task.Request)
+	restoredRequest, err := s.restoreTaskRequestReferences(ctx, task)
+	if err != nil {
+		return Result{}, err
+	}
+	body := normalizeRequest(task.Kind, restoredRequest)
 	modelType := modelTypeFromKind(task.Kind, body)
-	if err := s.store.MarkTaskRunning(ctx, task.ID, modelType, body); err != nil {
+	if err := s.store.MarkTaskRunning(ctx, task.ID, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
 		return Result{}, err
 	}
 	if task.Status != "running" {
@@ -194,7 +198,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			}
 			return Result{Task: failed, Output: failed.Result}, clientErr
 		}
-		if err := s.store.MarkTaskRunning(ctx, task.ID, candidates[0].ModelType, firstCandidateBody); err != nil {
+		if err := s.store.MarkTaskRunning(ctx, task.ID, candidates[0].ModelType, s.slimTaskRequestSnapshot(task, firstCandidateBody)); err != nil {
 			return Result{}, err
 		}
 		estimatedBillings := s.estimatedBillings(ctx, user, task.Kind, firstCandidateBody, candidates[0])
@@ -508,13 +512,13 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		QueueKey:        candidate.QueueKey,
 		Status:          "running",
 		Simulated:       simulated,
-		RequestSnapshot: body,
+		RequestSnapshot: s.slimTaskRequestSnapshot(task, body),
 		Metrics:         baseAttemptMetrics,
 	})
 	if err != nil {
 		return clients.Response{}, fmt.Errorf("create task attempt: %w", err)
 	}
-	if err := s.recordTaskParameterPreprocessing(ctx, task.ID, attemptID, attemptNo, candidate, preprocessing); err != nil {
+	if err := s.recordTaskParameterPreprocessing(ctx, task.ID, attemptID, attemptNo, candidate, s.slimParameterPreprocessingLog(task, preprocessing)); err != nil {
 		clientErr := &clients.ClientError{Code: "runtime_error", Message: err.Error(), Retryable: false}
 		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
 			AttemptID:    attemptID,
@@ -560,12 +564,24 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		return clients.Response{}, fmt.Errorf("prepare http client: %w", err)
 	}
 	client := s.clientFor(candidate, simulated)
+	providerBody, err := s.hydrateProviderRequestAssets(ctx, body)
+	if err != nil {
+		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
+			AttemptID:    attemptID,
+			Status:       "failed",
+			Retryable:    false,
+			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
+			ErrorCode:    clients.ErrorCode(err),
+			ErrorMessage: err.Error(),
+		})
+		return clients.Response{}, err
+	}
 	callStartedAt := time.Now()
 	response, err := client.Run(ctx, clients.Request{
 		Kind:              task.Kind,
 		ModelType:         candidate.ModelType,
 		Model:             task.Model,
-		Body:              body,
+		Body:              providerBody,
 		Candidate:         candidate,
 		HTTPClient:        requestHTTPClient,
 		RemoteTaskID:      task.RemoteTaskID,
@@ -576,7 +592,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			}
 			return s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, attemptID, remoteTaskID, payload)
 		},
-		Stream:      boolFromMap(body, "stream"),
+		Stream:      boolFromMap(providerBody, "stream"),
 		StreamDelta: onDelta,
 	})
 	callFinishedAt := time.Now()
@@ -826,7 +842,7 @@ func (s *Service) recordFailedAttempt(ctx context.Context, input failedAttemptRe
 		QueueKey:        queueKey,
 		Status:          "running",
 		Simulated:       input.Simulated,
-		RequestSnapshot: input.Body,
+		RequestSnapshot: s.slimTaskRequestSnapshot(input.Task, input.Body),
 		Metrics:         metrics,
 	})
 	if err != nil {
@@ -834,7 +850,8 @@ func (s *Service) recordFailedAttempt(ctx context.Context, input failedAttemptRe
 		return attemptNo
 	}
 	if input.Preprocessing != nil && input.Candidate != nil {
-		if err := s.recordTaskParameterPreprocessing(ctx, input.Task.ID, attemptID, attemptNo, *input.Candidate, *input.Preprocessing); err != nil {
+		preprocessing := s.slimParameterPreprocessingLog(input.Task, *input.Preprocessing)
+		if err := s.recordTaskParameterPreprocessing(ctx, input.Task.ID, attemptID, attemptNo, *input.Candidate, preprocessing); err != nil {
 			s.logger.Warn("record failed attempt parameter preprocessing failed", "taskID", input.Task.ID, "attempt", attemptNo, "error", err)
 		}
 	}