diff --git a/backend/agent-runtime/pi/agent-runtime.mjs b/backend/agent-runtime/pi/agent-runtime.mjs index fa13ee101..4e87c810e 100644 --- a/backend/agent-runtime/pi/agent-runtime.mjs +++ b/backend/agent-runtime/pi/agent-runtime.mjs @@ -193,6 +193,7 @@ async function run() { content: [{ type: "text", text: result.content || "操作正在等待用户审批。" }], details: { approvalId: result.approvalId, paused: true }, isError: true, + terminate: true, }; } // 工具结果必须是非空文本：交回 "null" 会让模型返回空回复。 @@ -202,6 +203,7 @@ async function run() { content: [{ type: "text", text }], details: result.details, isError: Boolean(result.isError), + terminate: Boolean(result.terminate), }; }, })); diff --git a/backend/internal/app/cloud_agent_pi_coordinator.go b/backend/internal/app/cloud_agent_pi_coordinator.go index 5fd3fbff7..dfb96be82 100644 --- a/backend/internal/app/cloud_agent_pi_coordinator.go +++ b/backend/internal/app/cloud_agent_pi_coordinator.go @@ -692,7 +692,11 @@ func (s *Service) cloudAgentPiModel(ctx context.Context, userID, runID string, p // 上游临时故障（5xx、429、超时、连接错误、空回复）自动重试，首次失败后最多再试 3 次， // 指数退避。参数/鉴权类 4xx 重试也不会变好，直接失败。 + // 模型吐出损坏的工具参数 JSON（截断或多一个括号）时，那次调用并没有被执行： + // 与 Go 执行循环的 correctCloudAgentTruncatedCalls 一致，附上 truncated_tool_arguments + // 纠偏上下文重做这一步，而不是把整轮判死。次数与临时故障共用同一个上限。 var lastErr error + var correction []map[string]any for attempt := 0; attempt <= cloudAgentModelStepRetries; attempt++ { if attempt > 0 { delay := time.Duration(1<<(attempt-1)) * time.Second @@ -703,7 +707,7 @@ func (s *Service) cloudAgentPiModel(ctx context.Context, userID, runID string, p case <-time.After(delay): } } - result, retryable, err := s.runCloudAgentModelStep(ctx, userID, runID, request.Messages, request.ThinkingLevel) + result, retryable, err := s.runCloudAgentModelStep(ctx, userID, runID, request.Messages, request.ThinkingLevel, correction...) if err == nil { return result, nil } @@ -711,6 +715,9 @@ func (s *Service) cloudAgentPiModel(ctx context.Context, userID, runID string, p if !retryable { return nil, err } + if errors.Is(err, errCloudAgentTruncatedToolArguments) && correction == nil { + correction = []map[string]any{cloudAgentRuntimeMessage(cloudAgentRuntimeContext{Kind: cloudAgentContextTruncatedArguments})} + } } return nil, lastErr } @@ -718,8 +725,12 @@ func (s *Service) cloudAgentPiModel(ctx context.Context, userID, runID string, p // cloudAgentModelStepRetries 是单步模型调用首次失败后的最大重试次数。 const cloudAgentModelStepRetries = 3 +// errCloudAgentTruncatedToolArguments 标记"模型返回的工具参数不是完整 JSON"： +// 调用未执行，可以带纠偏上下文重做同一步。 +var errCloudAgentTruncatedToolArguments = errors.New("truncated tool arguments") + // runCloudAgentModelStep 调度并等待一次模型步骤；第二个返回值表示失败是否值得重试。 -func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID string, messages []map[string]any, thinkingLevel string) (any, bool, error) { +func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID string, messages []map[string]any, thinkingLevel string, correction ...map[string]any) (any, bool, error) { // 运行时发起 /model 的同时会并发推送事件（message_start、session_snapshot 等）， // 这些事件也会推进运行 revision。调度模型步骤是 CAS 写入：冲突时重新读取 // 最新状态再调度，而不是把整轮判失败。冲突时事务整体回滚，不会产生任务或扣费。 @@ -755,11 +766,22 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri } canonical := canonicalFromRuntimeMessages(messages, state.Canonical.Tools, state.Canonical.SystemPrompt) + // A continuation starts a fresh Pi process. If its session snapshot was + // unavailable, the runtime transcript is shorter than the server-authored + // canonical history. Reuse canonical directly so provider tool_calls are + // not accidentally converted as if they were Pi content blocks. + if len(state.Canonical.Messages) > len(messages) { + canonical = state.Canonical + } canonical.PromptCacheKey = state.Canonical.PromptCacheKey if len(canonical.Messages) == 0 { return nil, false, fmt.Errorf("no canonical messages") } state.Canonical = canonical + if len(correction) > 0 { + // 纠偏上下文只用于这一次请求，不写回运行状态，避免下一步重复出现。 + canonical.Messages = append(append([]map[string]any(nil), canonical.Messages...), correction...) + } input := map[string]any{ "mode": "text", "prompt": state.Request.Prompt, @@ -804,13 +826,20 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri taskID := state.ActiveTaskID task, err := s.waitCloudAgentTask(ctx, taskID) if err != nil { - retryable := ctx.Err() == nil && s.cloudAgentModelTaskRetryable(taskID) + truncated := false + if failed, lookupErr := s.repo.Task(taskID); lookupErr == nil && failed != nil && failed.Status != model.TaskStatusSucceeded { + truncated = cloudAgentTruncatedToolArguments(failed) + } + retryable := ctx.Err() == nil && (truncated || s.cloudAgentModelTaskRetryable(taskID)) if retryable { // 释放失败的步骤，下一次重试才能重新入队。 if releaseErr := s.finishCloudAgentPiModelStep(userID, runID, taskID, "", ""); releaseErr != nil { return nil, false, releaseErr } } + if truncated { + err = fmt.Errorf("%w: %v", errCloudAgentTruncatedToolArguments, err) + } return nil, retryable, err } var result struct { @@ -1158,7 +1187,15 @@ func (s *Service) executeCloudAgentRuntimeTool(ctx context.Context, userID, runI if trimmed := strings.TrimSpace(content); trimmed == "" || trimmed == "null" { content = `{"error":"工具没有返回结果"}` } - return map[string]any{"content": content, "isError": cloudAgentToolContentIsError(content)}, nil + result := map[string]any{"content": content, "isError": cloudAgentToolContentIsError(content)} + if call.Function.Name == "ask_user" { + var payload map[string]any + if json.Unmarshal([]byte(content), &payload) == nil && stringValue(payload["phase"]) == "question" { + // ask_user 已经把本轮交给用户，Pi 不能再发起下一次模型调用。 + result["terminate"] = true + } + } + return result, nil } if state.MediaTaskID == "" { return nil, fmt.Errorf("tool %s produced no result", call.Function.Name) diff --git a/backend/internal/app/cloud_agent_pi_truncated_args_test.go b/backend/internal/app/cloud_agent_pi_truncated_args_test.go new file mode 100644 index 000000000..a4f5779b8 --- /dev/null +++ b/backend/internal/app/cloud_agent_pi_truncated_args_test.go @@ -0,0 +1,119 @@ +package app + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "infinite-canvas/backend/internal/model" +) + +// 线上复现（诊断号 ag5865c256d9fb9bcbf2f1719a4ec1023c）：模型一次写了很长的工具参数， +// 流式拼出来的 JSON 多了一个括号，任务报"Agent 工具参数不是完整 JSON"。Go 执行循环 +// 会带 truncated_tool_arguments 纠偏重做，但 Pi 模型桥接把它当成不可重试的失败，整轮 +// 以"Agent 执行中断"结束。修复后桥接必须带纠偏上下文重做同一步。 +func TestCloudAgentPiModelRetriesTruncatedToolArguments(t *testing.T) { + s, db, _ := agentMediaFixture(t) + s.disablePiRuntime = true + req := agentTestRequest() + req.PermissionMode = "auto" + root, err := s.CreateCloudAgentRun("user", req, "") + if err != nil { + t.Fatal(err) + } + runID := root.ID + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + payload := map[string]json.RawMessage{"messages": json.RawMessage(`[{"role":"user","content":"继续"}]`), "thinkingLevel": json.RawMessage(`"off"`)} + type outcome struct { + result any + err error + } + done := make(chan outcome, 1) + go func() { + result, stepErr := s.cloudAgentPiModel(ctx, "user", runID, payload) + done <- outcome{result, stepErr} + }() + + // 等待模型步骤入队，返回它的任务 ID（跳过已处理过的任务）。 + nextStep := func(skip string) string { + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + stored, readErr := s.repo.CloudAgent("user", runID) + if readErr != nil { + t.Fatal(readErr) + } + state, decodeErr := cloudAgentDecode(stored) + if decodeErr != nil { + t.Fatal(decodeErr) + } + if state.ActiveTaskID != "" && state.ActiveTaskID != skip { + return state.ActiveTaskID + } + select { + case got := <-done: + t.Fatalf("model bridge returned before retry: result=%v err=%v", got.result, got.err) + case <-time.After(50 * time.Millisecond): + } + } + t.Fatal("model step was never scheduled") + return "" + } + + first := nextStep("") + if err := db.Model(&model.Task{}).Where("id = ?", first).Updates(map[string]any{ + "status": model.TaskStatusFailed, + "error": "Agent 工具参数不是完整 JSON：invalid character '}' after array element", + }).Error; err != nil { + t.Fatal(err) + } + + second := nextStep(first) + var retried model.Task + if err := db.First(&retried, "id = ?", second).Error; err != nil { + t.Fatal(err) + } + if !strings.Contains(retried.InputJSON, string(cloudAgentContextTruncatedArguments)) { + t.Fatalf("retried step lacks the truncated_tool_arguments correction: %s", retried.InputJSON) + } + if err := db.Model(&model.Task{}).Where("id = ?", second).Updates(map[string]any{ + "status": model.TaskStatusSucceeded, + "result_json": `{"text":"已拆分写入","toolCalls":[]}`, + }).Error; err != nil { + t.Fatal(err) + } + + select { + case got := <-done: + if got.err != nil { + t.Fatalf("model bridge failed after correction: %v", got.err) + } + result, _ := got.result.(map[string]any) + if result["text"] != "已拆分写入" { + t.Fatalf("unexpected result: %#v", got.result) + } + case <-ctx.Done(): + t.Fatal("model bridge did not return") + } + + stored, err := s.repo.CloudAgent("user", runID) + if err != nil { + t.Fatal(err) + } + if stored.Status == "failed" { + t.Fatalf("run failed: %q", stored.FailureMessage) + } + state, err := cloudAgentDecode(stored) + if err != nil { + t.Fatal(err) + } + // 纠偏上下文只用于那一次请求，不能写回运行历史。 + for _, message := range state.Canonical.Messages { + if strings.Contains(stringField(message, "content"), string(cloudAgentContextTruncatedArguments)) { + t.Fatal("correction context leaked into persisted canonical history") + } + } +} diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index 78ae69baf..3cbb8c89e 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -287,6 +287,11 @@ func (s *Service) ensureCloudAgentExecution(task *model.Task, initial cloudAgent } state.event(task.ID, "tool_completed", map[string]any{"toolName": "skills_load", "skillIds": skillIDs, "text": fmt.Sprintf("已启用 %d 个技能，正文将按需读取", len(initial.Skills))}) } + if len(state.Plan) > 0 { + // 继承的待办清单必须挂到本轮：前端按 plan- 展示清单，不发这条事件时 + // 界面会一直停在上一轮（已结束）的清单上，显示"未完成项已停止"。 + state.event(task.ID, "plan_updated", map[string]any{"items": state.Plan, "pendingTitles": cloudAgentPendingPlanItems(state.Plan), "inherited": true}) + } pressure := s.cloudAgentContextPressure(input.Requests.Canonical, initial.Request.Prompt, initial.Request) if carrier { // The carrier is deliberately not a model call. Pi will create the first 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 f7ee756b3..dfe9f6687 100644 --- a/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx +++ b/web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx @@ -934,9 +934,25 @@ export function AgentWorkingMessage({ theme, label = WORKING_TEXT }: { theme: (t ); } -export function AgentPlanBar({ items, theme, minimized, onToggle, terminal = false }: { items: CloudAgentPlanItem[]; theme: (typeof canvasThemes)[keyof typeof canvasThemes]; minimized: boolean; onToggle: () => void; terminal?: boolean }) { +export function AgentPlanBar({ + items, + theme, + minimized, + onToggle, + terminal: runEnded = false, + waitingUser = false, +}: { + items: CloudAgentPlanItem[]; + theme: (typeof canvasThemes)[keyof typeof canvasThemes]; + minimized: boolean; + onToggle: () => void; + terminal?: boolean; + /** 本轮停在 ask_user 等用户拍板：清单是暂停，不是停止。 */ + waitingUser?: boolean; +}) { const doneCount = items.filter((entry) => entry.status === "done").length; const allDone = doneCount === items.length; + const terminal = runEnded && !waitingUser; return (