fix(agent): Agent 运行 - 截断工具参数纠偏重试并修复 ask_user 后续跑与清单状态

模型返回不完整的工具参数 JSON 时,Pi 桥接带 truncated_tool_arguments 纠偏上下文重做同一步,不再把整轮判为中断。ask_user 进入提问后结束本轮 Pi 调用;续轮会话快照缺失时改用服务端 canonical 历史。继承的待办清单挂到新一轮,等待用户确认时清单显示暂停而非停止。
This commit is contained in:
ddcat666
2026-09-30 03:41:25 +08:00
parent 10ea9d3fa0
commit 220cc543e1
6 changed files with 187 additions and 6 deletions

View File

@@ -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),
};
},
}));

View File

@@ -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)

View File

@@ -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")
}
}
}

View File

@@ -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-<runId> 展示清单,不发这条事件时
// 界面会一直停在上一轮(已结束)的清单上,显示"未完成项已停止"。
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

View File

@@ -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 (
<div className="agent-plan-bar mx-3 mb-2 overflow-hidden rounded-xl" style={{ color: theme.node.text }}>
<button type="button" className="flex w-full items-center gap-2 px-3 py-2 text-left focus-visible:outline focus-visible:outline-2" aria-expanded={!minimized} onClick={onToggle}>
@@ -945,6 +961,7 @@ export function AgentPlanBar({ items, theme, minimized, onToggle, terminal = fal
<span className="text-[11px] tabular-nums opacity-50">
{doneCount}/{items.length}
</span>
{waitingUser && !allDone ? <span className="shrink-0 text-[10px] opacity-55">等待你确认后继续</span> : null}
{terminal && !allDone ? <span className="shrink-0 text-[10px] opacity-55">本轮已结束,未完成项已停止</span> : null}
<span className="min-w-0 flex-1" />
<span className="shrink-0 text-[11px] opacity-50">{minimized ? "展开" : "收起"}</span>
@@ -959,7 +976,7 @@ export function AgentPlanBar({ items, theme, minimized, onToggle, terminal = fal
return (
<li key={entry.id} className="flex min-w-0 items-start gap-1.5 text-xs">
<Icon
className={doing && !terminal ? "mt-[3px] size-3 shrink-0 animate-spin" : "mt-[3px] size-3 shrink-0"}
className={doing && !terminal && !waitingUser ? "mt-[3px] size-3 shrink-0 animate-spin" : "mt-[3px] size-3 shrink-0"}
style={{ color: done ? "#429477" : terminal ? theme.node.muted : doing ? theme.accent.primary : theme.node.muted }}
/>
<span className={done ? "min-w-0 break-words line-through opacity-50" : terminal ? "min-w-0 break-words opacity-55" : "min-w-0 break-words"}>{entry.title}</span>

View File

@@ -1032,6 +1032,7 @@ export function CanvasCloudAgentPanel({ canvasId, domainProjectId, nodeCount, se
theme={theme}
minimized={planMinimized}
terminal={planTerminal || Boolean(run && ["completed", "failed", "cancelled", "rejected"].includes(run.status))}
waitingUser={Boolean(pendingQuestion)}
onToggle={() => setPlanMinimized((value) => !value)}
/>
) : null}