1

server-rs/crates/api-server/src/big_fish.rs

@@ -1,13 +1,17 @@
 use std::{
     collections::BTreeMap,
+    convert::Infallible,
     time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
 use axum::{
     Json,
     extract::{Extension, Path, State, rejection::JsonRejection},
-    http::{HeaderName, StatusCode, header},
-    response::{IntoResponse, Response},
+    http::StatusCode,
+    response::{
+        IntoResponse, Response,
+        sse::{Event, Sse},
+    },
 };
 use module_assets::{
     AssetObjectAccessPolicy, AssetObjectFieldError, build_asset_entity_binding_input,
@@ -323,93 +327,107 @@ pub async fn stream_big_fish_message(
             big_fish_error_response(&request_context, map_big_fish_client_error(error))
         })?;
     let quick_fill_requested = payload.quick_fill_requested.unwrap_or(false);
-    let mut draft_writer = AiGenerationDraftWriter::new(AiGenerationDraftContext::new(
-        "big_fish",
-        owner_user_id.as_str(),
-        session_id.as_str(),
-        payload.client_message_id.as_str(),
-        "大鱼吃小鱼模板生成草稿",
-    ));
-    if let Err(error) = draft_writer.ensure_started(state.spacetime_client()).await {
-        tracing::warn!(error = %error, "大鱼吃小鱼模板生成草稿任务启动失败，主生成流程继续执行");
-    }
-    let draft_sink = AiGenerationDraftSink::new(
-        AiGenerationDraftContext::new(
+    let state = state.clone();
+    let session_id_for_stream = session_id.clone();
+    let owner_user_id_for_stream = owner_user_id.clone();
+    let client_message_id_for_stream = payload.client_message_id.clone();
+    let stream = async_stream::stream! {
+        let mut draft_writer = AiGenerationDraftWriter::new(AiGenerationDraftContext::new(
             "big_fish",
-            owner_user_id.as_str(),
-            session_id.as_str(),
-            payload.client_message_id.as_str(),
+            owner_user_id_for_stream.as_str(),
+            session_id_for_stream.as_str(),
+            client_message_id_for_stream.as_str(),
             "大鱼吃小鱼模板生成草稿",
-        ),
-        state.spacetime_client().clone(),
-    );
-    let mut streamed_reply_text = String::new();
-    let turn_result = run_big_fish_agent_turn(
-        BigFishAgentTurnRequest {
-            llm_client: state.llm_client(),
-            session: &submitted_session,
-            quick_fill_requested,
-        },
-        |text| {
-            draft_sink.persist_visible_text_async(text);
-            streamed_reply_text = text.to_string();
-        },
-    )
-    .await;
-    if !streamed_reply_text.is_empty() {
-        draft_writer
-            .persist_visible_text(state.spacetime_client(), streamed_reply_text.as_str())
-            .await;
-    }
-    let reply_text = match &turn_result {
-        Ok(result) => result.assistant_reply_text.clone(),
-        Err(error) => error.to_string(),
-    };
-    let finalize_input = match turn_result {
-        Ok(turn_result) => build_finalize_record_input(
-            session_id.clone(),
-            owner_user_id.clone(),
-            build_prefixed_uuid_id("big-fish-message-"),
-            turn_result,
-            current_utc_micros(),
-        ),
-        Err(error) => build_failed_finalize_record_input(
-            session_id.clone(),
-            owner_user_id.clone(),
-            &submitted_session,
-            error.to_string(),
-            current_utc_micros(),
-        ),
-    };
-    let session = state
-        .spacetime_client()
-        .finalize_big_fish_agent_message(finalize_input)
-        .await
-        .map_err(|error| {
-            big_fish_error_response(&request_context, map_big_fish_client_error(error))
-        })?;
+        ));
+        if let Err(error) = draft_writer.ensure_started(state.spacetime_client()).await {
+            tracing::warn!(error = %error, "大鱼吃小鱼模板生成草稿任务启动失败，主生成流程继续执行");
+        }
+        let (reply_tx, mut reply_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+        // 与 RPG/拼图 Agent 保持同一语义：回复先流式展示，session 真相仍等 finalize 后下发。
+        let turn_result = {
+            let run_turn = run_big_fish_agent_turn(
+                BigFishAgentTurnRequest {
+                    llm_client: state.llm_client(),
+                    session: &submitted_session,
+                    quick_fill_requested,
+                },
+                move |text| {
+                    let _ = reply_tx.send(text.to_string());
+                },
+            );
+            tokio::pin!(run_turn);
 
-    let session_response = map_big_fish_session_response(session);
-    let mut sse_body = String::new();
-    append_sse_event(
-        &request_context,
-        &mut sse_body,
-        "reply_delta",
-        &json!({ "text": if streamed_reply_text.is_empty() { reply_text } else { streamed_reply_text } }),
-    )?;
-    append_sse_event(
-        &request_context,
-        &mut sse_body,
-        "session",
-        &json!({ "session": session_response }),
-    )?;
-    append_sse_event(
-        &request_context,
-        &mut sse_body,
-        "done",
-        &json!({ "ok": true }),
-    )?;
-    Ok(build_event_stream_response(sse_body))
+            loop {
+                // 每个 replyText 增量同时写草稿表并推给 SSE，避免前端等待完整模型响应。
+                tokio::select! {
+                    result = &mut run_turn => break result,
+                    maybe_text = reply_rx.recv() => {
+                        if let Some(text) = maybe_text {
+                            draft_writer
+                                .persist_visible_text(state.spacetime_client(), text.as_str())
+                                .await;
+                            yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
+                                "reply_delta",
+                                json!({ "text": text }),
+                            ));
+                        }
+                    }
+                }
+            }
+        };
+
+        while let Some(text) = reply_rx.recv().await {
+            draft_writer
+                .persist_visible_text(state.spacetime_client(), text.as_str())
+                .await;
+            yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
+                "reply_delta",
+                json!({ "text": text }),
+            ));
+        }
+
+        let finalize_input = match turn_result {
+            Ok(turn_result) => build_finalize_record_input(
+                session_id_for_stream.clone(),
+                owner_user_id_for_stream.clone(),
+                build_prefixed_uuid_id("big-fish-message-"),
+                turn_result,
+                current_utc_micros(),
+            ),
+            Err(error) => build_failed_finalize_record_input(
+                session_id_for_stream.clone(),
+                owner_user_id_for_stream.clone(),
+                &submitted_session,
+                error.to_string(),
+                current_utc_micros(),
+            ),
+        };
+        let session = match state
+            .spacetime_client()
+            .finalize_big_fish_agent_message(finalize_input)
+            .await
+        {
+            Ok(session) => session,
+            Err(error) => {
+                yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
+                    "error",
+                    json!({ "message": error.to_string() }),
+                ));
+                return;
+            }
+        };
+
+        let session_response = map_big_fish_session_response(session);
+        yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
+            "session",
+            json!({ "session": session_response }),
+        ));
+        yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
+            "done",
+            json!({ "ok": true }),
+        ));
+    };
+    Ok(Sse::new(stream).into_response())
 }
 
 pub async fn execute_big_fish_action(
@@ -1706,40 +1724,20 @@ fn big_fish_bad_request(request_context: &RequestContext, message: &str) -> Resp
     )
 }
 
