fix(agent): 审批恢复 - 补齐被审批暂停跳过的并行工具调用结果 (#670)
模型一步里发出多个工具调用、排在前面的进入用户审批时，运行时中止本轮，后面的调用既没执行也没有结果；批准后恢复运行，历史里留着有调用没结果的条目，整轮以「工具调用缺少结果」失败。转换前给这些调用补一条「未执行」结果：历史恢复合法，模型也知道那一步没做过，需要时会重新调用。 后端重启后恢复运行等场景下会被直接采用的服务端规范历史同样可能留着这类条目，一并补齐。
This commit is contained in:
jiabx
@@ -350,6 +350,10 @@ func (s *Service) adoptCloudAgentPiModelStep(userID, runID, taskID string) (adop
|
||||
|
||||
func cloudAgentPiCanonicalForModelRequest(messages []map[string]any, existing canonicalAgentRequest) (canonicalAgentRequest, error) {
|
||||
compactionSummaryRequest := cloudAgentPiCompactionSummaryRequest(messages)
|
||||
if !compactionSummaryRequest {
|
||||
// 审批暂停会留下没有结果的并行工具调用，先补齐再转换（见 cloud_agent_pi_unanswered_calls.go）。
|
||||
messages = repairRuntimeUnansweredCalls(messages)
|
||||
}
|
||||
tools, systemPrompt := existing.Tools, existing.SystemPrompt
|
||||
if compactionSummaryRequest {
|
||||
tools, systemPrompt = nil, ""
|
||||
@@ -365,11 +369,13 @@ func cloudAgentPiCanonicalForModelRequest(messages []map[string]any, existing ca
|
||||
if len(existing.Messages) == 0 {
|
||||
return canonicalAgentRequest{}, err
|
||||
}
|
||||
existing.Messages = repairCanonicalUnansweredCalls(existing.Messages)
|
||||
return existing, nil
|
||||
}
|
||||
// A projected Pi transcript containing a compaction summary is authoritative,
|
||||
// even though it is shorter than the server's pre-compaction canonical history.
|
||||
if !cloudAgentPiHasCompactionSummary(messages) && len(existing.Messages) > len(messages) {
|
||||
existing.Messages = repairCanonicalUnansweredCalls(existing.Messages)
|
||||
return existing, nil
|
||||
}
|
||||
canonical.PromptCacheKey = existing.PromptCacheKey
|
||||
|
||||
149
backend/internal/app/cloud_agent_pi_unanswered_calls.go
Normal file
149
backend/internal/app/cloud_agent_pi_unanswered_calls.go
Normal file
@@ -0,0 +1,149 @@
|
||||
package app
|
||||
|
||||
// 补齐没有结果的工具调用。
|
||||
//
|
||||
// 模型一步里发出多个工具调用、排在前面的那个进入用户审批时，运行时会中止本轮
|
||||
// （agent-runtime.mjs 的 session.abort()），排在后面的调用既没执行也没有 toolResult。
|
||||
// 用户批准后恢复运行，会话历史里就留着一个「有调用没结果」的条目：
|
||||
// validateRuntimeMessagesForModel 判它无效，回退到旧的规范历史，随后在
|
||||
// validateCanonicalAgentHistory 以「工具调用缺少结果」让整轮失败，恢复提示词也送不到模型。
|
||||
//
|
||||
// 这里在转换前给这些调用补一条明确的「未执行」结果：历史恢复合法，模型也知道
|
||||
// 那一步没做过，需要时会自己重新调用。只补缺失的，不改已有的结果。
|
||||
const cloudAgentPiUnansweredCallText = "该工具调用未被执行：本轮在它之前暂停（同一步里排在前面的操作进入了用户审批）或被中断。如仍需要它的结果，请重新调用。"
|
||||
|
||||
// repairRuntimeUnansweredCalls 返回补齐后的消息序列；没有缺失时原样返回同一个切片。
|
||||
// 调用 ID 解析不了的条目不在这里处理，留给后面的校验报出原始错误。
|
||||
func repairRuntimeUnansweredCalls(messages []map[string]any) []map[string]any {
|
||||
var repaired []map[string]any
|
||||
for index := 0; index < len(messages); index++ {
|
||||
message := messages[index]
|
||||
if repaired != nil {
|
||||
repaired = append(repaired, message)
|
||||
}
|
||||
if stringField(message, "role") != "assistant" {
|
||||
continue
|
||||
}
|
||||
calls := runtimeAssistantToolCalls(message)
|
||||
if len(calls) == 0 {
|
||||
continue
|
||||
}
|
||||
answered := make(map[string]struct{}, len(calls))
|
||||
next := index + 1
|
||||
for ; next < len(messages); next++ {
|
||||
role := stringField(messages[next], "role")
|
||||
if role != "toolResult" && role != "tool" {
|
||||
break
|
||||
}
|
||||
if callID, err := strictRuntimeStringAliases(messages[next], "Pi 工具结果 call id", "toolCallId", "tool_call_id", "call_id"); err == nil && callID != "" {
|
||||
answered[callID] = struct{}{}
|
||||
}
|
||||
}
|
||||
var missing []runtimeAssistantToolCall
|
||||
for _, call := range calls {
|
||||
if _, ok := answered[call.ID]; !ok {
|
||||
missing = append(missing, call)
|
||||
}
|
||||
}
|
||||
if len(missing) == 0 {
|
||||
continue
|
||||
}
|
||||
if repaired == nil {
|
||||
repaired = append(make([]map[string]any, 0, len(messages)+len(missing)), messages[:index+1]...)
|
||||
}
|
||||
repaired = append(repaired, messages[index+1:next]...)
|
||||
for _, call := range missing {
|
||||
repaired = append(repaired, map[string]any{
|
||||
"role": "toolResult",
|
||||
"toolCallId": call.ID,
|
||||
"toolName": call.Name,
|
||||
"content": []interface{}{map[string]interface{}{"type": "text", "text": cloudAgentPiUnansweredCallText}},
|
||||
"isError": true,
|
||||
})
|
||||
}
|
||||
index = next - 1
|
||||
}
|
||||
if repaired == nil {
|
||||
return messages
|
||||
}
|
||||
return repaired
|
||||
}
|
||||
|
||||
type runtimeAssistantToolCall struct {
|
||||
ID string
|
||||
Name string
|
||||
}
|
||||
|
||||
func runtimeAssistantToolCalls(message map[string]any) []runtimeAssistantToolCall {
|
||||
parts, _ := message["content"].([]interface{})
|
||||
var calls []runtimeAssistantToolCall
|
||||
for _, value := range parts {
|
||||
part, _ := value.(map[string]interface{})
|
||||
if stringField(part, "type") != "toolCall" {
|
||||
continue
|
||||
}
|
||||
callID, err := strictRuntimeStringAliases(part, "Pi 工具调用 ID", "id", "call_id", "toolCallId", "tool_call_id")
|
||||
if err != nil || callID == "" {
|
||||
continue
|
||||
}
|
||||
calls = append(calls, runtimeAssistantToolCall{ID: callID, Name: stringField(part, "name")})
|
||||
}
|
||||
return calls
|
||||
}
|
||||
|
||||
// repairCanonicalUnansweredCalls 是同一修复在服务端规范历史上的版本。
|
||||
// cloudAgentPiCanonicalForModelRequest 在服务端历史更长时会直接采用它而不是运行时
|
||||
// 消息（例如后端重启后恢复运行）；那份历史同样可能留着没有结果的调用，
|
||||
// 不补齐就会在 validateCanonicalAgentHistory 以「工具调用缺少结果」失败。
|
||||
// 没有缺失时原样返回同一个切片。
|
||||
func repairCanonicalUnansweredCalls(messages []map[string]any) []map[string]any {
|
||||
var repaired []map[string]any
|
||||
for index := 0; index < len(messages); index++ {
|
||||
message := messages[index]
|
||||
if repaired != nil {
|
||||
repaired = append(repaired, message)
|
||||
}
|
||||
if stringField(message, "role") != "assistant" {
|
||||
continue
|
||||
}
|
||||
calls := canonicalAgentToolCalls(message["tool_calls"])
|
||||
if len(calls) == 0 {
|
||||
continue
|
||||
}
|
||||
answered := make(map[string]struct{}, len(calls))
|
||||
next := index + 1
|
||||
for ; next < len(messages); next++ {
|
||||
if stringField(messages[next], "role") != "tool" {
|
||||
break
|
||||
}
|
||||
if callID, err := canonicalAgentStringAliases(messages[next], "画布 Agent 工具结果 call_id", "tool_call_id", "toolCallId", "call_id"); err == nil && callID != "" {
|
||||
answered[callID] = struct{}{}
|
||||
}
|
||||
}
|
||||
var missing []string
|
||||
for _, call := range calls {
|
||||
callID, err := canonicalAgentToolCallID(call, "画布 Agent 工具调用")
|
||||
if err != nil || callID == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := answered[callID]; !ok {
|
||||
missing = append(missing, callID)
|
||||
}
|
||||
}
|
||||
if len(missing) == 0 {
|
||||
continue
|
||||
}
|
||||
if repaired == nil {
|
||||
repaired = append(make([]map[string]any, 0, len(messages)+len(missing)), messages[:index+1]...)
|
||||
}
|
||||
repaired = append(repaired, messages[index+1:next]...)
|
||||
for _, callID := range missing {
|
||||
repaired = append(repaired, map[string]any{"role": "tool", "tool_call_id": callID, "content": cloudAgentPiUnansweredCallText})
|
||||
}
|
||||
index = next - 1
|
||||
}
|
||||
if repaired == nil {
|
||||
return messages
|
||||
}
|
||||
return repaired
|
||||
}
|
||||
131
backend/internal/app/cloud_agent_pi_unanswered_calls_test.go
Normal file
131
backend/internal/app/cloud_agent_pi_unanswered_calls_test.go
Normal file
@@ -0,0 +1,131 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func runtimeToolCallPart(id, name string) map[string]interface{} {
|
||||
return map[string]interface{}{"type": "toolCall", "id": id, "name": name, "arguments": map[string]interface{}{}}
|
||||
}
|
||||
|
||||
// 复现：一步里 plan_update（需审批）+ canvas_list_node_types，
|
||||
// 审批暂停后第二个调用没有结果，恢复运行时整轮以「工具调用缺少结果」失败。
|
||||
func TestCloudAgentPiCanonicalRepairsCallSkippedByApprovalPause(t *testing.T) {
|
||||
messages := []map[string]any{
|
||||
{"role": "user", "content": "做一段特惠视频"},
|
||||
{"role": "assistant", "content": []interface{}{
|
||||
map[string]interface{}{"type": "text", "text": "参数已确认"},
|
||||
runtimeToolCallPart("call_00_plan", "plan_update"),
|
||||
runtimeToolCallPart("call_01_types", "canvas_list_node_types"),
|
||||
}},
|
||||
{"role": "toolResult", "toolCallId": "call_00_plan", "toolName": "plan_update", "content": []interface{}{map[string]interface{}{"type": "text", "text": "操作正在等待用户审批。"}}},
|
||||
{"role": "user", "content": "用户已批准刚才等待审批的操作。"},
|
||||
}
|
||||
// 恢复前服务端保存的规范历史：停在审批那一步，比运行时消息短。
|
||||
existing := canonicalAgentRequest{
|
||||
SystemPrompt: "Agent system prompt",
|
||||
PromptCacheKey: "cache-key",
|
||||
Messages: []map[string]interface{}{
|
||||
{"role": "system", "content": "Agent system prompt"},
|
||||
{"role": "user", "content": "做一段特惠视频"},
|
||||
{"role": "assistant", "content": "参数已确认", "tool_calls": []interface{}{
|
||||
map[string]interface{}{"id": "call_00_plan", "function": map[string]interface{}{"name": "plan_update", "arguments": "{}"}},
|
||||
map[string]interface{}{"id": "call_01_types", "function": map[string]interface{}{"name": "canvas_list_node_types", "arguments": "{}"}},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
canonical, err := cloudAgentPiCanonicalForModelRequest(messages, existing)
|
||||
if err != nil {
|
||||
t.Fatalf("convert resumed transcript: %v", err)
|
||||
}
|
||||
if err := validateCanonicalAgentHistory(&canonical, false); err != nil {
|
||||
t.Fatalf("resumed history still invalid: %v", err)
|
||||
}
|
||||
// system + user + assistant + 两条工具结果 + 恢复提示词
|
||||
if len(canonical.Messages) != 6 {
|
||||
t.Fatalf("messages = %d, want 6: %#v", len(canonical.Messages), canonical.Messages)
|
||||
}
|
||||
skipped := canonical.Messages[4]
|
||||
if stringField(skipped, "role") != "tool" || stringField(skipped, "tool_call_id") != "call_01_types" || !strings.Contains(stringField(skipped, "content"), "未被执行") {
|
||||
t.Fatalf("skipped call was not answered: %#v", skipped)
|
||||
}
|
||||
last := canonical.Messages[5]
|
||||
if stringField(last, "role") != "user" || !strings.Contains(stringField(last, "content"), "已批准") {
|
||||
t.Fatalf("resume prompt did not reach the model: %#v", last)
|
||||
}
|
||||
if len(messages) != 4 {
|
||||
t.Fatalf("input transcript was mutated: %d messages", len(messages))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepairRuntimeUnansweredCallsLeavesCompleteTranscriptUntouched(t *testing.T) {
|
||||
messages := []map[string]any{
|
||||
{"role": "user", "content": "读取画布和当前选区"},
|
||||
{"role": "assistant", "content": []interface{}{
|
||||
runtimeToolCallPart("call-state", "canvas_get_state"),
|
||||
runtimeToolCallPart("call-selection", "canvas_get_selection"),
|
||||
}},
|
||||
{"role": "toolResult", "toolCallId": "call-selection", "content": "{}"},
|
||||
{"role": "toolResult", "toolCallId": "call-state", "content": "{}"},
|
||||
{"role": "assistant", "content": []interface{}{map[string]interface{}{"type": "text", "text": "读完了"}}},
|
||||
}
|
||||
repaired := repairRuntimeUnansweredCalls(messages)
|
||||
if len(repaired) != len(messages) || &repaired[0] != &messages[0] {
|
||||
t.Fatalf("complete transcript was rewritten: %#v", repaired)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepairRuntimeUnansweredCallsHandlesSeveralTurns(t *testing.T) {
|
||||
messages := []map[string]any{
|
||||
{"role": "user", "content": "开始"},
|
||||
{"role": "assistant", "content": []interface{}{runtimeToolCallPart("a1", "plan_update"), runtimeToolCallPart("a2", "canvas_get_state")}},
|
||||
{"role": "toolResult", "toolCallId": "a1", "content": "{}"},
|
||||
{"role": "user", "content": "继续"},
|
||||
{"role": "assistant", "content": []interface{}{runtimeToolCallPart("b1", "canvas_get_state")}},
|
||||
{"role": "toolResult", "toolCallId": "b1", "content": "{}"},
|
||||
{"role": "assistant", "content": []interface{}{runtimeToolCallPart("c1", "ask_user")}},
|
||||
}
|
||||
repaired := repairRuntimeUnansweredCalls(messages)
|
||||
if err := validateRuntimeMessagesForModel(repaired); err != nil {
|
||||
t.Fatalf("repaired transcript invalid: %v", err)
|
||||
}
|
||||
if len(repaired) != len(messages)+2 {
|
||||
t.Fatalf("repaired = %d messages, want %d", len(repaired), len(messages)+2)
|
||||
}
|
||||
if stringField(repaired[3], "toolCallId") != "a2" || stringField(repaired[len(repaired)-1], "toolCallId") != "c1" {
|
||||
t.Fatalf("synthetic results misplaced: %#v", repaired)
|
||||
}
|
||||
}
|
||||
|
||||
// 后端重启后恢复运行时，服务端规范历史比运行时消息长，会被直接采用；
|
||||
// 它里面没有结果的调用同样要补齐，否则恢复运行仍以「工具调用缺少结果」失败。
|
||||
func TestCloudAgentPiCanonicalRepairsServerHistoryWhenItIsAuthoritative(t *testing.T) {
|
||||
existing := canonicalAgentRequest{
|
||||
SystemPrompt: "Agent system prompt",
|
||||
Messages: []map[string]interface{}{
|
||||
{"role": "system", "content": "Agent system prompt"},
|
||||
{"role": "user", "content": "整理画布"},
|
||||
{"role": "assistant", "content": "", "tool_calls": []interface{}{
|
||||
map[string]interface{}{"id": "call-a", "function": map[string]interface{}{"name": "canvas_get_state", "arguments": "{}"}},
|
||||
map[string]interface{}{"id": "call-b", "function": map[string]interface{}{"name": "canvas_list_node_types", "arguments": "{}"}},
|
||||
}},
|
||||
{"role": "tool", "tool_call_id": "call-a", "content": "{}"},
|
||||
{"role": "user", "content": "继续"},
|
||||
},
|
||||
}
|
||||
// 运行时这边只剩一条恢复提示词，比服务端历史短。
|
||||
messages := []map[string]any{{"role": "user", "content": "继续"}}
|
||||
|
||||
canonical, err := cloudAgentPiCanonicalForModelRequest(messages, existing)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := validateCanonicalAgentHistory(&canonical, false); err != nil {
|
||||
t.Fatalf("authoritative server history still invalid: %v", err)
|
||||
}
|
||||
if len(canonical.Messages) != 6 || stringField(canonical.Messages[4], "tool_call_id") != "call-b" {
|
||||
t.Fatalf("missing result not inserted after existing results: %#v", canonical.Messages)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user