fix(canvas): 文本生成 - 修复分镜超时并贯通 Agent 与创作页流式返回

2026-08-27 22:39:30 +08:00
parent 482d8bdbc2
commit 7cb6aed27d
18 changed files with 944 additions and 44 deletions
--- a/backend/internal/handler/routes.go
+++ b/backend/internal/handler/routes.go
@@ -234,6 +234,10 @@ func streamTaskTextEvents(c *gin.Context, svc *service.Service, userID string, t
 		return
 	}
 	c.Writer.Flush()
+	lastStatus := replay.Status
+	lastStage := replay.Stage
+	lastProgress := replay.Progress
+	writeTaskTextSSE(c, "progress", 0, map[string]any{"status": replay.Status, "stage": replay.Stage, "progress": replay.Progress})

 	pollTicker := time.NewTicker(750 * time.Millisecond)
 	heartbeatTicker := time.NewTicker(15 * time.Second)
@@ -263,6 +267,10 @@ func streamTaskTextEvents(c *gin.Context, svc *service.Service, userID string, t
 			writeTaskTextSSE(c, "error", 0, map[string]string{"message": "任务文本流不可用"})
 			return
 		}
+		if next.Status != lastStatus || next.Stage != lastStage || next.Progress != lastProgress {
+			writeTaskTextSSE(c, "progress", 0, map[string]any{"status": next.Status, "stage": next.Stage, "progress": next.Progress})
+			lastStatus, lastStage, lastProgress = next.Status, next.Stage, next.Progress
+		}
 		replay = next
 	}
 }
--- a/backend/internal/handler/text_events_test.go
+++ b/backend/internal/handler/text_events_test.go
@@ -41,13 +41,16 @@ func TestStreamTaskTextEventsWritesSequenceAndTerminalEvent(t *testing.T) {
 		Deltas:    []model.TaskTextDelta{{Sequence: 8, Content: "增量"}},
 		Complete:  true,
 		FinalText: "增量",
+		Status:    model.TaskStatusSucceeded,
+		Stage:     "已完成",
+		Progress:  100,
 	}
 	streamTaskTextEvents(context, nil, "user-1", "task-1", 7, replay)
 	if response.Header().Get("Content-Type") != "text/event-stream; charset=utf-8" {
 		t.Fatalf("content type = %q", response.Header().Get("Content-Type"))
 	}
 	body := response.Body.String()
-	if !strings.Contains(body, "id: 8\nevent: delta") || !strings.Contains(body, "event: terminal") {
+	if !strings.Contains(body, "event: progress") || !strings.Contains(body, `"progress":100`) || !strings.Contains(body, "id: 8\nevent: delta") || !strings.Contains(body, "event: terminal") {
 		t.Fatalf("unexpected SSE body: %q", body)
 	}
 }
