From 201f3aac656aeb682dfdfd7a625a361bf686e96b Mon Sep 17 00:00:00 2001
From: ddCat <2695092045@qq.com>
Date: Tue, 15 Sep 2026 05:22:07 +0800
Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E4=BF=AE=E5=A4=8D=E6=B5=81?=
=?UTF-8?q?=E5=BC=8F=E6=8E=A8=E7=90=86=E4=B8=8E=E5=B7=A5=E5=85=B7=E5=8F=82?=
=?UTF-8?q?=E6=95=B0=E9=87=8D=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../app/cloud_agent_approval_preview.go | 2 +-
backend/internal/app/cloud_agent_json.go | 10 +++
backend/internal/app/cloud_agent_runtime.go | 9 ++
backend/internal/app/cloud_agent_stream.go | 66 ++++++++++++++
backend/internal/app/cloud_agent_tools.go | 5 +-
backend/internal/app/provider.go | 39 ++++----
.../internal/app/provider_agent_protocol.go | 21 +++--
backend/internal/app/provider_protocol.go | 8 ++
backend/internal/app/provider_text.go | 89 ++++++++++++++-----
backend/internal/app/task_text_stream.go | 9 +-
backend/internal/repository/cloud_agent.go | 6 ++
.../canvas/canvas-cloud-agent-chat-ui.tsx | 7 ++
.../canvas/canvas-cloud-agent-panel.tsx | 6 ++
13 files changed, 231 insertions(+), 46 deletions(-)
create mode 100644 backend/internal/app/cloud_agent_stream.go
diff --git a/backend/internal/app/cloud_agent_approval_preview.go b/backend/internal/app/cloud_agent_approval_preview.go
index 580437cfd..bf6e57ce1 100644
--- a/backend/internal/app/cloud_agent_approval_preview.go
+++ b/backend/internal/app/cloud_agent_approval_preview.go
@@ -54,7 +54,7 @@ type cloudAgentCanvasMutationPlan struct {
func prepareCloudAgentCanvasMutation(repo *repository.Repository, userID, canvasID string, call cloudAgentCall) (*cloudAgentCanvasMutationPlan, error) {
var args agentCanvasArgs
if err := decodeCloudAgentJSONObject(call.Function.Arguments, &args); err != nil {
- return nil, BadAuthRequest("画布工具参数必须是只含支持字段的单个JSON对象")
+ return nil, canvasArgumentError()
}
if len(args.Ops) < 1 || len(args.Ops) > 20 || args.SnapshotHash == "" {
return nil, BadAuthRequest("画布操作数量或快照无效")
diff --git a/backend/internal/app/cloud_agent_json.go b/backend/internal/app/cloud_agent_json.go
index 00fc62182..d8cfa7dea 100644
--- a/backend/internal/app/cloud_agent_json.go
+++ b/backend/internal/app/cloud_agent_json.go
@@ -9,6 +9,16 @@ import (
var errCloudAgentJSONSingleObject = errors.New("参数必须是单个 JSON 对象")
+// Only syntax/schema errors may be repaired by the model; authorization and
+// unsupported mutations still fail admission before any write or approval.
+type cloudAgentArgumentError struct{ error }
+
+func (e *cloudAgentArgumentError) Unwrap() error { return e.error }
+
+func canvasArgumentError() error {
+ return &cloudAgentArgumentError{BadAuthRequest("画布工具参数无效：仅允许一个 JSON 对象；顶层只含 snapshotHash 和 ops，snapshotHash 不得放入 ops。请按工具 schema 修正后重试")}
+}
+
// decodeCloudAgentJSONObject is used for model tool arguments. Tool arguments
// are an untrusted protocol boundary: reject non-objects, unknown fields and
// trailing JSON instead of silently accepting an ambiguous payload.
diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go
index a6c1c578a..31c88cd28 100644
--- a/backend/internal/app/cloud_agent_runtime.go
+++ b/backend/internal/app/cloud_agent_runtime.go
@@ -507,6 +507,7 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error)
}
var result struct {
Text string `json:"text"`
+ Reasoning string `json:"reasoning"`
ToolCalls []cloudAgentCall `json:"toolCalls"`
Legacy []cloudAgentCall `json:"tool_calls"`
}
@@ -534,6 +535,9 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error)
return cloudAgentSave(current, &state)
}
calls := result.ToolCalls
+ if result.Reasoning != "" {
+ state.event(run.ID, "reasoning_message", map[string]any{"messageId": task.ID + ":reasoning", "text": truncateRunes(result.Reasoning, 8000)})
+ }
if result.Text != "" {
state.event(run.ID, "assistant_message", map[string]any{"messageId": task.ID, "text": result.Text})
if len(calls) == 0 {
@@ -784,6 +788,11 @@ func (s *Service) advanceCloudAgentTool(run *model.CloudAgentExecution, state *c
} else {
canvasPlan, err := prepareCloudAgentCanvasMutation(repo, run.UserID, state.Request.CanvasID, call)
if err != nil {
+ var argumentErr *cloudAgentArgumentError
+ if errors.As(err, &argumentErr) {
+ cloudAgentToolResult(run.ID, state, call, nil, err)
+ return cloudAgentSave(current, state)
+ }
var appErr *AppError
if errors.As(err, &appErr) && appErr != nil {
return failCloudAgentAdmission(current, state, run.ID, err)
diff --git a/backend/internal/app/cloud_agent_stream.go b/backend/internal/app/cloud_agent_stream.go
new file mode 100644
index 000000000..71432bc1b
--- /dev/null
+++ b/backend/internal/app/cloud_agent_stream.go
@@ -0,0 +1,66 @@
+package app
+
+import (
+ "errors"
+ "unicode/utf8"
+
+ "gorm.io/gorm"
+ "infinite-canvas/backend/internal/model"
+ "infinite-canvas/backend/internal/repository"
+)
+
+// Publish upstream deltas to the same durable, revisioned transcript used by
+// Agent SSE/replay. Each flush re-reads state; never overwrite a scheduler's
+// checkpoint with the snapshot from when the model request started.
+func newCloudAgentStreamPublisher(s *Service, userID, taskID, kind string) *taskTextStreamPublisher {
+ p := newTaskTextStreamPublisher(s, userID, taskID)
+ written := 0
+ p.sink = func(delta string) error {
+ remaining := 32000 - written
+ if remaining <= 0 {
+ return nil
+ }
+ if len(delta) > remaining {
+ delta = delta[:remaining]
+ for !utf8.ValidString(delta) {
+ delta = delta[:len(delta)-1]
+ }
+ }
+ if delta == "" {
+ return nil
+ }
+ for attempt := 0; attempt < 3; attempt++ {
+ run, err := s.repo.CloudAgentForActiveTask(userID, taskID)
+ if errors.Is(err, gorm.ErrRecordNotFound) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ err = s.repo.MutateCloudAgent(userID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error {
+ state, err := cloudAgentDecode(current)
+ if err != nil {
+ return err
+ }
+ if state.ActiveTaskID != taskID || (current.Status != "running" && current.Status != "queued") {
+ return nil
+ }
+ messageID := taskID
+ if kind == "reasoning_delta" {
+ messageID += ":reasoning"
+ }
+ state.event(run.ID, kind, map[string]any{"messageId": messageID, "text": delta})
+ return cloudAgentSave(current, &state)
+ })
+ if errors.Is(err, repository.ErrCreationConflict) {
+ continue
+ }
+ if err == nil {
+ written += len(delta)
+ }
+ return err
+ }
+ return repository.ErrCreationConflict
+ }
+ return p
+}
diff --git a/backend/internal/app/cloud_agent_tools.go b/backend/internal/app/cloud_agent_tools.go
index bc154ead0..46c6ccff5 100644
--- a/backend/internal/app/cloud_agent_tools.go
+++ b/backend/internal/app/cloud_agent_tools.go
@@ -1,6 +1,7 @@
package app
import (
+ "crypto/sha256"
"encoding/json"
"fmt"
"sort"
@@ -103,7 +104,9 @@ func cloudAgentCanonical(system string, history []providerTextMessage, prompt st
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
}
messages = append(messages, map[string]any{"role": "user", "content": prompt})
- return canonicalAgentRequest{SystemPrompt: system, Messages: messages, Tools: cloudAgentTools(req), ToolChoice: "auto"}
+ // Keep routing stable across tool turns, without exposing canvas identifiers.
+ cacheHash := sha256.Sum256([]byte(req.CanvasID + "\x00" + system))
+ return canonicalAgentRequest{SystemPrompt: system, Messages: messages, Tools: cloudAgentTools(req), ToolChoice: "auto", PromptCacheKey: fmt.Sprintf("cloud-agent:%x", cacheHash[:24])}
}
func cloudAgentTools(req CloudAgentRequest) []map[string]any {
tools := []map[string]any{}
diff --git a/backend/internal/app/provider.go b/backend/internal/app/provider.go
index 2f1f7b2d9..654f122d5 100644
--- a/backend/internal/app/provider.go
+++ b/backend/internal/app/provider.go
@@ -25,22 +25,23 @@ import (
var sseFrameBoundaryPattern = regexp.MustCompile(`\r?\n\r?\n`)
type canvasGenerationInput struct {
- Mode string `json:"mode"`
- Prompt string `json:"prompt"`
- Config providerConfig `json:"config"`
- ReferenceImages []providerMedia `json:"referenceImages"`
- ReferenceVideos []providerMedia `json:"referenceVideos"`
- ReferenceAudios []providerMedia `json:"referenceAudios"`
- TextHistory []providerTextMessage `json:"textHistory"`
- Mask *providerMedia `json:"mask"`
- Metadata map[string]interface{} `json:"metadata"`
- AgentRequests *agentToolRequests `json:"agentRequests"`
- TextOptions canvasTextOptions `json:"textOptions"`
- ImageCapability *ImageCapabilityConfig `json:"-"`
- StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。
- MaxOutputTokens int `json:"-"`
- OnTextDelta func(string) `json:"-"`
- VideoCapability *VideoCapabilityConfig `json:"-"`
+ Mode string `json:"mode"`
+ Prompt string `json:"prompt"`
+ Config providerConfig `json:"config"`
+ ReferenceImages []providerMedia `json:"referenceImages"`
+ ReferenceVideos []providerMedia `json:"referenceVideos"`
+ ReferenceAudios []providerMedia `json:"referenceAudios"`
+ TextHistory []providerTextMessage `json:"textHistory"`
+ Mask *providerMedia `json:"mask"`
+ Metadata map[string]interface{} `json:"metadata"`
+ AgentRequests *agentToolRequests `json:"agentRequests"`
+ TextOptions canvasTextOptions `json:"textOptions"`
+ ImageCapability *ImageCapabilityConfig `json:"-"`
+ StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。
+ MaxOutputTokens int `json:"-"`
+ OnTextDelta func(string) `json:"-"`
+ OnReasoningDelta func(string) `json:"-"`
+ VideoCapability *VideoCapabilityConfig `json:"-"`
}
type canvasTextOptions struct {
@@ -359,6 +360,12 @@ func (s *Service) processCanvasGenerationTask(ctx context.Context, userID string
}
if input.Mode == "text" && strings.HasPrefix(taskType, "canvas_text") && input.StreamText {
textPublisher = newTaskTextStreamPublisher(s, userID, taskExecutionID(ctx))
+ if input.AgentRequests != nil {
+ textPublisher = newCloudAgentStreamPublisher(s, userID, taskExecutionID(ctx), "assistant_delta")
+ reasoningPublisher := newCloudAgentStreamPublisher(s, userID, taskExecutionID(ctx), "reasoning_delta")
+ input.OnReasoningDelta = reasoningPublisher.Publish
+ defer reasoningPublisher.Close()
+ }
input.OnTextDelta = textPublisher.Publish
defer textPublisher.Close()
}
diff --git a/backend/internal/app/provider_agent_protocol.go b/backend/internal/app/provider_agent_protocol.go
index ec97a4a37..96dd9d6ca 100644
--- a/backend/internal/app/provider_agent_protocol.go
+++ b/backend/internal/app/provider_agent_protocol.go
@@ -11,10 +11,11 @@ import (
// The browser sends one protocol-neutral conversation. Only the selected
// provider body is materialized; declarative plugins retain their request paths.
type canonicalAgentRequest struct {
- Messages []map[string]interface{} `json:"messages"`
- Tools []map[string]interface{} `json:"tools"`
- ToolChoice interface{} `json:"toolChoice"`
- SystemPrompt string `json:"systemPrompt"`
+ Messages []map[string]interface{} `json:"messages"`
+ Tools []map[string]interface{} `json:"tools"`
+ ToolChoice interface{} `json:"toolChoice"`
+ SystemPrompt string `json:"systemPrompt"`
+ PromptCacheKey string `json:"promptCacheKey,omitempty"`
}
func expandCanonicalAgentRequest(source *canonicalAgentRequest, config providerConfig, declarative bool) (*agentToolRequests, error) {
@@ -155,7 +156,11 @@ func canonicalAgentChatBody(source *canonicalAgentRequest, claude bool) map[stri
if named, ok := choice.(map[string]interface{}); ok {
choice = map[string]interface{}{"type": "function", "function": map[string]interface{}{"name": named["name"]}}
}
- return map[string]interface{}{"messages": messages, "tools": tools, "tool_choice": choice, "parallel_tool_calls": false}
+ body := map[string]interface{}{"messages": messages, "tools": tools, "tool_choice": choice, "parallel_tool_calls": false}
+ if !claude && source.PromptCacheKey != "" {
+ body["prompt_cache_key"] = source.PromptCacheKey
+ }
+ return body
}
func canonicalAgentResponsesBody(source *canonicalAgentRequest) map[string]interface{} {
@@ -183,7 +188,11 @@ func canonicalAgentResponsesBody(source *canonicalAgentRequest) map[string]inter
converted["type"] = "function"
tools = append(tools, converted)
}
- return map[string]interface{}{"input": messages, "tools": tools, "tool_choice": source.ToolChoice, "parallel_tool_calls": false}
+ body := map[string]interface{}{"input": messages, "tools": tools, "tool_choice": source.ToolChoice, "parallel_tool_calls": false}
+ if source.PromptCacheKey != "" {
+ body["prompt_cache_key"] = source.PromptCacheKey
+ }
+ return body
}
func canonicalAgentGeminiBody(source *canonicalAgentRequest) map[string]interface{} {
diff --git a/backend/internal/app/provider_protocol.go b/backend/internal/app/provider_protocol.go
index c4b01106b..2e98b1098 100644
--- a/backend/internal/app/provider_protocol.go
+++ b/backend/internal/app/provider_protocol.go
@@ -319,6 +319,10 @@ func executeProtocolRequest(ctx context.Context, config providerConfig, spec pro
// method/path/body/auth；最终 URL 校验、凭证注入、SSRF、超时、大小限制和审计仍由宿主统一执行，
// 插件不能通过自定义请求规格绕过这些安全约束。
func executeProtocolBinaryRequest(ctx context.Context, config providerConfig, spec protocol.RequestSpec) ([]byte, string, error) {
+ return executeProtocolBinaryRequestWithConsumer(ctx, config, spec, nil)
+}
+
+func executeProtocolBinaryRequestWithConsumer(ctx context.Context, config providerConfig, spec protocol.RequestSpec, consume func(string, []byte)) ([]byte, string, error) {
if err := spec.Validate(); err != nil {
return nil, "", err
}
@@ -345,6 +349,10 @@ func executeProtocolBinaryRequest(ctx context.Context, config providerConfig, sp
if err := applyProtocolAuth(req, config, spec.Auth); err != nil {
return nil, "", err
}
+ if consume != nil {
+ req.Header.Set("Accept", "text/event-stream")
+ return doBinaryWithConsumer(req, consume)
+ }
return doBinary(req)
}
diff --git a/backend/internal/app/provider_text.go b/backend/internal/app/provider_text.go
index 30ca6da32..a31fac246 100644
--- a/backend/internal/app/provider_text.go
+++ b/backend/internal/app/provider_text.go
@@ -79,7 +79,7 @@ func runAgentToolTask(ctx context.Context, input canvasGenerationInput) (map[str
func postAgentRequest(ctx context.Context, input canvasGenerationInput, path string, body map[string]interface{}, protocol string) (map[string]interface{}, error) {
if input.StreamText {
- return postStreamingAgent(ctx, input.Config, path, body, protocol, input.OnTextDelta)
+ return postStreamingAgent(ctx, input.Config, path, body, protocol, input.OnTextDelta, input.OnReasoningDelta)
}
delete(body, "stream")
var payload map[string]interface{}
@@ -90,7 +90,12 @@ func postAgentRequest(ctx context.Context, input canvasGenerationInput, path str
}
func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, adapter protocol.AgentAdapter) (map[string]interface{}, error) {
- if input.TextOptions.Thinking {
+ wire := input.Config.InterfaceType
+ if wire == string(model.ChannelInterfaceOpenAIResponse) {
+ wire = "responses"
+ }
+ knownWire := wire == "chat-completion" || wire == "responses" || wire == "claude-api"
+ if input.TextOptions.Thinking && !knownWire {
return nil, errors.New("当前声明式 Agent 渠道尚不支持思考模式，请关闭思考模式或切换内置协议渠道")
}
if input.AgentRequests == nil {
@@ -106,6 +111,37 @@ func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, a
if err != nil {
return nil, err
}
+ if knownWire {
+ body := protocolBodyObject(spec.Body)
+ if body == nil {
+ return nil, errors.New("声明式 Agent 请求体必须是 JSON 对象")
+ }
+ applyTextThinking(body, input, wire)
+ spec.Body = body
+ if input.StreamText {
+ body["stream"] = true
+ if wire == "chat-completion" {
+ if err := ensureChatCompletionStreamUsage(body); err != nil {
+ return nil, err
+ }
+ }
+ parser := newStreamingAgentParser(wire, input.OnTextDelta)
+ parser.emitReasoning = input.OnReasoningDelta
+ data, mime, err := executeProtocolBinaryRequestWithConsumer(ctx, input.Config, spec, parser.consume)
+ if err != nil {
+ return nil, err
+ }
+ if strings.Contains(strings.ToLower(mime), "event-stream") {
+ parser.flush()
+ return parser.result()
+ }
+ var payload map[string]interface{}
+ if err := json.Unmarshal(data, &payload); err != nil {
+ return nil, fmt.Errorf("Agent 接口返回格式无效：%w", err)
+ }
+ return parseAgentToolPayload(payload, wire)
+ }
+ }
body, err := executeProtocolRequest(ctx, input.Config, spec)
if err != nil {
return nil, err
@@ -330,17 +366,17 @@ 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) {
+func postStreamingAgent(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string), onReasoning ...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
- }
+ if err := ensureChatCompletionStreamUsage(body); err != nil {
+ return nil, err
}
}
parser := newStreamingAgentParser(protocol, onDelta)
+ if len(onReasoning) > 0 {
+ parser.emitReasoning = onReasoning[0]
+ }
data, mimeType, err := postStreamingBinary(ctx, config, path, body, parser.consume)
if err != nil {
return nil, err
@@ -363,15 +399,16 @@ type streamingAgentToolCall struct {
}
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)
+ 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)
+ emitReasoning func(string)
}
func newStreamingAgentParser(protocol string, emit func(string)) *streamingAgentParser {
@@ -448,7 +485,7 @@ func (p *streamingAgentParser) consumeResponsesEvent(eventName string, payload m
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"))
+ p.appendReasoning(stringField(payload, "delta"))
case "response.completed":
p.completed, _ = payload["response"].(map[string]interface{})
case "response.output_item.added":
@@ -482,7 +519,7 @@ func (p *streamingAgentParser) consumeChatCompletionEvent(payload map[string]int
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")))
+ p.appendReasoning(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)
@@ -509,7 +546,7 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
case "text":
p.appendText(stringField(block, "text"))
case "thinking":
- p.reasoning.WriteString(firstNonEmptyString(stringField(block, "thinking"), stringField(block, "text")))
+ p.appendReasoning(firstNonEmptyString(stringField(block, "thinking"), stringField(block, "text")))
case "tool_use":
arguments := ""
if input := block["input"]; input != nil {
@@ -526,7 +563,7 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
p.appendText(stringField(delta, "text"))
}
if stringField(delta, "type") == "thinking_delta" {
- p.reasoning.WriteString(firstNonEmptyString(stringField(delta, "thinking"), stringField(delta, "text")))
+ p.appendReasoning(firstNonEmptyString(stringField(delta, "thinking"), stringField(delta, "text")))
}
if stringField(delta, "type") == "input_json_delta" {
p.toolCall(index).arguments += stringField(delta, "partial_json")
@@ -551,6 +588,16 @@ func (p *streamingAgentParser) appendText(delta string) {
}
}
+func (p *streamingAgentParser) appendReasoning(delta string) {
+ if delta == "" {
+ return
+ }
+ p.reasoning.WriteString(delta)
+ if p.emitReasoning != nil {
+ p.emitReasoning(delta)
+ }
+}
+
func (p *streamingAgentParser) toolCall(index int) *streamingAgentToolCall {
if index < 0 {
index = 0
diff --git a/backend/internal/app/task_text_stream.go b/backend/internal/app/task_text_stream.go
index e59e08abc..2c558f32a 100644
--- a/backend/internal/app/task_text_stream.go
+++ b/backend/internal/app/task_text_stream.go
@@ -23,6 +23,7 @@ type taskTextStreamPublisher struct {
timer *time.Timer
disabled bool
closed bool
+ sink func(string) error
}
func newTaskTextStreamPublisher(service *Service, userID string, taskID string) *taskTextStreamPublisher {
@@ -77,7 +78,13 @@ func (p *taskTextStreamPublisher) flushLocked() {
}
content := p.buffer.String()
p.buffer.Reset()
- if _, err := p.service.AppendTaskTextDelta(p.userID, p.taskID, content); err != nil {
+ var err error
+ if p.sink != nil {
+ err = p.sink(content)
+ } else {
+ _, err = p.service.AppendTaskTextDelta(p.userID, p.taskID, content)
+ }
+ if err != nil {
p.disabled = true
_ = p.service.log(p.userID, p.taskID, "warn", "文本流持久化已降级", err.Error())
}
diff --git a/backend/internal/repository/cloud_agent.go b/backend/internal/repository/cloud_agent.go
index 3b3a09c13..bf5bda5fb 100644
--- a/backend/internal/repository/cloud_agent.go
+++ b/backend/internal/repository/cloud_agent.go
@@ -16,6 +16,12 @@ func (r *Repository) CloudAgent(userID, id string) (*model.CloudAgentExecution,
err := r.db.First(&run, "id = ? AND user_id = ?", id, userID).Error
return &run, err
}
+
+func (r *Repository) CloudAgentForActiveTask(userID, taskID string) (*model.CloudAgentExecution, error) {
+ var run model.CloudAgentExecution
+ err := r.db.Where("user_id = ? AND active_task_id = ? AND status IN ?", userID, taskID, []string{"running", "queued"}).First(&run).Error
+ return &run, err
+}
func (r *Repository) CloudAgentRoots() ([]model.Task, error) {
var tasks []model.Task
err := r.db.Where("operation = ? AND id NOT IN (SELECT id FROM cloud_agent_executions)", "cloud_agent").Order("created_at").Limit(50).Find(&tasks).Error
diff --git a/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx b/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx
index 508ec3905..49e9fe269 100644
--- a/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx
+++ b/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx
@@ -30,6 +30,7 @@ export type CloudAgentChatMessage = {
title?: string;
text: string;
streaming?: boolean;
+ reasoning?: boolean;
meta?: string;
detail?: unknown;
attachments?: CloudAgentChatAttachment[];
@@ -91,6 +92,12 @@ export function AgentChatMessage({
const isUser = item.role === "user";
const isSystem = item.role === "system";
const color = item.role === "error" ? "#ef4444" : theme.node.text;
+ if (item.reasoning) {
+ return
+ {item.streaming ? "模型推理中" : "模型推理摘要"}
+ {item.text}
+ ;
+ }
if (isSystem) {
return (
diff --git a/web/src/components/canvas/canvas-cloud-agent-panel.tsx b/web/src/components/canvas/canvas-cloud-agent-panel.tsx
index 8e4343523..dc94c2c41 100644
--- a/web/src/components/canvas/canvas-cloud-agent-panel.tsx
+++ b/web/src/components/canvas/canvas-cloud-agent-panel.tsx
@@ -1090,6 +1090,12 @@ function applyAgentEvent(event: AgentEvent, setMessages: Dispatch upsertTextMessage(current, String(payload.messageId || "assistant"), text, true));
return;
}
+ if (event.type === "reasoning_delta" || event.type === "reasoning_message") {
+ const id = String(payload.messageId || `${event.runId}:reasoning`);
+ setMessages((current) => upsertTextMessage(current, id, text, event.type === "reasoning_delta")
+ .map((item) => item.id === id ? { ...item, reasoning: true } : item));
+ return;
+ }
if (event.type === "assistant_message") {
setMessages((current) => upsertTextMessage(current, String(payload.messageId || event.eventId), text, false));
return;