diff --git a/apps/api/internal/runner/direct_oss_test.go b/apps/api/internal/runner/direct_oss_test.go
index 05e8a39..20330d8 100644
--- a/apps/api/internal/runner/direct_oss_test.go
+++ b/apps/api/internal/runner/direct_oss_test.go
@@ -126,3 +126,51 @@ func TestDirectOSSDoesNotReplaceGeneralUploadScene(t *testing.T) {
 		t.Fatalf("general upload channel=%+v", channel)
 	}
 }
+
+func TestGeneratedInlineAssetUsesDirectOSS(t *testing.T) {
+	var calls atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		calls.Add(1)
+		if request.Method != http.MethodPut {
+			t.Errorf("method=%s, want PUT", request.Method)
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	cfg := config.Config{
+		MediaOSSDirectEnabled:   true,
+		MediaOSSEndpoint:        server.URL,
+		MediaOSSBucket:          "media-bucket",
+		MediaOSSAccessKeyID:     "access-key",
+		MediaOSSAccessKeySecret: "access-secret",
+		MediaOSSPublicBaseURL:   "https://cdn.example.com",
+		MediaOSSObjectPrefix:    "easyai-ai-gateway/production/media",
+	}
+	service := &Service{cfg: cfg, directOSS: newDirectOSSUploader(cfg)}
+	upload, contentType, kind, strategy, err := service.uploadGeneratedAsset(
+		context.Background(),
+		"task-direct-oss",
+		&generatedInlineAsset{
+			Bytes:       []byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a, 0x1a, 0x0a},
+			ContentType: "image/png",
+			Kind:        "image",
+			SourceKey:   "b64_json",
+		},
+		0,
+		[]store.FileStorageChannel{{ID: "unused-channel"}},
+	)
+	if err != nil {
+		t.Fatalf("direct generated upload: %v", err)
+	}
+	if calls.Load() != 1 {
+		t.Fatalf("direct OSS calls=%d, want 1", calls.Load())
+	}
+	if contentType != "image/png" || kind != "image" || strategy != "direct_aliyun_oss" {
+		t.Fatalf("generated metadata=%s/%s/%s", contentType, kind, strategy)
+	}
+	channel, _ := upload["storageChannel"].(map[string]any)
+	if stringFromAny(channel["provider"]) != "aliyun_oss_direct" {
+		t.Fatalf("storage channel=%+v", channel)
+	}
+}
diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go
index 103e47e..486678b 100644
--- a/apps/api/internal/runner/service.go
+++ b/apps/api/internal/runner/service.go
@@ -952,6 +952,14 @@ candidatesLoop:
 				}, isSimulation(task, candidate)); err != nil {
 					return Result{}, err
 				}
+				if task.AsyncMode {
+					// Async callers hydrate the canonical stored result through
+					// the task detail endpoint. The River Worker does not consume
+					// Result.Output, so downloading and rebuilding Base64 here
+					// only retains the execution slot and media memory a second
+					// time after the task is already durably complete.
+					return Result{Task: finished, Output: response.Result}, nil
+				}
 				output, hydrateErr := s.HydrateTaskResult(ctx, task.ID, response.Result)
 				if hydrateErr != nil {
 					return Result{Task: finished}, hydrateErr
@@ -2105,23 +2113,34 @@ func (s *Service) withAttemptHistory(ctx context.Context, taskID string, metrics
 }
 
 func (s *Service) emit(ctx context.Context, taskID string, eventType string, status string, phase string, progress float64, message string, payload map[string]any, simulated bool) error {
-	event, err := s.store.AddTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated)
+	acceptanceURL, acceptance, callbackErr := s.store.AcceptanceCallbackURL(ctx, taskID)
+	if callbackErr != nil {
+		return callbackErr
+	}
+	callbackURL := ""
+	if acceptance && acceptanceURL != "" {
+		callbackURL = acceptanceURL
+	} else if s.cfg.TaskProgressCallbackEnabled && s.cfg.TaskProgressCallbackURL != "" {
+		callbackURL = s.cfg.TaskProgressCallbackURL
+	}
+	event, err := s.store.AddTaskEventWithCallback(
+		ctx,
+		taskID,
+		eventType,
+		status,
+		phase,
+		progress,
+		message,
+		payload,
+		simulated,
+		callbackURL,
+	)
 	if err != nil {
 		return err
 	}
 	if event.ID == "" {
 		s.observeTaskEventSkip(event.SkippedReason)
 	}
-	acceptanceURL, acceptance, callbackErr := s.store.AcceptanceCallbackURL(ctx, taskID)
-	if callbackErr != nil {
-		return callbackErr
-	}
-	if acceptance && acceptanceURL != "" {
-		return s.store.QueueTaskCallback(ctx, event, acceptanceURL)
-	}
-	if s.cfg.TaskProgressCallbackEnabled && s.cfg.TaskProgressCallbackURL != "" {
-		return s.store.QueueTaskCallback(ctx, event, s.cfg.TaskProgressCallbackURL)
-	}
 	return nil
 }
 