-fn append_sse_event(
-    request_context: &RequestContext,
-    body: &mut String,
-    event: &str,
-    payload: &Value,
-) -> Result<(), Response> {
-    let payload_text = serde_json::to_string(payload).map_err(|error| {
-        big_fish_error_response(
-            request_context,
-            AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_details(json!({
-                "provider": "big-fish",
-                "message": format!("SSE payload 序列化失败：{error}"),
-            })),
-        )
-    })?;
-    body.push_str("event: ");
-    body.push_str(event);
-    body.push('\n');
-    body.push_str("data: ");
-    body.push_str(&payload_text);
-    body.push_str("\n\n");
-    Ok(())
+fn big_fish_sse_json_event_or_error(event_name: &str, payload: Value) -> Event {
+    match serde_json::to_string(&payload) {
+        Ok(payload_text) => Event::default().event(event_name).data(payload_text),
+        Err(_) => big_fish_sse_error_event_message("SSE payload 序列化失败".to_string()),
+    }
 }
 
-fn build_event_stream_response(body: String) -> Response {
-    (
-        [
-            (header::CONTENT_TYPE, "text/event-stream; charset=utf-8"),
-            (header::CACHE_CONTROL, "no-cache"),
-            (HeaderName::from_static("x-accel-buffering"), "no"),
-        ],
-        body,
-    )
-        .into_response()
+fn big_fish_sse_error_event_message(message: String) -> Event {
+    let payload = format!(
+        "{{\"message\":{}}}",
+        serde_json::to_string(&message)
+            .unwrap_or_else(|_| "\"SSE 错误事件序列化失败\"".to_string())
+    );
+    Event::default().event("error").data(payload)
 }
 
 fn map_big_fish_client_error(error: SpacetimeClientError) -> AppError {

server-rs/crates/api-server/src/custom_world.rs

@@ -901,34 +901,49 @@ pub async fn stream_custom_world_agent_message(
         if let Err(error) = draft_writer.ensure_started(state.spacetime_client()).await {
             tracing::warn!(error = %error, "自定义世界模板生成草稿任务启动失败，主生成流程继续执行");
         }
-        let draft_sink = AiGenerationDraftSink::new(
-            AiGenerationDraftContext::new(
-                "custom_world",
-                owner_user_id_for_stream.as_str(),
-                session_id_for_stream.as_str(),
-                operation_id.as_str(),
-                "自定义世界模板生成草稿",
-            ),
-            state.spacetime_client().clone(),
-        );
-        // 聊天回复必须等本轮模型解析、进度与会话快照全部落库后，
-        // 再随最终 session 一次性返回，避免玩家先看到回复而进度仍停在旧状态。
-        let turn_result = run_custom_world_agent_turn(
-            CustomWorldAgentTurnRequest {
-                llm_client: state.llm_client(),
-                session: &session,
-                quick_fill_requested,
-                focus_card_id,
-            },
-            move |text| {
-                draft_sink.persist_visible_text_async(text);
-            },
-        )
-        .await;
-        if let Ok(result) = &turn_result {
+        let (reply_tx, mut reply_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+        // Agent turn 仍负责完整 JSON 解析和最终写回；这里把 replyText 增量桥接成前端可见的 SSE 分片。
+        let turn_result = {
+            let run_turn = run_custom_world_agent_turn(
+                CustomWorldAgentTurnRequest {
+                    llm_client: state.llm_client(),
+                    session: &session,
+                    quick_fill_requested,
+                    focus_card_id,
+                },
+                move |text| {
+                    let _ = reply_tx.send(text.to_string());
+                },
+            );
+            tokio::pin!(run_turn);
+
+            loop {
+                // 不等待最终 session 落库即可先推送回复进度；session/done 仍在 finalize 成功后发送。
+                tokio::select! {
+                    result = &mut run_turn => break result,
+                    maybe_text = reply_rx.recv() => {
+                        if let Some(text) = maybe_text {
+                            draft_writer
+                                .persist_visible_text(state.spacetime_client(), text.as_str())
+                                .await;
+                            yield Ok::<Event, Infallible>(custom_world_sse_json_event_or_error(
+                                "reply_delta",
+                                json!({ "text": text }),
+                            ));
+                        }
+                    }
+                }
+            }
+        };
+
+        while let Some(text) = reply_rx.recv().await {
             draft_writer
-                .persist_visible_text(state.spacetime_client(), result.assistant_reply_text.as_str())
+                .persist_visible_text(state.spacetime_client(), text.as_str())
                 .await;
+            yield Ok::<Event, Infallible>(custom_world_sse_json_event_or_error(
+                "reply_delta",
+                json!({ "text": text }),
+            ));
         }
 
         let finalize_input = match turn_result {

server-rs/crates/api-server/src/runtime_chat.rs

@@ -1,12 +1,16 @@
 use axum::{
     Json,
     extract::{Extension, State},
-    http::{StatusCode, header},
-    response::{IntoResponse, Response},
+    http::StatusCode,
+    response::{
+        IntoResponse, Response,
+        sse::{Event, Sse},
+    },
 };
 use platform_llm::{LlmMessage, LlmTextRequest};
 use serde::Deserialize;
 use serde_json::{Value, json};
+use std::convert::Infallible;
 
 use crate::{
     http_error::AppError,
@@ -58,7 +62,7 @@ pub async fn stream_runtime_npc_chat_turn(
     let npc_name = read_string_field(&payload.encounter, "npcName")
         .or_else(|| read_string_field(&payload.encounter, "name"))
         .unwrap_or_else(|| "对方".to_string());
-    let player_message = payload.player_message.trim();
+    let player_message = payload.player_message.trim().to_string();
     if player_message.is_empty() && !payload.npc_initiates_conversation {
         return Err(runtime_chat_error_response(
             &request_context,
@@ -69,75 +73,106 @@ pub async fn stream_runtime_npc_chat_turn(
         ));
     }
 
-    let llm_result =
-        generate_llm_npc_chat_turn(&state, &request_context, &payload, &npc_name).await;
-    let (mut body, npc_reply, suggestions, function_suggestions, force_exit) = match llm_result {
-        Some(result) => result,
-        None => {
-            let npc_reply = build_deterministic_npc_reply(
-                npc_name.as_str(),
-                player_message,
-                payload.npc_initiates_conversation,
-            );
-            let force_exit = should_force_chat_exit(payload.chat_directive.as_ref())
-                || should_hostile_chat_breakoff_deterministically(
-                    player_message,
-                    payload.chat_directive.as_ref(),
-                );
-            let suggestions = if force_exit {
-                Vec::new()
-            } else {
-                build_deterministic_chat_suggestions(npc_name.as_str(), player_message)
-            };
-            let function_suggestions = if force_exit {
-                Vec::new()
-            } else {
-                build_fallback_function_suggestions(payload.chat_directive.as_ref())
-            };
-            let mut body = String::new();
-            append_sse_event(
-                &request_context,
-                &mut body,
+    let stream = async_stream::stream! {
+        let (reply_tx, mut reply_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+        // `platform-llm` 在当前任务内持续回调增量文本；外层用 channel 把增量转成真正的 SSE 分片。
+        let llm_turn = generate_llm_npc_chat_turn(
+            &state,
+            &payload,
+            &npc_name,
+            move |text| {
+                let _ = reply_tx.send(text.to_string());
+            },
+        );
+        tokio::pin!(llm_turn);
+
+        let llm_result = loop {
+            // 模型尚未结束时优先把已收到的累计回复推出去，避免等完整建议生成后才一次性返回。
+            tokio::select! {
+                result = &mut llm_turn => break result,
+                maybe_text = reply_rx.recv() => {
+                    if let Some(text) = maybe_text {
+                        yield Ok::<Event, Infallible>(runtime_chat_sse_json_event_or_error(
+                            "reply_delta",
+                            json!({ "text": text }),
+                        ));
+                    }
+                }
+            }
+        };
+
+        while let Some(text) = reply_rx.recv().await {
+            yield Ok::<Event, Infallible>(runtime_chat_sse_json_event_or_error(
                 "reply_delta",
-                &json!({ "text": npc_reply }),
-            )?;
-            (
-                body,
-                npc_reply,
-                suggestions,
-                function_suggestions,
-                force_exit,
-            )
+                json!({ "text": text }),
+            ));
         }
-    };
 
-    let chatted_count = read_number_field(&payload.npc_state, "chattedCount").unwrap_or(0.0);
-    let affinity_delta = if payload.npc_initiates_conversation {
-        0
-    } else {
-        compute_npc_chat_affinity_delta(player_message, npc_reply.as_str(), chatted_count)
-    };
-    let complete_payload = json!({
-        "npcReply": npc_reply,
-        "affinityDelta": affinity_delta,
-        "affinityText": describe_affinity_shift(affinity_delta),
-        "suggestions": suggestions,
-        "functionSuggestions": function_suggestions,
-        "pendingQuestOffer": null,
-        "chatDirective": build_completion_directive(payload.chat_directive.as_ref(), force_exit),
-    });
+        let (npc_reply, suggestions, function_suggestions, force_exit) = match llm_result {
+            Some(result) => result,
+            None => {
+                let npc_reply = build_deterministic_npc_reply(
+                    npc_name.as_str(),
+                    player_message.as_str(),
+                    payload.npc_initiates_conversation,
+                );
+                let force_exit = should_force_chat_exit(payload.chat_directive.as_ref())
+                    || should_hostile_chat_breakoff_deterministically(
+                        player_message.as_str(),
+                        payload.chat_directive.as_ref(),
+                    );
+                let suggestions = if force_exit {
+                    Vec::new()
+                } else {
+                    build_deterministic_chat_suggestions(npc_name.as_str(), player_message.as_str())
+                };
+                let function_suggestions = if force_exit {
+                    Vec::new()
+                } else {
+                    build_fallback_function_suggestions(payload.chat_directive.as_ref())
+                };
+                yield Ok::<Event, Infallible>(runtime_chat_sse_json_event_or_error(
+                    "reply_delta",
+                    json!({ "text": npc_reply }),
+                ));
+                (npc_reply, suggestions, function_suggestions, force_exit)
+            }
+        };
 
-    append_sse_event(&request_context, &mut body, "complete", &complete_payload)?;
-    body.push_str("data: [DONE]\n\n");
-    Ok(build_event_stream_response(body))
+        let chatted_count = read_number_field(&payload.npc_state, "chattedCount").unwrap_or(0.0);
+        let affinity_delta = if payload.npc_initiates_conversation {
+            0
+        } else {
+            compute_npc_chat_affinity_delta(player_message.as_str(), npc_reply.as_str(), chatted_count)
+        };
+        let complete_payload = json!({
+            "npcReply": npc_reply,
+            "affinityDelta": affinity_delta,
+            "affinityText": describe_affinity_shift(affinity_delta),
+            "suggestions": suggestions,
+            "functionSuggestions": function_suggestions,
+            "pendingQuestOffer": null,
+            "chatDirective": build_completion_directive(payload.chat_directive.as_ref(), force_exit),
+        });
+
+        yield Ok::<Event, Infallible>(runtime_chat_sse_json_event_or_error(
+            "complete",
+            complete_payload,
+        ));
+        yield Ok::<Event, Infallible>(Event::default().data("[DONE]"));
+    };
+    Ok(Sse::new(stream).into_response())
 }
 
-async fn generate_llm_npc_chat_turn(
+async fn generate_llm_npc_chat_turn<F>(
     state: &AppState,
-    request_context: &RequestContext,
     payload: &NpcChatTurnRequest,
     npc_name: &str,
-) -> Option<(String, String, Vec<String>, Vec<Value>, bool)> {
+    mut on_reply_update: F,
+) -> Option<(String, Vec<String>, Vec<Value>, bool)>
+where
+    F: FnMut(&str),
+{
     let llm_client = state.llm_client()?;
     let character = payload
         .character
@@ -160,7 +195,6 @@ async fn generate_llm_npc_chat_turn(
         chat_directive: payload.chat_directive.as_ref(),
     };
 
-    let mut body = String::new();
     let reply_prompt = build_npc_chat_turn_reply_prompt(&prompt_input);
     let mut reply_request = LlmTextRequest::new(vec![
         LlmMessage::system(NPC_CHAT_TURN_REPLY_SYSTEM_PROMPT),
@@ -171,12 +205,7 @@ async fn generate_llm_npc_chat_turn(
 
     let reply_response = llm_client
         .stream_text(reply_request, |delta| {
-            let _ = append_sse_event(
-                request_context,
-                &mut body,
-                "reply_delta",
-                &json!({ "text": delta.accumulated_text }),
-            );
+            on_reply_update(delta.accumulated_text.as_str());
         })
         .await
         .ok()?;
@@ -189,7 +218,7 @@ async fn generate_llm_npc_chat_turn(
     });
 
     if should_force_chat_exit(payload.chat_directive.as_ref()) {
-        return Some((body, npc_reply, Vec::new(), Vec::new(), true));
+        return Some((npc_reply, Vec::new(), Vec::new(), true));
     }
 
     let suggestion_prompt =
@@ -224,13 +253,7 @@ async fn generate_llm_npc_chat_turn(
         suggestions = build_fallback_npc_chat_suggestions(payload.player_message.as_str());
     }
 
-    Some((
-        body,
-        npc_reply,
-        suggestions,
-        function_suggestions,
-        force_exit,
-    ))
+    Some((npc_reply, suggestions, function_suggestions, force_exit))
 }
 
 fn build_deterministic_npc_reply(
@@ -595,39 +618,20 @@ fn describe_affinity_shift(affinity_delta: i32) -> &'static str {
     "这轮对话暂时没有带来明显关系变化。"
 }
 
-fn append_sse_event(
-    request_context: &RequestContext,
-    body: &mut String,
-    event: &str,
-    payload: &Value,
-) -> Result<(), Response> {
-    let payload_text = serde_json::to_string(payload).map_err(|error| {
-        runtime_chat_error_response(
-            request_context,
-            AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_details(json!({
-                "provider": "runtime-chat",
-                "message": format!("SSE payload 序列化失败：{error}"),
-            })),
-        )
-    })?;
-    body.push_str("event: ");
-    body.push_str(event);
-    body.push('\n');
-    body.push_str("data: ");
-    body.push_str(&payload_text);
-    body.push_str("\n\n");
-    Ok(())
+fn runtime_chat_sse_json_event_or_error(event_name: &str, payload: Value) -> Event {
+    match serde_json::to_string(&payload) {
+        Ok(payload_text) => Event::default().event(event_name).data(payload_text),
+        Err(_) => runtime_chat_sse_error_event_message("SSE payload 序列化失败".to_string()),
+    }
 }
 
-fn build_event_stream_response(body: String) -> Response {
-    (
-        [
-            (header::CONTENT_TYPE, "text/event-stream; charset=utf-8"),
-            (header::CACHE_CONTROL, "no-cache"),
-        ],
-        body,
-    )
-        .into_response()
+fn runtime_chat_sse_error_event_message(message: String) -> Event {
+    let payload = format!(
+        "{{\"message\":{}}}",
+        serde_json::to_string(&message)
+            .unwrap_or_else(|_| "\"SSE 错误事件序列化失败\"".to_string())
+    );
+    Event::default().event("error").data(payload)
 }
 
 fn runtime_chat_error_response(request_context: &RequestContext, error: AppError) -> Response {