--- a/backend/internal/service/provider.go
+++ b/backend/internal/service/provider.go
@@ -17,6 +17,8 @@ import (
 	"net/http"
 	"net/textproto"
 	"net/url"
+	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -27,6 +29,8 @@ import (
 	"gorm.io/gorm"
 )

+var sseFrameBoundaryPattern = regexp.MustCompile(`\r?\n\r?\n`)
+
 type canvasGenerationInput struct {
 	Mode            string                 `json:"mode"`
 	Prompt          string                 `json:"prompt"`
@@ -40,6 +44,8 @@ type canvasGenerationInput struct {
 	AgentRequests   *agentToolRequests     `json:"agentRequests"`
 	ImageCapability *ImageCapabilityConfig `json:"-"`
 	StreamText      bool                   `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。
+	MaxOutputTokens int                    `json:"-"`
+	OnTextDelta     func(string)           `json:"-"`
 	VideoCapability *VideoCapabilityConfig `json:"-"`
 }

@@ -321,6 +327,13 @@ func (s *Service) processCanvasGenerationTask(ctx context.Context, userID string
 	}
 	input.Config = config
 	ctx = withProviderOutboundPolicy(ctx, input.Config)
+	var textPublisher *taskTextStreamPublisher
+	if input.Mode == "text" && strings.HasPrefix(taskType, "canvas_text") {
+		textPublisher = newTaskTextStreamPublisher(s, userID, taskExecutionID(ctx))
+		input.StreamText = true
+		input.OnTextDelta = textPublisher.Publish
+		defer textPublisher.Close()
+	}
 	if isWorkflowProviderInterface(input.Config.InterfaceType) {
 		if err := s.RequireWorkflowPluginForInterface(input.Config.InterfaceType); err != nil {
 			return nil, err
@@ -427,26 +440,23 @@ func runAgentToolTask(ctx context.Context, input canvasGenerationInput) (map[str
 		body = claudeAgentBody(body)
 	}
 	body["model"] = input.Config.Model
-	var payload map[string]interface{}
-	err := postJSON(ctx, input.Config, path, body, &payload)
+	result, err := postStreamingAgent(ctx, input.Config, path, body, protocol, input.OnTextDelta)
 	if protocol == "chat-completion" && isAgentToolChoiceCompatibilityError(err) {
 		if !isAutoAgentToolChoice(body["tool_choice"]) {
 			autoBody := cloneStringAnyMap(body)
 			autoBody["tool_choice"] = "auto"
-			payload = nil
-			err = postJSON(ctx, input.Config, path, autoBody, &payload)
+			result, err = postStreamingAgent(ctx, input.Config, path, autoBody, protocol, input.OnTextDelta)
 		}
 		if isAgentToolChoiceCompatibilityError(err) {
 			withoutToolChoice := cloneStringAnyMap(body)
 			delete(withoutToolChoice, "tool_choice")
-			payload = nil
-			err = postJSON(ctx, input.Config, path, withoutToolChoice, &payload)
+			result, err = postStreamingAgent(ctx, input.Config, path, withoutToolChoice, protocol, input.OnTextDelta)
 		}
 	}
 	if err != nil {
 		return nil, err
 	}
-	return parseAgentToolPayload(payload, protocol)
+	return result, nil
 }

 func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, adapter protocol.AgentAdapter) (map[string]interface{}, error) {
@@ -604,6 +614,10 @@ func isAgentToolChoiceCompatibilityError(err error) bool {
 	if errors.As(err, &payloadErr) {
 		message += " " + strings.ToLower(payloadErr.raw)
 	}
+	var httpErr providerHTTPError
+	if errors.As(err, &httpErr) {
+		message += " " + strings.ToLower(httpErr.Body)
+	}
 	return strings.Contains(message, "tool_choice") || strings.Contains(message, "tool choice") || strings.Contains(message, "tool-choice") || strings.Contains(message, "thinking mode")
 }

@@ -675,6 +689,293 @@ func parseAgentToolPayload(payload map[string]interface{}, protocol string) (map
 	return result, nil
 }

+func postStreamingAgent(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string)) (map[string]interface{}, error) {
+	body["stream"] = true
+	if protocol == "chat-completion" {
+		metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
+		if metadata.BillingMode == "token" {
+			if err := ensureChatCompletionStreamUsage(body); err != nil {
+				return nil, err
+			}
+		}
+	}
+	parser := newStreamingAgentParser(protocol, onDelta)
+	data, mimeType, err := postStreamingBinary(ctx, config, path, body, parser.consume)
+	if err != nil {
+		return nil, err
+	}
+	if !strings.Contains(strings.ToLower(mimeType), "event-stream") {
+		var payload map[string]interface{}
+		if err := json.Unmarshal(data, &payload); err != nil {
+			return nil, fmt.Errorf("Agent 接口返回格式无效：%w", err)
+		}
+		return parseAgentToolPayload(payload, protocol)
+	}
+	parser.flush()
+	return parser.result()
+}
+
+type streamingAgentToolCall struct {
+	id        string
+	name      string
+	arguments string
+}
+
+type streamingAgentParser struct {
+	protocol     string
+	buffer       string
+	text         strings.Builder
+	reasoning    strings.Builder
+	toolCalls    map[int]*streamingAgentToolCall
+	toolCallByID map[string]int
+	completed    map[string]interface{}
+	err          error
+	emit         func(string)
+}
+
+func newStreamingAgentParser(protocol string, emit func(string)) *streamingAgentParser {
+	return &streamingAgentParser{protocol: protocol, toolCalls: map[int]*streamingAgentToolCall{}, toolCallByID: map[string]int{}, emit: emit}
+}
+
+func (p *streamingAgentParser) consume(mimeType string, chunk []byte) {
+	if p == nil || p.err != nil || !strings.Contains(strings.ToLower(mimeType), "event-stream") || len(chunk) == 0 {
+		return
+	}
+	p.buffer += string(chunk)
+	p.consumeFrames(false)
+}
+
+func (p *streamingAgentParser) flush() {
+	if p == nil || p.err != nil {
+		return
+	}
+	p.consumeFrames(true)
+}
+
+func (p *streamingAgentParser) consumeFrames(flush bool) {
+	for p.err == nil {
+		match := sseFrameBoundaryPattern.FindStringIndex(p.buffer)
+		if match == nil {
+			break
+		}
+		p.consumeFrame(p.buffer[:match[0]])
+		p.buffer = p.buffer[match[1]:]
+	}
+	if flush && p.err == nil && strings.TrimSpace(p.buffer) != "" {
+		p.consumeFrame(p.buffer)
+		p.buffer = ""
+	}
+}
+
+func (p *streamingAgentParser) consumeFrame(frame string) {
+	var eventName string
+	var dataLines []string
+	for _, line := range strings.Split(strings.ReplaceAll(frame, "\r\n", "\n"), "\n") {
+		switch {
+		case strings.HasPrefix(line, "event:"):
+			eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
+		case strings.HasPrefix(line, "data:"):
+			dataLines = append(dataLines, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " "))
+		}
+	}
+	raw := strings.TrimSpace(strings.Join(dataLines, "\n"))
+	if raw == "" || raw == "[DONE]" {
+		return
+	}
+	var payload map[string]interface{}
+	if err := json.Unmarshal([]byte(raw), &payload); err != nil {
+		p.err = fmt.Errorf("Agent 流式事件解析失败：%w", err)
+		return
+	}
+	if err := validateTextPayload(payload); err != nil {
+		p.err = err
+		return
+	}
+	switch p.protocol {
+	case "responses":
+		p.consumeResponsesEvent(eventName, payload)
+	case "claude-api":
+		p.consumeClaudeEvent(payload)
+	default:
+		p.consumeChatCompletionEvent(payload)
+	}
+}
+
+func (p *streamingAgentParser) consumeResponsesEvent(eventName string, payload map[string]interface{}) {
+	eventType := firstNonEmptyString(strings.TrimSpace(eventName), stringField(payload, "type"))
+	switch eventType {
+	case "response.output_text.delta", "output_text.delta":
+		p.appendText(stringField(payload, "delta"))
+	case "response.reasoning.delta", "response.reasoning_text.delta", "response.reasoning_summary_text.delta":
+		p.reasoning.WriteString(stringField(payload, "delta"))
+	case "response.completed":
+		p.completed, _ = payload["response"].(map[string]interface{})
+	case "response.output_item.added":
+		item, _ := payload["item"].(map[string]interface{})
+		if stringField(item, "type") == "function_call" {
+			index := intField(payload, "output_index", len(p.toolCalls))
+			p.setToolCall(index, firstNonEmptyString(stringField(item, "call_id"), stringField(item, "id")), stringField(item, "name"), stringField(item, "arguments"))
+		}
+	case "response.function_call_arguments.delta":
+		index := p.responseToolCallIndex(payload)
+		p.toolCall(index).arguments += stringField(payload, "delta")
+	case "response.function_call_arguments.done":
+		index := p.responseToolCallIndex(payload)
+		if arguments := stringField(payload, "arguments"); arguments != "" {
+			p.toolCall(index).arguments = arguments
+		}
+	}
+}
+
+func (p *streamingAgentParser) responseToolCallIndex(payload map[string]interface{}) int {
+	itemID := firstNonEmptyString(stringField(payload, "item_id"), stringField(payload, "call_id"))
+	if index, ok := p.toolCallByID[itemID]; ok {
+		return index
+	}
+	return intField(payload, "output_index", len(p.toolCalls))
+}
+
+func (p *streamingAgentParser) consumeChatCompletionEvent(payload map[string]interface{}) {
+	choices, _ := payload["choices"].([]interface{})
+	for _, value := range choices {
+		choice, _ := value.(map[string]interface{})
+		delta, _ := choice["delta"].(map[string]interface{})
+		p.appendText(streamContentText(delta["content"]))
+		p.reasoning.WriteString(firstNonEmptyString(stringField(delta, "reasoning_content"), stringField(delta, "reasoning"), stringField(delta, "reasoning_text")))
+		for fallbackIndex, toolValue := range interfaceSlice(delta["tool_calls"]) {
+			tool, _ := toolValue.(map[string]interface{})
+			index := intField(tool, "index", fallbackIndex)
+			function, _ := tool["function"].(map[string]interface{})
+			current := p.toolCall(index)
+			if id := stringField(tool, "id"); id != "" {
+				current.id = id
+				p.toolCallByID[id] = index
+			}
+			if name := stringField(function, "name"); name != "" {
+				current.name += name
+			}
+			current.arguments += stringField(function, "arguments")
+		}
+	}
+}
+
+func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}) {
+	switch stringField(payload, "type") {
+	case "content_block_start":
+		block, _ := payload["content_block"].(map[string]interface{})
+		index := intField(payload, "index", len(p.toolCalls))
+		switch stringField(block, "type") {
+		case "text":
+			p.appendText(stringField(block, "text"))
+		case "tool_use":
+			arguments := ""
+			if input := block["input"]; input != nil {
+				if encoded, err := json.Marshal(input); err == nil && string(encoded) != "{}" {
+					arguments = string(encoded)
+				}
+			}
+			p.setToolCall(index, stringField(block, "id"), stringField(block, "name"), arguments)
+		}
+	case "content_block_delta":
+		delta, _ := payload["delta"].(map[string]interface{})
+		index := intField(payload, "index", len(p.toolCalls)-1)
+		if stringField(delta, "type") == "text_delta" {
+			p.appendText(stringField(delta, "text"))
+		}
+		if stringField(delta, "type") == "input_json_delta" {
+			p.toolCall(index).arguments += stringField(delta, "partial_json")
+		}
+	case "error":
+		errValue, _ := payload["error"].(map[string]interface{})
+		p.err = errors.New(defaultString(stringField(errValue, "message"), "Claude 上游返回失败"))
+	}
+}
+
+func (p *streamingAgentParser) appendText(delta string) {
+	if delta == "" {
+		return
+	}
+	p.text.WriteString(delta)
+	if p.emit != nil {
+		p.emit(delta)
+	}
+}
+
+func (p *streamingAgentParser) toolCall(index int) *streamingAgentToolCall {
+	if index < 0 {
+		index = 0
+	}
+	if p.toolCalls[index] == nil {
+		p.toolCalls[index] = &streamingAgentToolCall{}
+	}
+	return p.toolCalls[index]
+}
+
+func (p *streamingAgentParser) setToolCall(index int, id string, name string, arguments string) {
+	call := p.toolCall(index)
+	call.id, call.name, call.arguments = id, name, arguments
+	if id != "" {
+		p.toolCallByID[id] = index
+	}
+}
+
+func (p *streamingAgentParser) result() (map[string]interface{}, error) {
+	if p.err != nil {
+		return nil, p.err
+	}
+	if p.completed != nil {
+		result, err := parseAgentToolPayload(p.completed, p.protocol)
+		if err != nil {
+			return nil, err
+		}
+		if p.text.Len() > 0 {
+			result["text"] = p.text.String()
+		}
+		if p.reasoning.Len() > 0 {
+			result["reasoning"] = p.reasoning.String()
+		}
+		return result, nil
+	}
+	result := map[string]interface{}{"mode": "text", "text": p.text.String(), "toolCalls": []interface{}{}}
+	if p.reasoning.Len() > 0 {
+		result["reasoning"] = p.reasoning.String()
+	}
+	indices := make([]int, 0, len(p.toolCalls))
+	for index := range p.toolCalls {
+		indices = append(indices, index)
+	}
+	sort.Ints(indices)
+	calls := make([]interface{}, 0, len(indices))
+	for _, index := range indices {
+		call := p.toolCalls[index]
+		if call == nil || strings.TrimSpace(call.id) == "" || strings.TrimSpace(call.name) == "" {
+			continue
+		}
+		arguments := call.arguments
+		if strings.TrimSpace(arguments) == "" {
+			arguments = "{}"
+		}
+		var parsed interface{}
+		if err := json.Unmarshal([]byte(arguments), &parsed); err != nil {
+			return nil, fmt.Errorf("Agent 工具参数不是完整 JSON：%w", err)
+		}
+		calls = append(calls, map[string]interface{}{"id": call.id, "type": "function", "function": map[string]interface{}{"name": call.name, "arguments": arguments}})
+	}
+	result["toolCalls"] = calls
+	if p.text.Len() == 0 && len(calls) == 0 {
+		return nil, errors.New("画布 Agent 接口没有返回内容")
+	}
+	return result, nil
+}
+
+func intField(value map[string]interface{}, key string, fallback int) int {
+	number, ok := value[key].(float64)
+	if !ok || math.IsNaN(number) || math.IsInf(number, 0) {
+		return fallback
+	}
+	return int(number)
+}
+
 func extractResponseReasoning(payload map[string]interface{}) string {
 	var chunks []string
 	for _, value := range interfaceSlice(payload["output"]) {
@@ -1673,7 +1974,8 @@ func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[st
 		return nil, err
 	}
 	body := map[string]interface{}{"model": input.Config.Model, "input": responseInput}
-	text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText)
+	applyTextOutputLimit(body, input.MaxOutputTokens, "max_output_tokens")
+	text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
 	if err != nil {
 		if !shouldFallbackTextToChat(err) {
 			return nil, err
@@ -1693,7 +1995,8 @@ func runResponsesTextTask(ctx context.Context, input canvasGenerationInput) (map
 		return nil, err
 	}
 	body := map[string]interface{}{"model": input.Config.Model, "input": responseInput}
-	text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText)
+	applyTextOutputLimit(body, input.MaxOutputTokens, "max_output_tokens")
+	text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
 	if err != nil {
 		return nil, err
 	}
@@ -1712,7 +2015,8 @@ func runChatCompletionsTextTask(ctx context.Context, input canvasGenerationInput
 	}
 	messages = append(messages, map[string]interface{}{"role": "user", "content": userContent})
 	body := map[string]interface{}{"model": input.Config.Model, "messages": messages}
-	text, err := requestTextProvider(ctx, input.Config, "/chat/completions", body, "chat-completion", input.StreamText)
+	applyTextOutputLimit(body, input.MaxOutputTokens, "max_tokens")
+	text, err := requestTextProvider(ctx, input.Config, "/chat/completions", body, "chat-completion", input.StreamText, input.OnTextDelta)
 	if err != nil {
 		return nil, err
 	}
@@ -1732,17 +2036,27 @@ func runClaudeTextTask(ctx context.Context, input canvasGenerationInput) (map[st
 		return nil, err
 	}
 	messages = append(messages, map[string]interface{}{"role": "user", "content": content})
-	body := map[string]interface{}{"model": input.Config.Model, "max_tokens": 4096, "messages": messages}
+	maxTokens := 4096
+	if input.MaxOutputTokens > 0 {
+		maxTokens = input.MaxOutputTokens
+	}
+	body := map[string]interface{}{"model": input.Config.Model, "max_tokens": maxTokens, "messages": messages}
 	if systemPrompt := strings.TrimSpace(input.Config.SystemPrompt); systemPrompt != "" {
 		body["system"] = systemPrompt
 	}
-	text, err := requestTextProvider(ctx, input.Config, "/messages", body, "claude-api", input.StreamText)
+	text, err := requestTextProvider(ctx, input.Config, "/messages", body, "claude-api", input.StreamText, input.OnTextDelta)
 	if err != nil {
 		return nil, err
 	}
 	return map[string]interface{}{"mode": "text", "text": text}, nil
 }

+func applyTextOutputLimit(body map[string]interface{}, limit int, field string) {
+	if limit > 0 {
+		body[field] = limit
+	}
+}
+
 func claudeTextContent(input canvasGenerationInput) (interface{}, error) {
 	if len(input.ReferenceImages) == 0 {
 		return input.Prompt, nil
@@ -2025,6 +2339,10 @@ func protocolRequestFromInput(input canvasGenerationInput) protocol.GenerationRe
 			"count":        input.Config.Count,
 		},
 	}
+	if input.MaxOutputTokens > 0 {
+		request.Extra["max_output_tokens"] = input.MaxOutputTokens
+		request.Extra["max_tokens"] = input.MaxOutputTokens
+	}
 	if duration, err := strconv.Atoi(strings.TrimSpace(input.Config.VideoSeconds)); err == nil && duration > 0 {
 		request.Duration = duration
 	}
@@ -3613,7 +3931,7 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn
 	return nil, fmt.Errorf("%s视频生成超时", providerName)
 }

-func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool) (string, error) {
+func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool, onDelta func(string)) (string, error) {
 	if stream {
 		metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
 		if protocol == "chat-completion" && metadata.BillingMode == "token" {
@@ -3621,7 +3939,7 @@ func requestTextProvider(ctx context.Context, config providerConfig, path string
 				return "", err
 			}
 		}
-		return postStreamingText(ctx, config, path, body, protocol)
+		return postStreamingText(ctx, config, path, body, protocol, onDelta)
 	}
 	var payload map[string]interface{}
 	if err := postJSON(ctx, config, path, body, &payload); err != nil {
@@ -3634,13 +3952,15 @@ func requestTextProvider(ctx context.Context, config providerConfig, path string
 	return text, nil
 }

-func postStreamingText(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string) (string, error) {
+func postStreamingText(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string)) (string, error) {
 	// 只把分镜规划/修复切到上游 SSE，完整 JSON 仍在流结束后校验，避免半截结构污染画布。
 	body["stream"] = true
-	data, mimeType, err := postStreamingBinary(ctx, config, path, body)
+	parser := newStreamingTextDeltaParser(protocol, onDelta)
+	data, mimeType, err := postStreamingBinary(ctx, config, path, body, parser.consume)
 	if err != nil {
 		return "", err
 	}
+	parser.flush()
 	if !strings.Contains(strings.ToLower(mimeType), "event-stream") {
 		var payload map[string]interface{}
 		if err := json.Unmarshal(data, &payload); err != nil {
@@ -3682,11 +4002,12 @@ func extractTextPayload(payload map[string]interface{}, protocol string) string

 func validateTextPayload(payload map[string]interface{}) error {
 	if code, ok := payload["code"].(float64); ok && code != 0 {
-		return errors.New(defaultString(stringField(payload, "msg"), "请求失败"))
+		rawMessage := defaultString(stringField(payload, "msg"), "请求失败")
+		return providerPayloadError{raw: rawMessage, message: providerPayloadErrorMessage(rawMessage)}
 	}
 	if errValue, ok := payload["error"].(map[string]interface{}); ok {
 		if message := stringField(errValue, "message"); message != "" {
-			return errors.New(message)
+			return providerPayloadError{raw: message, message: providerPayloadErrorMessage(message)}
 		}
 	}
 	return nil
@@ -3786,7 +4107,7 @@ func streamContentText(value interface{}) string {
 	return result.String()
 }

-func postStreamingBinary(ctx context.Context, config providerConfig, path string, body interface{}) ([]byte, string, error) {
+func postStreamingBinary(ctx context.Context, config providerConfig, path string, body interface{}, onChunk func(string, []byte)) ([]byte, string, error) {
 	data, _ := json.Marshal(body)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL(config.BaseURL, path), bytes.NewReader(data))
 	if err != nil {
@@ -3796,7 +4117,96 @@ func postStreamingBinary(ctx context.Context, config providerConfig, path string
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "text/event-stream")
 	ApplyOutboundHeaders(req, config.Headers)
-	return doBinary(req)
+	return doBinaryWithConsumer(req, onChunk)
+}
+
+type streamingTextDeltaParser struct {
+	protocol string
+	buffer   string
+	emit     func(string)
+}
+
+func newStreamingTextDeltaParser(protocol string, emit func(string)) *streamingTextDeltaParser {
+	return &streamingTextDeltaParser{protocol: protocol, emit: emit}
+}
+
+func (p *streamingTextDeltaParser) consume(mimeType string, chunk []byte) {
+	if p == nil || p.emit == nil || !strings.Contains(strings.ToLower(mimeType), "event-stream") || len(chunk) == 0 {
+		return
+	}
+	p.buffer += string(chunk)
+	p.consumeFrames(false)
+}
+
+func (p *streamingTextDeltaParser) flush() {
+	if p == nil || p.emit == nil {
+		return
+	}
+	p.consumeFrames(true)
+}
+
+func (p *streamingTextDeltaParser) consumeFrames(flush bool) {
+	for {
+		match := sseFrameBoundaryPattern.FindStringIndex(p.buffer)
+		if match == nil {
+			break
+		}
+		p.consumeFrame(p.buffer[:match[0]])
+		p.buffer = p.buffer[match[1]:]
+	}
+	if flush && strings.TrimSpace(p.buffer) != "" {
+		p.consumeFrame(p.buffer)
+		p.buffer = ""
+	}
+}
+
+func (p *streamingTextDeltaParser) consumeFrame(frame string) {
+	var eventName string
+	var dataLines []string
+	for _, line := range strings.Split(strings.ReplaceAll(frame, "\r\n", "\n"), "\n") {
+		switch {
+		case strings.HasPrefix(line, "event:"):
+			eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
+		case strings.HasPrefix(line, "data:"):
+			dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
+		}
+	}
+	raw := strings.TrimSpace(strings.Join(dataLines, "\n"))
+	if raw == "" || raw == "[DONE]" {
+		return
+	}
+	var payload map[string]interface{}
+	if json.Unmarshal([]byte(raw), &payload) != nil {
+		return
+	}
+	if delta := streamingTextDelta(p.protocol, eventName, payload); delta != "" {
+		p.emit(delta)
+	}
+}
+
+func streamingTextDelta(protocol string, eventName string, payload map[string]interface{}) string {
+	if protocol == "responses" {
+		eventType := firstNonEmptyString(strings.TrimSpace(eventName), stringField(payload, "type"))
+		if eventType == "response.output_text.delta" || eventType == "output_text.delta" || eventType == "" || eventType == "message" {
+			return stringField(payload, "delta")
+		}
+		return ""
+	}
+	if protocol == "claude-api" {
+		delta, _ := payload["delta"].(map[string]interface{})
+		if stringField(delta, "type") == "text_delta" {
+			return stringField(delta, "text")
+		}
+		return ""
+	}
+	choices, _ := payload["choices"].([]interface{})
+	var text strings.Builder
+	for _, choice := range choices {
+		record, _ := choice.(map[string]interface{})
+		delta, _ := record["delta"].(map[string]interface{})
+		text.WriteString(streamContentText(delta["content"]))
+	}
+	return text.String()
 }

 func postJSON(ctx context.Context, config providerConfig, path string, body interface{}, target interface{}) error {
@@ -3951,6 +4361,10 @@ func providerLoopbackPolicyForRequest(req *http.Request) (OutboundPolicy, bool)
 }

 func doBinary(req *http.Request) ([]byte, string, error) {
+	return doBinaryWithConsumer(req, nil)
+}
+
+func doBinaryWithConsumer(req *http.Request, onChunk func(string, []byte)) ([]byte, string, error) {
 	startedAt := time.Now()
 	requestTimeout := providerHTTPTimeout
 	if deadline, ok := req.Context().Deadline(); ok {
@@ -4022,17 +4436,37 @@ func doBinary(req *http.Request) ([]byte, string, error) {
 		recordProviderRequest(req, startedAt, resp.StatusCode, nil, err)
 		return nil, "", err
 	}
-	data, err := io.ReadAll(io.LimitReader(resp.Body, responseLimit+1))
-	if err != nil {
-		recordProviderRequest(req, startedAt, resp.StatusCode, nil, err)
-		return nil, "", err
+	mimeType := resp.Header.Get("Content-Type")
+	var buffered bytes.Buffer
+	reader := io.LimitReader(resp.Body, responseLimit+1)
+	chunk := make([]byte, 32<<10)
+	for {
+		readCount, readErr := reader.Read(chunk)
+		if readCount > 0 {
+			if int64(buffered.Len()+readCount) > responseLimit {
+				err = fmt.Errorf("上游响应超过 %s 限制", formatStorageLimit(responseLimit))
+				recordProviderRequest(req, startedAt, resp.StatusCode, buffered.Bytes(), err)
+				return nil, "", err
+			}
+			_, _ = buffered.Write(chunk[:readCount])
+			if onChunk != nil {
+				onChunk(mimeType, chunk[:readCount])
+			}
+		}
+		if errors.Is(readErr, io.EOF) {
+			break
+		}
+		if readErr != nil {
+			recordProviderRequest(req, startedAt, resp.StatusCode, buffered.Bytes(), readErr)
+			return nil, "", readErr
+		}
 	}
+	data := buffered.Bytes()
 	if int64(len(data)) > responseLimit {
 		err = fmt.Errorf("上游响应超过 %s 限制", formatStorageLimit(responseLimit))
 		recordProviderRequest(req, startedAt, resp.StatusCode, nil, err)
 		return nil, "", err
 	}
-	mimeType := resp.Header.Get("Content-Type")
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		if runtimeService != nil {
 			_ = runtimeService.RecordChannelResult(req.Context(), channelID, resp.StatusCode >= 500)
--- a/backend/internal/service/provider_test.go
+++ b/backend/internal/service/provider_test.go
@@ -346,10 +346,86 @@ data: [DONE]
 	}))
 	defer server.Close()

-	got, err := postStreamingText(context.Background(), providerConfig{BaseURL: server.URL, APIKey: "test-key"}, "/chat/completions", map[string]interface{}{"model": "test-model"}, "chat-completion")
+	var deltas strings.Builder
+	got, err := postStreamingText(context.Background(), providerConfig{BaseURL: server.URL, APIKey: "test-key"}, "/chat/completions", map[string]interface{}{"model": "test-model"}, "chat-completion", func(delta string) {
+		deltas.WriteString(delta)
+	})
 	if err != nil || got != "流式分镜" {
 		t.Fatalf("postStreamingText() = %q, err = %v", got, err)
 	}
+	if deltas.String() != "流式分镜" {
+		t.Fatalf("stream deltas = %q", deltas.String())
+	}
+}
+
+func TestStreamingAgentParserReassemblesChatToolCallsAcrossChunks(t *testing.T) {
+	var deltas strings.Builder
+	parser := newStreamingAgentParser("chat-completion", func(delta string) {
+		deltas.WriteString(delta)
+	})
+	stream := `data: {"choices":[{"delta":{"content":"准备","tool_calls":[{"index":0,"id":"call-1","function":{"name":"canvas_apply_ops","arguments":"{\"ops\":"}}]}}]}
+
+data: {"choices":[{"delta":{"content":"执行","tool_calls":[{"index":0,"function":{"arguments":"[]}"}}]}}]}
+
+data: [DONE]
+
+`
+	parser.consume("text/event-stream", []byte(stream[:47]))
+	parser.consume("text/event-stream", []byte(stream[47:]))
+	parser.flush()
+	result, err := parser.result()
+	if err != nil {
+		t.Fatalf("streamingAgentParser.result() error = %v", err)
+	}
+	if result["text"] != "准备执行" || deltas.String() != "准备执行" {
+		t.Fatalf("text = %v, deltas = %q", result["text"], deltas.String())
+	}
+	calls, _ := result["toolCalls"].([]interface{})
+	call, _ := calls[0].(map[string]interface{})
+	function, _ := call["function"].(map[string]interface{})
+	if call["id"] != "call-1" || function["name"] != "canvas_apply_ops" || function["arguments"] != `{"ops":[]}` {
+		t.Fatalf("tool call = %#v", call)
+	}
+}
+
+func TestStreamingAgentParserSeparatesResponsesReasoningFromVisibleText(t *testing.T) {
+	var deltas strings.Builder
+	parser := newStreamingAgentParser("responses", func(delta string) {
+		deltas.WriteString(delta)
+	})
+	parser.consume("text/event-stream", []byte(`event: response.reasoning_summary_text.delta
+data: {"type":"response.reasoning_summary_text.delta","delta":"内部分析"}
+
+event: response.output_text.delta
+data: {"type":"response.output_text.delta","delta":"可见回答"}
+
+event: response.completed
+data: {"type":"response.completed","response":{"output":[{"type":"message","content":[{"type":"output_text","text":"可见回答"}]}]}}
+
+`))
+	parser.flush()
+	result, err := parser.result()
+	if err != nil {
+		t.Fatalf("streamingAgentParser.result() error = %v", err)
+	}
+	if result["text"] != "可见回答" || result["reasoning"] != "内部分析" || deltas.String() != "可见回答" {
+		t.Fatalf("result = %#v, deltas = %q", result, deltas.String())
+	}
+}
+
+func TestStreamingAgentParserWaitsForCompleteClaudeToolJSON(t *testing.T) {
+	parser := newStreamingAgentParser("claude-api", nil)
+	parser.consume("text/event-stream", []byte(`event: content_block_start
+data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call-2","name":"canvas_get_state","input":{}}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"include\":"}}
+
+`))
+	parser.flush()
+	if _, err := parser.result(); err == nil || !strings.Contains(err.Error(), "完整 JSON") {
+		t.Fatalf("incomplete tool arguments error = %v", err)
+	}
 }

 func TestProviderHTTPErrorWarnsAboutUncertain524Billing(t *testing.T) {
--- a/backend/internal/service/storyboard.go
+++ b/backend/internal/service/storyboard.go
@@ -9,6 +9,8 @@ import (
 	"unicode/utf8"
 )

+const maxStoryboardShots = 12
+
 type agentStoryboardInput struct {
 	References     []string                  `json:"references"`
 	CanvasSnapshot map[string]any            `json:"canvasSnapshot"`
@@ -105,8 +107,8 @@ func parseAgentStoryboardPlan(raw string) (agentStoryboardPlan, error) {
 	if len(plan.Shots) == 0 {
 		return agentStoryboardPlan{}, errors.New("分镜模型没有返回 shots")
 	}
-	if len(plan.Shots) > 12 {
-		return agentStoryboardPlan{}, fmt.Errorf("分镜数量最多 12 个，实际返回 %d 个", len(plan.Shots))
+	if len(plan.Shots) > maxStoryboardShots {
+		return agentStoryboardPlan{}, fmt.Errorf("分镜数量最多 %d 个，实际返回 %d 个", maxStoryboardShots, len(plan.Shots))
 	}
 	for i := range plan.Shots {
 		if strings.TrimSpace(plan.Shots[i].Title) == "" {
--- a/backend/internal/service/storyboard_prompt.go
+++ b/backend/internal/service/storyboard_prompt.go
@@ -37,7 +37,7 @@ func storyboardPromptValues(brief string, requirements string, assets []storyboa
 	if shotDuration == 5 || shotDuration == 10 || shotDuration == 15 || shotDuration == 30 {
 		durationRule = fmt.Sprintf("本次生成单个镜头时长必须严格等于 %d 秒。", shotDuration)
 	}
-	countRule := "镜头数量由模型按剧情节奏自动决定。"
+	countRule := fmt.Sprintf("镜头数量由模型按剧情节奏自动决定，但 shots 数组必须为 1 到 %d 个镜头，并优先使用完整表达剧情所需的最少镜头数。", maxStoryboardShots)
 	if shotCount >= 1 && shotCount <= 10 {
 		countRule = fmt.Sprintf("shots 数组必须严格输出 %d 个镜头。", shotCount)
 	}
@@ -48,6 +48,13 @@ func storyboardPromptValues(brief string, requirements string, assets []storyboa
 	}
 }

+func storyboardOutputTokenLimit(shotCount int) int {
+	if shotCount >= 1 && shotCount <= 10 {
+		return min(12_000, 2_000+shotCount*800)
+	}
+	return 12_000
+}
+
 // 兼容历史模板中强制真人媒介的冲突规则；项目画风才是视觉媒介的唯一来源。
 func removeFixedMediaRestrictions(template string) string {
 	lines := strings.Split(template, "\n")
--- a/backend/internal/service/storyboard_prompt_test.go
+++ b/backend/internal/service/storyboard_prompt_test.go
@@ -1,8 +1,10 @@
 package service

 import (
+	"context"
 	"strings"
 	"testing"
+	"time"
 	"unicode/utf8"
 )

@@ -23,6 +25,29 @@ func TestStoryboardCinematicQualityContractIncludesCameraLanguageGuide(t *testin
 			t.Fatalf("camera language guide is missing %q: %s", term, contract)
 		}
 	}
+	if !strings.Contains(contract, "1 到 12 个镜头") || !strings.Contains(contract, "所需的最少镜头数") {
+		t.Fatalf("automatic shot count contract is not bounded: %s", contract)
+	}
+}
+
+func TestStoryboardOutputTokenLimitTracksRequestedShotCount(t *testing.T) {
+	if got := storyboardOutputTokenLimit(1); got != 2800 {
+		t.Fatalf("one-shot token limit = %d, want 2800", got)
+	}
+	if got := storyboardOutputTokenLimit(10); got != 10000 {
+		t.Fatalf("ten-shot token limit = %d, want 10000", got)
+	}
+	if got := storyboardOutputTokenLimit(0); got != 12000 {
+		t.Fatalf("automatic token limit = %d, want 12000", got)
+	}
+}
+
+func TestStoryboardRepairContextRejectsInsufficientBudget(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), storyboardFinalizeReserve+storyboardMinimumRepairTime-time.Second)
+	defer cancel()
+	if _, _, err := storyboardRepairContext(ctx); err == nil {
+		t.Fatal("expected insufficient repair budget to fail")
+	}
 }

 func TestStoryboardPromptsLeaveAspectRatioToVideoNode(t *testing.T) {
--- a/backend/internal/service/storyboard_workflow.go
+++ b/backend/internal/service/storyboard_workflow.go
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"

 	"infinite-canvas/backend/internal/model"
 )
@@ -59,12 +60,18 @@ func (s *Service) processStoryboardRowsTask(ctx context.Context, task model.Task
 	return map[string]interface{}{"title": plan.Title, "rows": rows}, nil, nil
 }

-const maxStoryboardRepairAttempts = 2
+const (
+	maxStoryboardRepairAttempts = 1
+	storyboardRepairMaxDuration = 4 * time.Minute
+	storyboardFinalizeReserve   = 30 * time.Second
+	storyboardMinimumRepairTime = 90 * time.Second
+)

 func (s *Service) repairStoryboardPlan(ctx context.Context, task model.Task, input agentStoryboardInput, config providerConfig, originalText string, validationErr error, shotDuration int, shotCount int) (agentStoryboardPlan, error) {
 	currentText := originalText
 	currentErr := validationErr
 	for attempt := 1; attempt <= maxStoryboardRepairAttempts; attempt++ {
+		_ = s.log(task.UserID, task.ID, "warn", "分镜结构校验失败", fmt.Sprintf("第 %d 次修复前：%s", attempt, currentErr.Error()))
 		if err := s.repo.UpdateTaskProgress(task.ID, "修复分镜结构", 55+attempt*10); err != nil {
 			return agentStoryboardPlan{}, fmt.Errorf("更新分镜修复进度失败，上游修复请求未发出：%w", err)
 		}
@@ -72,7 +79,12 @@ func (s *Service) repairStoryboardPlan(ctx context.Context, task model.Task, inp
 		if promptErr != nil {
 			return agentStoryboardPlan{}, promptErr
 		}
-		repaired, repairErr := runTextTask(withProviderRequestKind(ctx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config, StreamText: true})
+		repairCtx, cancel, budgetErr := storyboardRepairContext(ctx)
+		if budgetErr != nil {
+			return agentStoryboardPlan{}, budgetErr
+		}
+		repaired, repairErr := runTextTask(withProviderRequestKind(repairCtx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config, StreamText: true, MaxOutputTokens: storyboardOutputTokenLimit(shotCount)})
+		cancel()
 		if repairErr != nil {
 			return agentStoryboardPlan{}, fmt.Errorf("分镜结构修复失败：%w", repairErr)
 		}
@@ -128,7 +140,7 @@ func (s *Service) generateStoryboardPlan(ctx context.Context, task model.Task, i
 	if err != nil {
 		return agentStoryboardPlan{}, nil, err
 	}
-	result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config, StreamText: true})
+	result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config, StreamText: true, MaxOutputTokens: storyboardOutputTokenLimit(shotCount)})
 	if err != nil {
 		return agentStoryboardPlan{}, nil, err
 	}
@@ -150,6 +162,21 @@ func (s *Service) generateStoryboardPlan(ctx context.Context, task model.Task, i
 	return plan, assets, nil
 }

+func storyboardRepairContext(ctx context.Context) (context.Context, context.CancelFunc, error) {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		repairCtx, cancel := context.WithTimeout(ctx, storyboardRepairMaxDuration)
+		return repairCtx, cancel, nil
+	}
+	remaining := time.Until(deadline)
+	if remaining <= storyboardFinalizeReserve+storyboardMinimumRepairTime {
+		return nil, nil, fmt.Errorf("分镜结构校验失败，但任务剩余时间不足以安全修复：剩余 %s", remaining.Round(time.Second))
+	}
+	repairDuration := min(storyboardRepairMaxDuration, remaining-storyboardFinalizeReserve)
+	repairCtx, cancel := context.WithTimeout(ctx, repairDuration)
+	return repairCtx, cancel, nil
+}
+
 func (s *Service) buildAgentStoryboardResult(task model.Task, plan agentStoryboardPlan, assets []storyboardAsset, projectStyle storyboardProjectStyle) (map[string]interface{}, []map[string]interface{}, error) {
 	prefix := "agent-" + task.ID
 	scriptID := prefix + "-script"
--- a/backend/internal/service/task_text_stream.go
+++ b/backend/internal/service/task_text_stream.go
@@ -0,0 +1,84 @@
+package service
+
+import (
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	taskTextStreamFlushBytes    = 2 << 10
+	taskTextStreamFlushInterval = 500 * time.Millisecond
+)
+
+// taskTextStreamPublisher batches model deltas before persisting them. Streaming
+// is an observability and recovery enhancement: persistence failure must not turn
+// an otherwise successful model response into a failed generation task.
+type taskTextStreamPublisher struct {
+	service  *Service
+	userID   string
+	taskID   string
+	mu       sync.Mutex
+	buffer   strings.Builder
+	timer    *time.Timer
+	disabled bool
+	closed   bool
+}
+
+func newTaskTextStreamPublisher(service *Service, userID string, taskID string) *taskTextStreamPublisher {
+	return &taskTextStreamPublisher{service: service, userID: userID, taskID: taskID}
+}
+
+func (p *taskTextStreamPublisher) Publish(delta string) {
+	if p == nil || delta == "" {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.disabled || p.closed {
+		return
+	}
+	p.buffer.WriteString(delta)
+	if p.buffer.Len() >= taskTextStreamFlushBytes {
+		p.flushLocked()
+		return
+	}
+	if p.timer == nil {
+		p.timer = time.AfterFunc(taskTextStreamFlushInterval, p.flush)
+	}
+}
+
+func (p *taskTextStreamPublisher) Close() {
+	if p == nil {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.closed = true
+	p.flushLocked()
+}
+
+func (p *taskTextStreamPublisher) flush() {
+	if p == nil {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.flushLocked()
+}
+
+func (p *taskTextStreamPublisher) flushLocked() {
+	if p.timer != nil {
+		p.timer.Stop()
+		p.timer = nil
+	}
+	if p.disabled || p.buffer.Len() == 0 {
+		return
+	}
+	content := p.buffer.String()
+	p.buffer.Reset()
+	if _, err := p.service.AppendTaskTextDelta(p.userID, p.taskID, content); err != nil {
+		p.disabled = true
+		_ = p.service.log(p.userID, p.taskID, "warn", "文本流持久化已降级", err.Error())
+	}
+}
--- a/backend/internal/service/text_replay.go
+++ b/backend/internal/service/text_replay.go
@@ -27,6 +27,10 @@ type TextReplayResult struct {
 	TextDraft string                `json:"textDraft,omitempty"`
 	FinalText string                `json:"finalText,omitempty"`
 	Complete  bool                  `json:"complete"`
+	Status    model.TaskStatus      `json:"status"`
+	Stage     string                `json:"stage,omitempty"`
+	Progress  int                   `json:"progress"`
+	Error     string                `json:"error,omitempty"`
 }

 // isTextReplayTaskRequest 识别前端自管的文本持久化请求（input 带 replay=true）。
@@ -107,7 +111,11 @@ func (s *Service) TaskTextReplay(userID string, taskID string, after int64) (*Te
 	if err != nil {
 		return nil, err
 	}
-	result := &TextReplayResult{Deltas: deltas, TextDraft: task.TextDraft, Complete: task.Status == model.TaskStatusSucceeded || task.Status == model.TaskStatusFailed || task.Status == model.TaskStatusCancelled}
+	result := &TextReplayResult{
+		Deltas: deltas, TextDraft: task.TextDraft,
+		Complete: task.Status == model.TaskStatusSucceeded || task.Status == model.TaskStatusFailed || task.Status == model.TaskStatusCancelled,
+		Status:   task.Status, Stage: task.Stage, Progress: task.Progress, Error: task.Error,
+	}
 	if task.Status == model.TaskStatusSucceeded {
 		result.FinalText = taskResultText(task.ResultJSON)
 	}
--- a/web/src/components/canvas/canvas-assistant-panel.tsx
+++ b/web/src/components/canvas/canvas-assistant-panel.tsx
@@ -1665,8 +1665,7 @@ function toolCallToResponseInput(call: ResponseToolCall): ResponseInputMessage {

 async function requestOnlineAgentModel(config: AiConfig, messages: ResponseInputMessage[], toolChoice: "auto" | "required", prompt: string, onDelta: (text: string) => void, promptCacheKey?: string) {
    if (backendModelRuntimeRequired(config)) {
-        const result = await runBackendToolGenerationTask({ prompt, config, messages, tools: ONLINE_AGENT_TOOLS, toolChoice });
-        if (result.content.trim()) onDelta(result.content);
+        const result = await runBackendToolGenerationTask({ prompt, config, messages, tools: ONLINE_AGENT_TOOLS, toolChoice, onDelta });
        return result;
    }
    return requestToolResponse(config, messages, ONLINE_AGENT_TOOLS, toolChoice, onDelta, { promptCacheKey });
--- a/web/src/lib/canvas/canvas-agent-session.ts
+++ b/web/src/lib/canvas/canvas-agent-session.ts
@@ -7,7 +7,6 @@ import {
 } from "@/services/api/task-center";

 const CINEMATIC_SESSION_POLL_INTERVAL_MS = 2000;
-const CINEMATIC_SESSION_MAX_POLLS = 120;

 type CinematicSessionWaitOptions = {
    signal?: AbortSignal;
@@ -42,14 +41,15 @@ export function isAgentSessionPollingAbort(error: unknown) {

 async function waitForCinematicAgentSession(initialDetail: AgentSessionDetail, options: CinematicSessionWaitOptions) {
    let detail = initialDetail;
-    for (let attempt = 0; attempt < CINEMATIC_SESSION_MAX_POLLS; attempt += 1) {
+    // 后端任务策略负责生成超时和终态收敛；前端只在调用方 Abort 时停止监听，
+    // 避免用独立的固定轮询次数把仍在服务端执行的分镜误判为失败。
+    for (;;) {
        throwIfAborted(options.signal);
        if (detail.session.status === "completed") return detail;
        if (detail.session.status === "failed") throw new Error(agentSessionFailureMessage(detail));
        await abortableDelay(CINEMATIC_SESSION_POLL_INTERVAL_MS, options.signal);
        detail = await queryAgentSession(initialDetail.session.id);
    }
-    throw new Error("后端影视 Agent 会话超时");
 }

 function abortableDelay(ms: number, signal?: AbortSignal) {
--- a/web/src/pages/canvas/use-canvas-storyboard.ts
+++ b/web/src/pages/canvas/use-canvas-storyboard.ts
@@ -158,6 +158,7 @@ export function useCanvasStoryboard({
            setNodes((current) => current.map((node) => node.id === nodeId ? { ...node, metadata: { ...node.metadata, ...generationTaskMetadata(task), status: NODE_STATUS_LOADING } } : node));
            const completed = await waitForGenerationTask(task.id, {
                initialTask: task,
+                useTextEvents: true,
                onTaskUpdate: (next) => setNodes((current) => current.map((node) => node.id === nodeId ? { ...node, metadata: { ...node.metadata, ...generationTaskMetadata(next), status: NODE_STATUS_LOADING } } : node)),
            });
            const result = storyboardRowsFromTask(completed);
--- a/web/src/pages/create/index.tsx
+++ b/web/src/pages/create/index.tsx
@@ -622,6 +622,7 @@ export default function CreatePage() {
 						signal: requestLifecycle.signal,
 						metadata: { source: "create-page", conversationId: activeConversation.id, messageId: assistantMessage.id, ...referenceMetadata },
 						onTaskUpdate: bindTask,
+						onTextDelta: (text) => updateOriginAssistant((item) => ({ ...item, content: text })),
 						...retryContext,
 					}));
 					if (!result.text?.trim()) throw new Error("后端任务没有返回文本");
--- a/web/src/services/api/generation-task.ts
+++ b/web/src/services/api/generation-task.ts
@@ -52,6 +52,7 @@ type BackendGenerationTaskOptions = {
    signal?: AbortSignal;
    metadata?: Record<string, unknown>;
    onTaskUpdate?: (task: GenerationTask) => void;
+    onTextDelta?: (text: string) => void;
    localIdempotencyKey?: string;
    localResumeOnly?: boolean;
    clientOperationId?: string;
@@ -130,6 +131,7 @@ export async function runBackendToolGenerationTask(options: {
    tools: ResponseFunctionTool[];
    toolChoice: ToolChoice;
    signal?: AbortSignal;
+    onDelta?: (text: string) => void;
 }): Promise<ToolResponseResult> {
    throwIfAborted(options.signal);
    const logicalModelId = logicalModelIDForConfig(options.config);
@@ -149,7 +151,7 @@ export async function runBackendToolGenerationTask(options: {
            metadata: { source: "canvas-online-agent" },
        },
    });
-    const completed = await waitForGenerationTask(task.id, { signal: options.signal, initialTask: task });
+    const completed = await waitForGenerationTask(task.id, { signal: options.signal, initialTask: task, onTextDelta: options.onDelta });
    const result = parseBackendGenerationResult(completed);
    return {
        content: result.text || "",
@@ -399,7 +401,7 @@ async function prepareGenerationReferences({
 }

 async function createAndWaitGenerationTask(options: BackendGenerationTaskOptions, prepared: PreparedGenerationReferences, dependencies: GenerationTaskDependencies) {
-    const { projectId, mode, prompt, config, signal, metadata, onTaskUpdate } = options;
+    const { projectId, mode, prompt, config, signal, metadata, onTaskUpdate, onTextDelta } = options;
    const videoOperation = generationOperation(options);
    const workflow = resolveGenerationWorkflowExecution(config, mode);
    const logicalModelId = workflow ? "" : logicalModelIDForConfig(config);
@@ -426,7 +428,7 @@ async function createAndWaitGenerationTask(options: BackendGenerationTaskOptions
        },
    });
    onTaskUpdate?.(task);
-    const completed = await dependencies.waitTask(task.id, { signal, initialTask: task, onTaskUpdate });
+    const completed = await dependencies.waitTask(task.id, { signal, initialTask: task, onTaskUpdate, onTextDelta });
    return parseBackendGenerationResult(completed);
 }

--- a/web/src/services/api/task-center.ts
+++ b/web/src/services/api/task-center.ts
@@ -1,5 +1,6 @@
 import { DREAMINA_SUBMIT_ERROR_MESSAGES, generationErrorMessage } from "@/lib/generation-error";
-import { apiClient, request, type BackendEnvelope } from "@/services/api/request";
+import { apiBaseURL, apiClient, request, type BackendEnvelope } from "@/services/api/request";
+import { consumeTaskTextStream, createTaskTextStreamParser, type TaskTextStreamEvent } from "@/services/api/task-text-stream";
 import { recordDiagnosticEvent } from "@/services/diagnostics/client-diagnostics";
 import {
    deleteLocalDreaminaGenerationTask,
@@ -100,6 +101,10 @@ export type TaskTextReplay = {
    textDraft?: string;
    finalText?: string;
    complete: boolean;
+    status: TaskStatus;
+    stage?: string;
+    progress: number;
+    error?: string;
 };

 export type AgentSession = {
@@ -491,7 +496,17 @@ function safeTaskLogErrorCode(value: unknown) {
    return undefined;
 }

-export async function waitForGenerationTask(id: string, options?: { signal?: AbortSignal; intervalMs?: number; timeoutMs?: number; initialTask?: GenerationTask; onTaskUpdate?: (task: GenerationTask) => void }) {
+export type WaitForGenerationTaskOptions = {
+    signal?: AbortSignal;
+    intervalMs?: number;
+    timeoutMs?: number;
+    initialTask?: GenerationTask;
+    onTaskUpdate?: (task: GenerationTask) => void;
+    onTextDelta?: (text: string) => void;
+    useTextEvents?: boolean;
+};
+
+export async function waitForGenerationTask(id: string, options?: WaitForGenerationTaskOptions) {
    if (isLocalDreaminaTaskId(id)) {
        try {
            const task = await waitForLocalGenerationTask(id, { signal: options?.signal });
@@ -506,6 +521,7 @@ export async function waitForGenerationTask(id: string, options?: { signal?: Abo
            throw error;
        }
    }
+    if (options?.onTextDelta || options?.useTextEvents) return waitForGenerationTaskTextEvents(id, options);
    const startedAt = Date.now();
    const intervalMs = options?.intervalMs || 2000;
    let lastTask = options?.initialTask;
@@ -544,6 +560,129 @@ export async function waitForGenerationTask(id: string, options?: { signal?: Abo
    throw new Error(lastQueryError instanceof Error ? `任务状态同步失败：${lastQueryError.message}` : "任务执行超时，请稍后重试");
 }

+async function waitForGenerationTaskTextEvents(id: string, options: WaitForGenerationTaskOptions) {
+    const startedAt = Date.now();
+    const intervalMs = options.intervalMs || 1000;
+    let lastTask = options.initialTask;
+    let lastEventId = 0;
+    let fullText = lastTask?.textDraft || "";
+    let lastStreamError: unknown;
+    if (!lastTask) {
+        lastTask = await queryGenerationTask(id, { signal: options.signal });
+        options.onTaskUpdate?.(lastTask);
+    }
+    const timeoutMs = options.timeoutMs || taskWaitTimeoutMs(lastTask);
+    while (Date.now() - startedAt < timeoutMs) {
+        if (options.signal?.aborted) throw new DOMException("Aborted", "AbortError");
+        let terminalReceived = false;
+        try {
+            const base = String(apiBaseURL).replace(/\/+$/, "");
+            const cursor = lastEventId > 0 ? `?after=${encodeURIComponent(String(lastEventId))}` : "";
+            const response = await fetch(`${base}/tasks/${encodeURIComponent(id)}/text-events${cursor}`, {
+                headers: { Accept: "text/event-stream" },
+                credentials: "include",
+                signal: options.signal,
+            });
+            if (!response.ok) throw new TaskTextStreamFatalError(await taskTextStreamHTTPError(response));
+            if (!(response.headers.get("content-type") || "").toLowerCase().includes("text/event-stream")) {
+                throw new TaskTextStreamFatalError("任务文本流接口未返回事件流，请检查后端地址和反向代理配置");
+            }
+            if (!response.body) throw new Error("任务文本流不可用");
+            const reader = response.body.getReader();
+            const decoder = new TextDecoder();
+            const parser = createTaskTextStreamParser();
+            const onEvent = (event: TaskTextStreamEvent) => {
+                const payload = asTaskTextStreamRecord(event.data);
+                if (event.event === "delta") {
+                    const sequence = numberValue(payload.sequence) || event.id || 0;
+                    if (sequence <= lastEventId) return;
+                    const content = typeof payload.content === "string" ? payload.content : "";
+                    lastEventId = sequence;
+                    if (!content) return;
+                    fullText += content;
+                    options.onTextDelta?.(fullText);
+                    if (lastTask) {
+                        lastTask = { ...lastTask, textDraft: fullText };
+                        options.onTaskUpdate?.(lastTask);
+                    }
+                    return;
+                }
+                if (event.event === "progress") {
+                    if (!lastTask) return;
+                    lastTask = {
+                        ...lastTask,
+                        ...(isTaskStatus(payload.status) ? { status: payload.status } : {}),
+                        ...(typeof payload.stage === "string" ? { stage: payload.stage } : {}),
+                        ...(numberValue(payload.progress) !== undefined ? { progress: numberValue(payload.progress) } : {}),
+                    };
+                    options.onTaskUpdate?.(lastTask);
+                    return;
+                }
+                if (event.event === "terminal") {
+                    terminalReceived = true;
+                    if (typeof payload.finalText === "string" && payload.finalText !== fullText) {
+                        fullText = payload.finalText;
+                        options.onTextDelta?.(fullText);
+                    }
+                    return;
+                }
+                if (event.event === "error") throw new Error(typeof payload.message === "string" ? payload.message : "任务文本流不可用");
+            };
+            for (;;) {
+                const { done, value } = await reader.read();
+                if (done) break;
+                consumeTaskTextStream(parser, decoder.decode(value, { stream: true }), onEvent);
+            }
+            consumeTaskTextStream(parser, decoder.decode(), onEvent, true);
+            if (terminalReceived) {
+                const completed = await queryGenerationTask(id, { signal: options.signal });
+                options.onTaskUpdate?.(completed);
+                window.dispatchEvent(new CustomEvent("wallet:updated"));
+                if (completed.status === "succeeded") return completed;
+                if (completed.status === "failed" || completed.status === "cancelled") {
+                    throw new TaskTextStreamFatalError(completed.error ? generationErrorMessage(completed.error) : `任务${completed.status === "cancelled" ? "已取消" : "失败"}`);
+                }
+            }
+            lastStreamError = new Error("任务文本流连接提前结束");
+        } catch (error) {
+            if (options.signal?.aborted) throw new DOMException("Aborted", "AbortError");
+            if (error instanceof TaskTextStreamFatalError) throw error;
+            lastStreamError = error;
+        }
+        await delay(intervalMs, options.signal);
+    }
+    throw new Error(lastStreamError instanceof Error ? `任务文本流同步失败：${lastStreamError.message}` : "任务执行超时，请稍后重试");
+}
+
+function asTaskTextStreamRecord(value: unknown): Record<string, unknown> {
+    return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
+}
+
+function numberValue(value: unknown) {
+    return typeof value === "number" && Number.isFinite(value) ? value : undefined;
+}
+
+function isTaskStatus(value: unknown): value is TaskStatus {
+    return value === "queued" || value === "running" || value === "succeeded" || value === "failed" || value === "cancelled";
+}
+
+class TaskTextStreamFatalError extends Error {
+    constructor(message: string) {
+        super(message);
+        this.name = "TaskTextStreamFatalError";
+    }
+}
+
+async function taskTextStreamHTTPError(response: Response) {
+    try {
+        const payload = await response.json() as { msg?: unknown };
+        if (typeof payload.msg === "string" && payload.msg.trim()) return payload.msg;
+    } catch {
+        // SSE 错误响应可能是空正文或网关 HTML，只返回状态码，避免泄露正文。
+    }
+    return `任务文本流请求失败（HTTP ${response.status}）`;
+}
+
 function taskWaitTimeoutMs(task?: GenerationTask) {
    const type = task?.type || "";
    if (type.includes("storyboard")) return 13 * 60 * 1000;
--- a/web/src/services/api/task-text-stream.ts
+++ b/web/src/services/api/task-text-stream.ts
@@ -0,0 +1,57 @@
+export type TaskTextStreamEvent = {
+    event: string;
+    id?: number;
+    data?: unknown;
+};
+
+export type TaskTextStreamParser = {
+    buffer: string;
+};
+
+export function createTaskTextStreamParser(): TaskTextStreamParser {
+    return { buffer: "" };
+}
+
+export function consumeTaskTextStream(
+    parser: TaskTextStreamParser,
+    chunk: string,
+    onEvent: (event: TaskTextStreamEvent) => void,
+    flush = false,
+) {
+    parser.buffer += chunk;
+    for (;;) {
+        const boundary = parser.buffer.match(/\r?\n\r?\n/);
+        if (!boundary) break;
+        const index = boundary.index ?? 0;
+        emitTaskTextStreamBlock(parser.buffer.slice(0, index), onEvent);
+        parser.buffer = parser.buffer.slice(index + boundary[0].length);
+    }
+    if (flush && parser.buffer.trim()) {
+        emitTaskTextStreamBlock(parser.buffer, onEvent);
+        parser.buffer = "";
+    }
+}
+
+function emitTaskTextStreamBlock(block: string, onEvent: (event: TaskTextStreamEvent) => void) {
+    let event = "message";
+    let id: number | undefined;
+    const data: string[] = [];
+    for (const line of block.split(/\r?\n/)) {
+        if (!line || line.startsWith(":")) continue;
+        const separator = line.indexOf(":");
+        const field = separator >= 0 ? line.slice(0, separator) : line;
+        const value = separator >= 0 ? line.slice(separator + 1).replace(/^ /, "") : "";
+        if (field === "event") event = value || "message";
+        else if (field === "id" && /^\d+$/.test(value)) id = Number(value);
+        else if (field === "data") data.push(value);
+    }
+    if (!data.length) return;
+    const raw = data.join("\n");
+    let parsed: unknown;
+    try {
+        parsed = JSON.parse(raw);
+    } catch {
+        throw new Error("任务文本流返回了无效数据");
+    }
+    onEvent({ event, ...(id !== undefined ? { id } : {}), data: parsed });
+}
--- a/web/test/task-text-stream.test.ts
+++ b/web/test/task-text-stream.test.ts
@@ -0,0 +1,27 @@
+import { describe, expect, test } from "bun:test";
+
+import { consumeTaskTextStream, createTaskTextStreamParser, type TaskTextStreamEvent } from "../src/services/api/task-text-stream";
+
+describe("task text SSE parser", () => {
+    test("reassembles split frames and keeps event ids for reconnect cursors", () => {
+        const parser = createTaskTextStreamParser();
+        const events: TaskTextStreamEvent[] = [];
+        const stream = `: connected\n\nevent: progress\ndata: {"status":"running","stage":"生成中","progress":30}\n\nid: 8\nevent: delta\ndata: {"sequence":8,"content":"第一段"}\n\nid: 9\nevent: delta\ndata: {"sequence":9,"content":"第二段"}\n\nevent: terminal\ndata: {"status":"succeeded","complete":true,"finalText":"第一段第二段"}\n\n`;
+
+        consumeTaskTextStream(parser, stream.slice(0, 73), (event) => events.push(event));
+        consumeTaskTextStream(parser, stream.slice(73), (event) => events.push(event), true);
+
+        expect(events.map((event) => [event.event, event.id])).toEqual([
+            ["progress", undefined],
+            ["delta", 8],
+            ["delta", 9],
+            ["terminal", undefined],
+        ]);
+        expect(events[2]?.data).toEqual({ sequence: 9, content: "第二段" });
+    });
+
+    test("rejects malformed JSON instead of silently dropping task output", () => {
+        const parser = createTaskTextStreamParser();
+        expect(() => consumeTaskTextStream(parser, "event: delta\ndata: {broken}\n\n", () => undefined)).toThrow("任务文本流返回了无效数据");
+    });
+});