diff --git a/apps/api/internal/runner/upload.go b/apps/api/internal/runner/upload.go
index a5e97bf..2ef1b34 100644
--- a/apps/api/internal/runner/upload.go
+++ b/apps/api/internal/runner/upload.go
@@ -167,9 +167,11 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task
 	var channels []store.FileStorageChannel
 	channelsLoaded := false
 	if needsUpload || rawNeedsUpload {
-		channels, err = s.activeFileStorageChannels(ctx, store.FileStorageSceneImageResult)
-		if err != nil {
-			return nil, &clients.ClientError{Code: "upload_config_failed", Message: err.Error(), Retryable: true}
+		if s.directOSS == nil {
+			channels, err = s.activeFileStorageChannels(ctx, store.FileStorageSceneImageResult)
+			if err != nil {
+				return nil, &clients.ClientError{Code: "upload_config_failed", Message: err.Error(), Retryable: true}
+			}
 		}
 		channelsLoaded = true
 	}
@@ -661,6 +663,10 @@ func (s *Service) uploadGeneratedAsset(ctx context.Context, taskID string, asset
 		Scene:       store.FileStorageSceneImageResult,
 		Source:      "ai-gateway",
 	}
+	if s.directOSS != nil {
+		upload, err := s.directOSS.upload(ctx, payload)
+		return upload, contentType, kind, "direct_aliyun_oss", err
+	}
 	if len(channels) == 0 {
 		upload, err := s.storeFileLocally(payload, s.cfg.LocalGeneratedStorageDir, config.DefaultLocalGeneratedStorageDir, localStaticGeneratedPathPrefix)
 		return upload, contentType, kind, "local_static_inline_media", err
@@ -683,6 +689,10 @@ func (s *Service) uploadGeneratedURLAsset(ctx context.Context, taskID string, as
 		Scene:       store.FileStorageSceneImageResult,
 		Source:      "ai-gateway",
 	}
+	if s.directOSS != nil {
+		upload, err := s.directOSS.upload(ctx, uploadPayload)
+		return upload, contentType, kind, "direct_aliyun_oss", err
+	}
 	if len(channels) == 0 {
 		upload, err := s.storeFileLocally(uploadPayload, s.cfg.LocalGeneratedStorageDir, config.DefaultLocalGeneratedStorageDir, localStaticGeneratedPathPrefix)
 		return upload, contentType, kind, "local_static_url_media", err
diff --git a/apps/api/internal/store/file_storage_channels.go b/apps/api/internal/store/file_storage_channels.go
index e8c3987..041bac7 100644
--- a/apps/api/internal/store/file_storage_channels.go
+++ b/apps/api/internal/store/file_storage_channels.go
@@ -237,28 +237,48 @@ func (s *Store) MarkFileStorageChannelFailure(ctx context.Context, id string, me
 	if strings.TrimSpace(id) == "" {
 		return nil
 	}
-	_, err := s.pool.Exec(ctx, `
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer rollbackTransaction(tx)
+	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `
 UPDATE file_storage_channels
 SET last_error = NULLIF($2, ''),
     last_failed_at = now(),
     updated_at = now()
 WHERE id = $1::uuid
-  AND deleted_at IS NULL`, id, strings.TrimSpace(message))
-	return err
+  AND deleted_at IS NULL`, id, strings.TrimSpace(message)); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
 }
 
 func (s *Store) MarkFileStorageChannelSuccess(ctx context.Context, id string) error {
 	if strings.TrimSpace(id) == "" {
 		return nil
 	}
-	_, err := s.pool.Exec(ctx, `
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer rollbackTransaction(tx)
+	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `
 UPDATE file_storage_channels
 SET last_error = NULL,
     last_succeeded_at = now(),
     updated_at = now()
 WHERE id = $1::uuid
-  AND deleted_at IS NULL`, id)
-	return err
+  AND deleted_at IS NULL`, id); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
 }
 
 func scanFileStorageChannel(scanner fileStorageChannelScanner) (FileStorageChannel, error) {
diff --git a/apps/api/internal/store/task_history_test.go b/apps/api/internal/store/task_history_test.go
index a954f6c..6f2cc1a 100644
--- a/apps/api/internal/store/task_history_test.go
+++ b/apps/api/internal/store/task_history_test.go
@@ -63,13 +63,21 @@ func TestTaskEventsAreMinimalAndConsecutiveDuplicatesAreSkipped(t *testing.T) {
 	if err != nil || synthetic.ID != "" || synthetic.SkippedReason != "unknown_type" {
 		t.Fatalf("synthetic=%+v err=%v", synthetic, err)
 	}
-	completed, err := db.AddTaskEvent(ctx, task.ID, "task.completed", "succeeded", "completed", 1, "ignored", map[string]any{"result": map[string]any{"duplicate": true}}, true)
+	completed, err := db.AddTaskEventWithCallback(
+		ctx,
+		task.ID,
+		"task.completed",
+		"succeeded",
+		"completed",
+		1,
+		"ignored",
+		map[string]any{"result": map[string]any{"duplicate": true}},
+		true,
+		"https://callback.invalid/task",
+	)
 	if err != nil || completed.ID == "" || len(completed.Payload) != 0 {
 		t.Fatalf("completed=%+v err=%v", completed, err)
 	}
-	if err := db.QueueTaskCallback(ctx, completed, "https://callback.invalid/task"); err != nil {
-		t.Fatal(err)
-	}
 
 	var events int
 	var nonEmptyPayloads int
diff --git a/apps/api/internal/store/tasks_runtime.go b/apps/api/internal/store/tasks_runtime.go
index 47c9df9..8972a49 100644
--- a/apps/api/internal/store/tasks_runtime.go
+++ b/apps/api/internal/store/tasks_runtime.go
@@ -2049,6 +2049,40 @@ func taskEventPlatformID(payload map[string]any) string {
 }
 
 func (s *Store) AddTaskEvent(ctx context.Context, taskID string, eventType string, status string, phase string, progress float64, message string, payload map[string]any, simulated bool) (TaskEvent, error) {
+	return s.addTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated, "")
+}
+
+// AddTaskEventWithCallback persists the event and its callback outbox item in
+// one transaction. This closes the crash window between event creation and
+// callback registration and avoids a second synchronous-replication commit for
+// every emitted task event.
+func (s *Store) AddTaskEventWithCallback(
+	ctx context.Context,
+	taskID string,
+	eventType string,
+	status string,
+	phase string,
+	progress float64,
+	message string,
+	payload map[string]any,
+	simulated bool,
+	callbackURL string,
+) (TaskEvent, error) {
+	return s.addTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated, callbackURL)
+}
+
+func (s *Store) addTaskEvent(
+	ctx context.Context,
+	taskID string,
+	eventType string,
+	status string,
+	phase string,
+	progress float64,
+	message string,
+	payload map[string]any,
+	simulated bool,
+	callbackURL string,
+) (TaskEvent, error) {
 	eventType = strings.TrimSpace(eventType)
 	if !taskEventAllowed(eventType) {
 		return TaskEvent{SkippedReason: "unknown_type"}, nil
@@ -2142,6 +2176,20 @@ RETURNING id::text, task_id::text, seq, event_type, COALESCE(status, ''), COALES
 		return TaskEvent{}, err
 	}
 	event.Payload = decodeObject(payloadBytes)
+	callbackURL = strings.TrimSpace(callbackURL)
+	if callbackURL != "" {
+		if _, err := tx.Exec(ctx, `
+INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
+VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
+ON CONFLICT (task_id, seq, callback_url) DO NOTHING`,
+			event.TaskID,
+			event.ID,
+			event.Seq,
+			callbackURL,
+		); err != nil {
+			return TaskEvent{}, err
+		}
+	}
 	if err := tx.Commit(ctx); err != nil {
 		return TaskEvent{}, err
 	}