From 5bf838ebbe6a2e261b4c26a7f4e35efd16f59d70 Mon Sep 17 00:00:00 2001 From: jiabx Date: Mon, 5 Oct 2026 13:39:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E6=8F=92=E8=AF=9D=20-=20?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B=E8=BF=90=E8=A1=8C=E6=97=B6=E8=B7=AF=E5=BE=84?= =?UTF-8?q?=E7=9A=84=E9=80=81=E8=BE=BE=E4=B8=8E=E9=80=80=E5=9B=9E=20(#673)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复现：Agent 运行中发送插话，独立 Agent 运行时（Pi）下既不送达模型也不退回用户，插话一直停在待送队列里。 原因与修法：插话送达点原本只有旧 Go 执行循环（advanceCloudAgent 开步前出队）；Pi 运行时接手执行后调度循环不再推进。现把 Pi 路径的送达搬到 /model 模型步：出队、规范历史追加与 delivered 事件同在入队的一次 CAS 写入里，响应携带 steeringMessages 交给运行时 session.steer。完成、失败、取消、审批拒绝与步数耗尽统一退回；最终回答之后的插话在预算富余时续跑一步。完成闸一次带走多条插话时，合并后的续跑提示词也记入指纹，避免被恢复闸当成遗留提示词清空。 --- .../app/cloud_agent_interjection_pi.go | 181 +++++++ .../app/cloud_agent_interjection_pi_test.go | 497 ++++++++++++++++++ .../app/cloud_agent_pi_coordinator.go | 34 +- backend/internal/app/cloud_agent_pi_model.go | 17 +- backend/internal/app/cloud_agent_runtime.go | 5 +- .../internal/app/cloud_agent_runtime_media.go | 2 + 6 files changed, 730 insertions(+), 6 deletions(-) create mode 100644 backend/internal/app/cloud_agent_interjection_pi.go create mode 100644 backend/internal/app/cloud_agent_interjection_pi_test.go diff --git a/backend/internal/app/cloud_agent_interjection_pi.go b/backend/internal/app/cloud_agent_interjection_pi.go new file mode 100644 index 000000000..76c9e1f18 --- /dev/null +++ b/backend/internal/app/cloud_agent_interjection_pi.go @@ -0,0 +1,181 @@ +// Pi 运行时路径的插话送达与退回。 +// +// 插话的入队、幂等与事件格式见 cloud_agent_interjection.go；送达点原本只有旧 +// Go 执行循环（advanceCloudAgent 开步前 drain）。独立 Agent 运行时（Pi）接手 +// 执行后调度循环不再推进，插话会一直停在待送队列里——本文件把 Pi 路径的送达 +// 搬到 /model 模型步桥上： +// +// - 出队发生在构建模型请求的 CAS 写入里（runCloudAgentModelStep）：插话追加 +// 进本次请求的 canonical，模型这一步就能看到它；出队、追加与 delivered +// 事件在同一次写入落库，重试不会重复送达。 +// - 响应携带 steeringMessages，运行时逐条 session.steer：实测运行时在当前轮 +// 工具执行完、下一次模型调用前把它作为 user 消息写进会话并落盘；即使模型 +// 本轮没有工具调用，挂着的 steer 也会让会话再走一步，因此「最终回答时的 +// 插话」由运行时自然续跑，无需服务端另开一步。 +// - 两份历史各有一份插话：下一次 /model 的 canonical 选取 +// （cloudAgentPiCanonicalForModelRequest）在长度相等时采用运行时投影， +// 插话不会重复；投影不带来源标记，转换后按已送达指纹补回。 + +package app + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +// cloudAgentInterjectionContent 插话进入对话的正文格式：前缀是给模型看的结构 +// 信号——按「用户中途改了要求」对待，而不是上一句的延续。与旧循环保持一致。 +func cloudAgentInterjectionContent(text string) string { + return "【用户插话】" + text +} + +func cloudAgentInterjectionDigest(content string) string { + sum := sha256.Sum256([]byte(content)) + return hex.EncodeToString(sum[:]) +} + +// cloudAgentPiDeliverInterjections 把待送插话注入本轮对话（Pi 路径的送达点）。 +// 必须在 runCloudAgentModelStep 的 CAS 写入里调用：出队、canonical 追加与 +// delivered 事件由同一次写入落库。返回交给运行时 steer 的正文列表。 +func cloudAgentPiDeliverInterjections(runID string, state *cloudAgentRuntime) []string { + if state == nil || len(state.PendingInterjections) == 0 { + return nil + } + texts := make([]string, 0, len(state.PendingInterjections)) + for _, item := range state.PendingInterjections { + content := cloudAgentInterjectionContent(item.Text) + state.Canonical.Messages = append(state.Canonical.Messages, map[string]any{ + "role": "user", + "content": content, + cloudAgentContextSourceKey: "user_interjection", + }) + state.event(runID, "user_interjection_delivered", map[string]any{"messageId": item.ID, "text": item.Text}) + cloudAgentRememberInterjectionID(state, item.ID) + // 指纹与插话 ID 共用同一个有界上限：只用于运行时投影后的标记补回， + // 不承担幂等职责（那由 InterjectionIDs 负责）。 + state.DeliveredInterjectionDigests = append(state.DeliveredInterjectionDigests, cloudAgentInterjectionDigest(content)) + if len(state.DeliveredInterjectionDigests) > cloudAgentInterjectionIDMemory { + state.DeliveredInterjectionDigests = state.DeliveredInterjectionDigests[len(state.DeliveredInterjectionDigests)-cloudAgentInterjectionIDMemory:] + } + texts = append(texts, content) + } + state.PendingInterjections = nil + return texts +} + +// cloudAgentRestoreInterjectionSources 按已送达指纹给 canonical 里的插话补回 +// 来源标记。运行时投影只携带正文，每次模型步转换都会洗掉标记；跨轮续跑的 +// 历史选取（cloudAgentContinuationHistory）靠标记识别插话，必须在每次转换后 +// 补回。幂等：已有来源标记的消息不动。 +func cloudAgentRestoreInterjectionSources(state *cloudAgentRuntime) { + if state == nil || len(state.DeliveredInterjectionDigests) == 0 { + return + } + for _, message := range state.Canonical.Messages { + if stringField(message, "role") != "user" || stringField(message, cloudAgentContextSourceKey) != "" { + continue + } + digest := cloudAgentInterjectionDigest(stringField(message, "content")) + for _, delivered := range state.DeliveredInterjectionDigests { + if digest == delivered { + message[cloudAgentContextSourceKey] = "user_interjection" + break + } + } + } +} + +// cloudAgentPiInterjectionResumePrompt 完成闸把待送插话合成一条续跑提示词。 +// 多条插话合并后正文与任何单条的指纹都对不上，恢复闸 +// （cloudAgentPiResumePromptDeliversInterjection）会把它当成遗留提示词清空， +// 已标记送达的插话就到不了模型；所以合并后的正文也记一份指纹。 +func cloudAgentPiInterjectionResumePrompt(runID string, state *cloudAgentRuntime) string { + texts := cloudAgentPiDeliverInterjections(runID, state) + prompt := strings.Join(texts, "\n") + if len(texts) > 1 { + state.DeliveredInterjectionDigests = append(state.DeliveredInterjectionDigests, cloudAgentInterjectionDigest(prompt)) + if len(state.DeliveredInterjectionDigests) > cloudAgentInterjectionIDMemory { + state.DeliveredInterjectionDigests = state.DeliveredInterjectionDigests[len(state.DeliveredInterjectionDigests)-cloudAgentInterjectionIDMemory:] + } + } + return prompt +} + +// cloudAgentPiResumePromptDeliversInterjection 判断恢复提示词是否为完成闸写下 +// 的插话续跑（见 completeCloudAgentPiRun）。这类提示词必须真正跑出去；崩溃 +// 恢复遇到的「快照已定格 + 上一次恢复遗留的提示词」才按原口径清空收尾。 +func cloudAgentPiResumePromptDeliversInterjection(state *cloudAgentRuntime) bool { + if state == nil || state.PiResumePrompt == "" || len(state.DeliveredInterjectionDigests) == 0 { + return false + } + digest := cloudAgentInterjectionDigest(state.PiResumePrompt) + for _, delivered := range state.DeliveredInterjectionDigests { + if digest == delivered { + return true + } + } + return false +} + +// cloudAgentPiRunAtFinalAnswer 运行是否停在「模型已给出最终回答」的边界： +// 没有挂起的工具批次、任务或审批，canonical 最后一条是不带工具调用的助手回复。 +func cloudAgentPiRunAtFinalAnswer(state *cloudAgentRuntime) bool { + if state == nil || state.ActiveTaskID != "" || state.MediaTaskID != "" || state.Approval != nil || state.ContextCompaction != nil { + return false + } + if state.CallIndex < len(state.Calls) { + return false + } + messages := state.Canonical.Messages + if len(messages) == 0 { + return false + } + last := messages[len(messages)-1] + if stringField(last, "role") != "assistant" { + return false + } + _, hasCalls := last["tool_calls"] + return !hasCalls +} + +// settleCloudAgentPiFinalAnswer 处理「最终回答之后插话 steer 续步、而步数预算 +// 已在最后一步用光」的收尾：回答本身已带着插话完成，按旧循环口径收尾为 +// completed 并退回剩余插话，而不是把整轮判成失败。普通步数耗尽（工具调用之后 +// 想再开一步）不满足最终回答边界，保持原有失败路径。收尾后运行时以模型步失败 +// 退出，failPiRunner 看到终态不会再写状态。 +func (s *Service) settleCloudAgentPiFinalAnswer(userID, runID string) { + for attempt := 0; attempt < cloudAgentInterjectionAttempts; attempt++ { + run, err := s.repo.CloudAgent(userID, runID) + if err != nil { + return + } + state, err := cloudAgentDecode(run) + if err != nil { + return + } + if cloudAgentRunTerminal(run.Status) || run.Status == "waiting_approval" || run.Status == "waiting_user" { + return + } + if !cloudAgentPiRunAtFinalAnswer(&state) { + return + } + err = s.repo.MutateCloudAgent(userID, runID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { + if cloudAgentRunTerminal(current.Status) || current.Status == "waiting_approval" || current.Status == "waiting_user" { + return nil + } + cloudAgentDropInterjections(runID, "本轮已达到模型调用上限", &state) + current.Status = "completed" + state.event(runID, "run_completed", map[string]any{"text": "Agent 已完成", "timestamp": time.Now().Unix()}) + return cloudAgentSave(current, &state) + }) + if !errors.Is(err, repository.ErrCreationConflict) { + return + } + } +} diff --git a/backend/internal/app/cloud_agent_interjection_pi_test.go b/backend/internal/app/cloud_agent_interjection_pi_test.go new file mode 100644 index 000000000..2bc64a49d --- /dev/null +++ b/backend/internal/app/cloud_agent_interjection_pi_test.go @@ -0,0 +1,497 @@ +package app + +// Pi 运行时路径的插话送达与退回测试。见 cloud_agent_interjection_pi.go。 +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "gorm.io/gorm" + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +// interjectionPiRoot 建立 Pi 模式的运行（root 任务不占活动任务位，插话送达发生在 /model 桥）。 +func interjectionPiRoot(t *testing.T) (*Service, *gorm.DB, *CloudAgentRun) { + t.Helper() + s, db, _, _ := creationTestService(t) + s.legacyCloudAgentRootTask = false + if err := db.Create(&model.CanvasProject{ID: "agent-canvas", UserID: "user", PayloadJSON: `{"nodes":[]}`}).Error; err != nil { + t.Fatal(err) + } + root, err := s.CreateCloudAgentRun("user", agentTestRequest(), "") + if err != nil { + t.Fatal(err) + } + return s, db, root +} + +// interjectionApprovalRun 把运行摆进等待审批状态：挂起的工具调用与审批一一对应 +// （检查点校验要求 Approval.Call 与 Calls[CallIndex] 一致）。 +func interjectionApprovalRun(t *testing.T, s *Service, db *gorm.DB, run *model.CloudAgentExecution, state *cloudAgentRuntime) { + t.Helper() + call := cloudAgentCall{ID: "call-1"} + call.Function.Name = "generate_media" + call.Function.Arguments = "{}" + run.Status = "waiting_approval" + state.Calls = []cloudAgentCall{call} + state.CallIndex = 0 + state.Approval = &cloudAgentApproval{ID: "ap-1", Call: call} + if state.Decisions == nil { + state.Decisions = map[string]string{} + } + interjectionSaveState(t, s, db, run, state) +} + +// interjectionModelPayload 组装运行时风格的 /model 请求（conversation）。 +func interjectionModelPayload(t *testing.T, messages []map[string]any) map[string]json.RawMessage { + t.Helper() + encoded, err := json.Marshal(messages) + if err != nil { + t.Fatal(err) + } + return map[string]json.RawMessage{ + "modelId": json.RawMessage(`"text-test"`), + "purpose": json.RawMessage(`"conversation"`), + "messages": encoded, + "thinkingLevel": json.RawMessage(`"off"`), + } +} + +// interjectionModelStep 桥接一次模型步：等待步任务入队（跳过建轮时的活动任务）、 +// 把任务改成给定结果、等待桥返回。 +func interjectionModelStep(t *testing.T, s *Service, db *gorm.DB, userID, runID string, messages []map[string]any, resultJSON string) map[string]any { + t.Helper() + stored, err := s.repo.CloudAgent(userID, runID) + if err != nil { + t.Fatal(err) + } + initialState, err := cloudAgentDecode(stored) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + type outcome struct { + result any + err error + } + done := make(chan outcome, 1) + go func() { + result, stepErr := s.cloudAgentPiModel(ctx, userID, runID, interjectionModelPayload(t, messages)) + done <- outcome{result: result, err: stepErr} + }() + var taskID string + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + stored, readErr := s.repo.CloudAgent(userID, runID) + if readErr != nil { + t.Fatal(readErr) + } + state, decodeErr := cloudAgentDecode(stored) + if decodeErr != nil { + t.Fatal(decodeErr) + } + if state.ActiveTaskID != "" && state.ActiveTaskID != initialState.ActiveTaskID { + taskID = state.ActiveTaskID + break + } + select { + case got := <-done: + t.Fatalf("model bridge returned before scheduling: result=%v err=%v", got.result, got.err) + case <-time.After(25 * time.Millisecond): + } + } + if taskID == "" { + t.Fatal("model step was never scheduled") + } + if err := db.Model(&model.Task{}).Where("id = ?", taskID).Updates(map[string]any{ + "status": model.TaskStatusSucceeded, "result_json": resultJSON, + }).Error; err != nil { + t.Fatal(err) + } + select { + case got := <-done: + if got.err != nil { + t.Fatalf("model bridge failed: %v", got.err) + } + result, _ := got.result.(map[string]any) + if result == nil { + t.Fatalf("unexpected bridge result: %#v", got.result) + } + return result + case <-ctx.Done(): + t.Fatal("model bridge did not return") + } + return nil +} + +// interjectionSteerTexts 取响应里的 steer 正文（Go 内直接组装，是 []string）。 +func interjectionSteerTexts(result map[string]any) []string { + texts, _ := result["steeringMessages"].([]string) + return texts +} + +// interjectionSaveState 用生产同款 CAS 路径落库改过的运行态（直接 db.Save 会把 +// Transcript 关联行追加而不是替换，污染下一次 decode 的 canonical）。 +func interjectionSaveState(t *testing.T, s *Service, db *gorm.DB, run *model.CloudAgentExecution, state *cloudAgentRuntime) { + t.Helper() + if err := s.repo.MutateCloudAgent(run.UserID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { + current.Status = run.Status + return cloudAgentSave(current, state) + }); err != nil { + t.Fatal(err) + } +} + +// 运行中插话在下一次 /model 出队：进入本次请求的 canonical（带来源标记）、 +// 发 delivered 事件、响应携带 steeringMessages 交给运行时 session.steer。 +func TestCloudAgentPiInterjectionReachesModelRequest(t *testing.T) { + s, db, root := interjectionPiRoot(t) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-1", "改成 16:9"); err != nil { + t.Fatal(err) + } + messages := []map[string]any{{"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}} + result := interjectionModelStep(t, s, db, "user", root.ID, messages, `{"text":"好的","toolCalls":[]}`) + + steered := interjectionSteerTexts(result) + if len(steered) != 1 || steered[0] != "【用户插话】改成 16:9" { + t.Fatalf("响应没有携带 steer 正文: %#v", result["steeringMessages"]) + } + _, state := agentInterjectionState(t, s, root.ID) + if len(state.PendingInterjections) != 0 { + t.Fatalf("插话已送达却没有出队: %+v", state.PendingInterjections) + } + if !agentHasEvent(state, "user_interjection_delivered") { + t.Fatal("送达时没有发 user_interjection_delivered 事件") + } + if len(state.InterjectionIDs) == 0 || state.InterjectionIDs[0] != "msg-1" { + t.Fatalf("送达没有记住插话 id: %#v", state.InterjectionIDs) + } + // 最关键的一条：插话必须出现在真正发给模型的输入里，并带来源标记。 + var task model.Task + last := state.TaskIDs[len(state.TaskIDs)-1] + if err := db.First(&task, "id = ?", last).Error; err != nil { + t.Fatal(err) + } + var input struct { + AgentRequests struct { + Canonical struct { + Messages []map[string]any `json:"messages"` + } `json:"canonical"` + } `json:"agentRequests"` + } + if err := json.Unmarshal([]byte(task.InputJSON), &input); err != nil { + t.Fatal(err) + } + delivered := false + for _, message := range input.AgentRequests.Canonical.Messages { + content, _ := message["content"].(string) + if message["role"] == "user" && strings.Contains(content, "改成 16:9") { + delivered = true + if message[cloudAgentContextSourceKey] != "user_interjection" { + t.Fatalf("送达进 canonical 的插话缺少来源标记: %#v", message) + } + } + } + if !delivered { + t.Fatalf("插话没有进入模型请求的输入: %s", task.InputJSON) + } +} + +// 等审批期间入队的插话，在恢复后的第一次 /model 送达（模型这一步就能看到， +// 不用等 steer 在下一轮生效）。 +func TestCloudAgentPiInterjectionQueuedDuringApprovalDeliversOnResume(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + interjectionApprovalRun(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-approval", "批准后按新要求来"); err != nil { + t.Fatal(err) + } + // 审批批准、被暂停的工具执行完毕后的状态：审批已清、回到 running。 + // （真实链路里 advanceCloudAgentTool 执行工具时清理审批，之后才恢复会话。） + run, state = agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Approval = nil + state.Calls = nil + state.CallIndex = 0 + interjectionSaveState(t, s, db, run, &state) + messages := []map[string]any{{"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}} + interjectionModelStep(t, s, db, "user", root.ID, messages, `{"text":"好的","toolCalls":[]}`) + _, state = agentInterjectionState(t, s, root.ID) + if len(state.PendingInterjections) != 0 { + t.Fatal("恢复后的第一步没有送达等审批时的插话") + } + if !agentHasEvent(state, "user_interjection_delivered") { + t.Fatal("恢复送达没有发 user_interjection_delivered 事件") + } +} + +// 已送达插话经过运行时投影回写后来源标记被洗掉：下一步模型步按指纹补回， +// 且插话在两份历史之间不重复（长度相等时采用运行时投影，插话只出现一次）。 +func TestCloudAgentPiInterjectionMarkerSurvivesRuntimeProjection(t *testing.T) { + s, db, root := interjectionPiRoot(t) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-1", "改成 16:9"); err != nil { + t.Fatal(err) + } + first := []map[string]any{{"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}} + interjectionModelStep(t, s, db, "user", root.ID, first, `{"text":"好的","toolCalls":[]}`) + _, state := agentInterjectionState(t, s, root.ID) + if len(state.Canonical.Messages) == 0 || stringField(state.Canonical.Messages[len(state.Canonical.Messages)-1], "role") != "assistant" { + t.Fatalf("这一步的助手回复没有落检查点: %+v", state.Canonical.Messages) + } + // 下一拍的运行时会话：插话（steer 落盘）排在助手回复之后，且不带标记。 + second := []map[string]any{ + {"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}, + {"role": "assistant", "content": []any{map[string]any{"type": "text", "text": "好的"}}}, + {"role": "user", "content": []any{map[string]any{"type": "text", "text": "【用户插话】改成 16:9"}}}, + } + result := interjectionModelStep(t, s, db, "user", root.ID, second, `{"text":"收到","toolCalls":[]}`) + if len(interjectionSteerTexts(result)) != 0 { + t.Fatalf("队列已空却再次返回 steer 正文: %#v", result["steeringMessages"]) + } + _, state = agentInterjectionState(t, s, root.ID) + var task model.Task + last := state.TaskIDs[len(state.TaskIDs)-1] + if err := db.First(&task, "id = ?", last).Error; err != nil { + t.Fatal(err) + } + if got := strings.Count(task.InputJSON, "改成 16:9"); got != 1 { + t.Fatalf("插话在两份历史之间重复或丢失: count=%d input=%s", got, task.InputJSON) + } + marked := false + for _, message := range state.Canonical.Messages { + content, _ := message["content"].(string) + if strings.Contains(content, "改成 16:9") && message[cloudAgentContextSourceKey] == "user_interjection" { + marked = true + } + } + if !marked { + t.Fatalf("运行时投影回写后插话标记没有被补回: %+v", state.Canonical.Messages) + } +} + +// 运行自然结束但队列里还有插话：预算够就作为续跑提示词再走一步（不标记完成）。 +func TestCloudAgentPiInterjectionResumesAfterFinalAnswer(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Request.Budget.MaxSteps = 3 + state.Step = 1 + state.PiAssistantResponses = 1 + state.Canonical.Messages = []map[string]any{ + {"role": "user", "content": "分析剧情"}, + {"role": "assistant", "content": "好的"}, + } + interjectionSaveState(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-late", "补一句：加字幕"); err != nil { + t.Fatal(err) + } + if err := s.completeCloudAgentPiRun("user", root.ID); err != nil { + t.Fatal(err) + } + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "running" { + t.Fatalf("还有未送达插话且预算富余时不应完成，实际 status=%s", run.Status) + } + if len(state.PendingInterjections) != 0 { + t.Fatal("完成闸没有出队插话") + } + if state.PiResumePrompt != "【用户插话】补一句：加字幕" { + t.Fatalf("插话没有被写进续跑提示词: %q", state.PiResumePrompt) + } + if !agentHasEvent(state, "user_interjection_delivered") { + t.Fatal("完成闸送达没有发 user_interjection_delivered 事件") + } + // 续跑提示词必须被识别为插话投递：恢复会话遇到定格快照时照常跑出去。 + if !cloudAgentPiResumePromptDeliversInterjection(&state) { + t.Fatal("恢复闸没有识别插话续跑提示词") + } +} + +// 步数已尽时插话退回、运行照常完成。 +func TestCloudAgentPiInterjectionDroppedWhenStepBudgetExhausted(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Request.Budget.MaxSteps = 1 + state.Step = 1 + state.PiAssistantResponses = 1 + state.Canonical.Messages = []map[string]any{ + {"role": "user", "content": "分析剧情"}, + {"role": "assistant", "content": "好的"}, + } + interjectionSaveState(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-late", "补一句：加字幕"); err != nil { + t.Fatal(err) + } + if err := s.completeCloudAgentPiRun("user", root.ID); err != nil { + t.Fatal(err) + } + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "completed" { + t.Fatalf("步数已尽时应完成收尾，实际 status=%s", run.Status) + } + if len(state.PendingInterjections) != 0 { + t.Fatal("步数已尽时插话还挂在队列里") + } + if !agentHasEvent(state, "user_interjection_dropped") { + t.Fatal("步数已尽时没有发 user_interjection_dropped 事件") + } +} + +// 插话 steer 续步撞上步数上限：模型已带着插话给出最终回答，收尾为 completed +// 并退回剩余插话，而不是把整轮判失败；停在工具结果上的普通步数耗尽不受影响。 +func TestCloudAgentPiSteeredContinuationCompletesAtFinalAnswer(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Request.Budget.MaxSteps = 1 + state.Step = 1 + state.PiAssistantResponses = 1 + state.Canonical.Messages = []map[string]any{ + {"role": "user", "content": "分析剧情"}, + {"role": "assistant", "content": "好的"}, + } + interjectionSaveState(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-late", "补一句"); err != nil { + t.Fatal(err) + } + s.settleCloudAgentPiFinalAnswer("user", root.ID) + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "completed" { + t.Fatalf("最终回答边界上应完成收尾，实际 status=%s", run.Status) + } + if len(state.PendingInterjections) != 0 || !agentHasEvent(state, "user_interjection_dropped") { + t.Fatal("收尾时没有退回插话") + } + + // 对照：canonical 停在工具结果上（工具批次之后想再开一步）不是最终回答边界， + // 保持原有失败路径。 + run, state = agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Canonical.Messages = []map[string]any{ + {"role": "user", "content": "分析剧情"}, + {"role": "assistant", "content": "画一张", "tool_calls": []any{map[string]any{"id": "call-1"}}}, + {"role": "tool", "tool_call_id": "call-1", "content": "{}"}, + } + interjectionSaveState(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-late-2", "补一句"); err != nil { + t.Fatal(err) + } + s.settleCloudAgentPiFinalAnswer("user", root.ID) + run, _ = agentInterjectionState(t, s, root.ID) + if run.Status != "running" { + t.Fatalf("工具批次后的步数耗尽不属于最终回答边界，实际 status=%s", run.Status) + } +} + +// 运行失败、被取消、审批被拒时，队列里的插话全部退回。 +func TestCloudAgentPiInterjectionsDroppedOnTerminalPaths(t *testing.T) { + t.Run("failed", func(t *testing.T) { + s, _, root := interjectionPiRoot(t) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-f", "还在吗"); err != nil { + t.Fatal(err) + } + s.failPiRunner(root.ID, "user", errors.New("模型调用失败")) + run, state := agentInterjectionState(t, s, root.ID) + if run.Status != "failed" || len(state.PendingInterjections) != 0 || !agentHasEvent(state, "user_interjection_dropped") { + t.Fatalf("失败收尾没有退回插话: status=%s pending=%+v", run.Status, state.PendingInterjections) + } + }) + t.Run("cancelled", func(t *testing.T) { + s, _, root := interjectionPiRoot(t) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-c", "还在吗"); err != nil { + t.Fatal(err) + } + if err := s.CancelCloudAgent(context.Background(), "user", root.ID); err != nil { + t.Fatal(err) + } + run, state := agentInterjectionState(t, s, root.ID) + if run.Status != "cancelled" || len(state.PendingInterjections) != 0 || !agentHasEvent(state, "user_interjection_dropped") { + t.Fatalf("取消没有退回插话: status=%s pending=%+v", run.Status, state.PendingInterjections) + } + }) + t.Run("rejected", func(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + interjectionApprovalRun(t, s, db, run, &state) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-r", "还在吗"); err != nil { + t.Fatal(err) + } + if err := s.DecideCloudAgentApproval("user", root.ID, "ap-1", "reject", "不要生成"); err != nil { + t.Fatal(err) + } + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "rejected" || len(state.PendingInterjections) != 0 || !agentHasEvent(state, "user_interjection_dropped") { + t.Fatalf("拒绝审批没有退回插话: status=%s pending=%+v", run.Status, state.PendingInterjections) + } + }) +} + +// 同一 messageId 重复提交不会送达两次：已送达的 id 被记住，重发直接短路。 +func TestCloudAgentPiInterjectionRedeliveryIsIdempotent(t *testing.T) { + s, db, root := interjectionPiRoot(t) + if _, err := s.InterjectCloudAgent("user", root.ID, "msg-1", "改成 16:9"); err != nil { + t.Fatal(err) + } + messages := []map[string]any{{"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}} + interjectionModelStep(t, s, db, "user", root.ID, messages, `{"text":"好的","toolCalls":[]}`) + count, err := s.InterjectCloudAgent("user", root.ID, "msg-1", "改成 16:9") + if err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("已送达的插话被重新入队: pending=%d", count) + } + _, state := agentInterjectionState(t, s, root.ID) + if len(state.PendingInterjections) != 0 { + t.Fatal("重复提交把插话又放回了队列") + } + delivered := 0 + for _, event := range state.Events { + if event.Type == "user_interjection_delivered" { + delivered++ + } + } + if delivered != 1 { + t.Fatalf("delivered 事件出现 %d 次，应当只有 1 次", delivered) + } +} + +// 完成闸一次带走多条插话时，合并后的续跑提示词也必须被恢复闸识别； +// 否则它会被当成遗留提示词清空，已标记送达的插话到不了模型。 +func TestCloudAgentPiInterjectionResumePromptRecognisedForSeveralInterjections(t *testing.T) { + s, db, root := interjectionPiRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + run.Status = "running" + state.Request.Budget.MaxSteps = 3 + state.Step = 1 + state.PiAssistantResponses = 1 + state.Canonical.Messages = []map[string]any{ + {"role": "user", "content": "分析剧情"}, + {"role": "assistant", "content": "好的"}, + } + interjectionSaveState(t, s, db, run, &state) + for id, text := range map[string]string{"msg-late-1": "加字幕", "msg-late-2": "再配个背景音乐"} { + if _, err := s.InterjectCloudAgent("user", root.ID, id, text); err != nil { + t.Fatal(err) + } + } + if err := s.completeCloudAgentPiRun("user", root.ID); err != nil { + t.Fatal(err) + } + run, state = agentInterjectionState(t, s, root.ID) + if run.Status != "running" || len(state.PendingInterjections) != 0 { + t.Fatalf("status=%s pending=%d", run.Status, len(state.PendingInterjections)) + } + if !strings.Contains(state.PiResumePrompt, "加字幕") || !strings.Contains(state.PiResumePrompt, "再配个背景音乐") { + t.Fatalf("续跑提示词缺少插话: %q", state.PiResumePrompt) + } + if !cloudAgentPiResumePromptDeliversInterjection(&state) { + t.Fatal("恢复闸没有识别多条插话合并后的续跑提示词") + } +} diff --git a/backend/internal/app/cloud_agent_pi_coordinator.go b/backend/internal/app/cloud_agent_pi_coordinator.go index 045db6162..49b86d9dd 100644 --- a/backend/internal/app/cloud_agent_pi_coordinator.go +++ b/backend/internal/app/cloud_agent_pi_coordinator.go @@ -195,6 +195,7 @@ func (s *Service) failPiRunner(runID, userID string, cause error) { state, decodeErr := cloudAgentDecode(current) if decodeErr == nil { state.LastError = cause.Error() + cloudAgentDropInterjections(runID, "本轮已结束："+truncateRunes(cause.Error(), 120), &state) state.event(runID, "run_failed", map[string]any{"error": cause.Error()}) return cloudAgentSave(current, &state) } @@ -302,13 +303,17 @@ func (s *Service) runCloudAgentPiSession(ctx context.Context, userID, runID stri // 崩溃恢复：最后答案已写进会话快照、运行却没来得及标记完成。这一回合已经结束， // 直接收尾；再启动运行时会对同一提示词再调用一次模型、多扣一次费。 + // 例外：完成闸写下的插话续跑提示词（completeCloudAgentPiRun）晚于快照里的 + // 最终回答，必须真正跑出去——清空收尾会让已标记送达的插话永远到不了模型。 if ownSession && cloudAgentPiTurnSettled(sessionJSONL, &runtimeState) { - if runtimeState.PiResumePrompt != "" { - if err := s.saveCloudAgentPiResumePrompt(userID, runID, ""); err != nil { - return err + if !cloudAgentPiResumePromptDeliversInterjection(&runtimeState) { + if runtimeState.PiResumePrompt != "" { + if err := s.saveCloudAgentPiResumePrompt(userID, runID, ""); err != nil { + return err + } } + return s.completeCloudAgentPiRun(userID, runID) } - return s.completeCloudAgentPiRun(userID, runID) } // 5. 构建完整的 Pi 请求（核心重构） @@ -443,11 +448,28 @@ func (s *Service) completeCloudAgentPiRun(userID, runID string) error { } // 原子更新状态 + resumed := false err = s.repo.MutateCloudAgent(userID, runID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { if cloudAgentRunTerminal(current.Status) || current.Status == "waiting_approval" || current.Status == "waiting_user" { return nil } + // 运行自然结束但队列里还有插话：它们没赶上最后一次 /model 出队，模型是 + // 在看不到它们的情况下给出的回答。预算还开得起下一步时，把插话作为续跑 + // 提示词再走一步（运行时按恢复提示词拉起新会话）；开不起时如实退回。 + // 续跑消耗的仍是本轮剩余预算，不是替用户自动开新一轮。 + if len(state.PendingInterjections) > 0 { + if limits, limitsErr := s.cloudAgentStepLimits(); limitsErr == nil { + state.StepLimits = limits + } + if !cloudAgentStepBudgetExhausted(&state) { + state.PiResumePrompt = cloudAgentPiInterjectionResumePrompt(runID, &state) + resumed = true + return cloudAgentSave(current, &state) + } + cloudAgentDropInterjections(runID, "本轮已达到模型调用上限", &state) + } + current.Status = "completed" state.event(runID, "run_completed", map[string]any{ "text": "Agent 已完成", @@ -458,6 +480,10 @@ func (s *Service) completeCloudAgentPiRun(userID, runID string) error { }) if !errors.Is(err, repository.ErrCreationConflict) { + if resumed && err == nil { + s.resumeCloudAgentPi(userID, runID) + return nil + } return err } } diff --git a/backend/internal/app/cloud_agent_pi_model.go b/backend/internal/app/cloud_agent_pi_model.go index 177ba8035..7bb108735 100644 --- a/backend/internal/app/cloud_agent_pi_model.go +++ b/backend/internal/app/cloud_agent_pi_model.go @@ -172,6 +172,8 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri // 这些事件也会推进运行 revision。调度模型步骤是 CAS 写入：冲突时重新读取 // 最新状态再调度，而不是把整轮判失败。冲突时事务整体回滚，不会产生任务或扣费。 var state cloudAgentRuntime + // 本步送达的插话正文，随响应交给运行时 session.steer（见 cloud_agent_interjection_pi.go）。 + var steeringMessages []string for attempt := 0; attempt < 8; attempt++ { run, err := s.repo.CloudAgent(userID, runID) if err != nil { @@ -185,6 +187,9 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri return nil, false, fmt.Errorf("run already terminated") } if cloudAgentStepBudgetExhausted(&state) { + // 最终回答之后的插话续步会在这里撞上限；若确实停在最终回答边界， + // 按完成收尾而不是把整轮判失败（见 settleCloudAgentPiFinalAnswer）。 + s.settleCloudAgentPiFinalAnswer(userID, runID) return nil, false, fmt.Errorf("step budget exhausted") } if state.StepLimits, err = s.cloudAgentStepLimits(); err != nil { @@ -222,8 +227,14 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri if len(canonical.Messages) == 0 { return nil, false, fmt.Errorf("no canonical messages") } + // 运行时投影回写会洗掉插话来源标记，先补回再送达新插话；送达必须发生在 + // 入队这次 CAS 写入里（出队、canonical 追加与 delivered 事件同批落库）， + // 冲突重试时从最新状态重新出队，不会重复送达。压缩摘要请求不参与。 if !compactionSummary { state.Canonical = canonical + cloudAgentRestoreInterjectionSources(&state) + steeringMessages = cloudAgentPiDeliverInterjections(runID, &state) + canonical = state.Canonical } if len(correction) > 0 { // 纠偏上下文只用于这一次请求，不写回运行状态，避免下一步重复出现。 @@ -328,7 +339,11 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri } else if err := s.finishCloudAgentPiModelStep(userID, runID, task.ID, finishText, finishReasoning, result.ToolCalls); err != nil { return nil, false, err } - return map[string]any{"text": result.Text, "reasoning": result.Reasoning, "toolCalls": runtimeToolCalls(result.ToolCalls)}, false, nil + delivered := map[string]any{"text": result.Text, "reasoning": result.Reasoning, "toolCalls": runtimeToolCalls(result.ToolCalls)} + if len(steeringMessages) > 0 { + delivered["steeringMessages"] = steeringMessages + } + return delivered, false, nil } // adoptCloudAgentPiModelStep 判断挂着的活动任务能否作为本步结果直接接手。 diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index b23bde8b7..dfe3b52ef 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -162,7 +162,10 @@ type cloudAgentRuntime struct { PiResumePrompt string `json:"piResumePrompt,omitempty"` TransientReferences map[string]cloudAgentTransientReference `json:"transientReferences,omitempty"` InterjectionIDs []string `json:"interjectionIds,omitempty"` - Events []CloudAgentEvent `json:"events"` + // 已送达插话正文的 SHA-256 指纹（有界）。运行时投影回写 canonical 时不带 + // 来源标记，模型步按指纹补回 user_interjection，见 cloud_agent_interjection_pi.go。 + DeliveredInterjectionDigests []string `json:"deliveredInterjectionDigests,omitempty"` + Events []CloudAgentEvent `json:"events"` // PiAssistantResponses counts successful assistant message_end events from // the Pi runtime. Completion must not be inferred from a clean Node exit: // a provider/session error can otherwise be reported as a successful run. diff --git a/backend/internal/app/cloud_agent_runtime_media.go b/backend/internal/app/cloud_agent_runtime_media.go index 91faca4ab..f6f63d564 100644 --- a/backend/internal/app/cloud_agent_runtime_media.go +++ b/backend/internal/app/cloud_agent_runtime_media.go @@ -462,6 +462,7 @@ func (s *Service) DecideCloudAgentApproval(userID, id, approvalID, decision, rea current.Status = "rejected" current.FailureMessage = "" state.Approval = nil + cloudAgentDropInterjections(id, "本轮已结束：已拒绝本次生成", &state) state.event(id, "approval_decided", map[string]any{ "approvalId": approvalID, "decision": decision, @@ -641,6 +642,7 @@ func (s *Service) CancelCloudAgent(ctx context.Context, userID, id string) error current.CanvasID, current.ActiveTaskID, current.MediaTaskID = state.Request.CanvasID, state.ActiveTaskID, state.MediaTaskID if firstCancellation { state.event(id, "run_cancelled", map[string]any{"source": "user_request", "activeTaskId": state.ActiveTaskID, "mediaTaskId": state.MediaTaskID, "text": "用户取消接口已接收请求，正在取消关联任务"}) + cloudAgentDropInterjections(id, "本轮已结束：用户已取消本轮", &state) if saveErr := cloudAgentSave(current, &state); saveErr != nil { // Cancellation must still work if the transcript is oversized. log.Printf("[cloud-agent] cancellation event unavailable for run %s: checkpoint rejected", id)