From 4b1ce20d2d5891649b35409200bb11f0de0517fd Mon Sep 17 00:00:00 2001 From: jiabx Date: Mon, 5 Oct 2026 13:56:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E5=AE=A1=E6=89=B9=E6=9A=82?= =?UTF-8?q?=E5=81=9C=20-=20=E6=9A=82=E5=81=9C=E5=90=8E=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E6=8E=A5=E5=8F=97=E6=96=B0=E7=9A=84=E6=A8=A1=E5=9E=8B=E6=AD=A5?= =?UTF-8?q?=E9=AA=A4=20(#675)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 运行时在工具进入审批时调用 session.abort()，但中止是异步生效的，SDK 可能抢在它生效前再开一轮并请求 /model。服务端照常入队：多计一次费，返回的新工具调用顶掉待审批的调用，保存时以「approval does not match current call」的检查点校验失败，整轮中断，用户还没来得及审批。 三处防护：运行时在暂停后不再发模型请求；服务端在等审批期间拒绝模型步骤（不入队、不改状态）；已因审批暂停的会话收尾报错不再判运行失败，由审批结论接管。 --- backend/agent-runtime/pi/agent-runtime.mjs | 8 ++ .../cloud_agent_pi_awaiting_approval_test.go | 105 ++++++++++++++++++ .../app/cloud_agent_pi_coordinator.go | 6 + backend/internal/app/cloud_agent_pi_model.go | 9 ++ 4 files changed, 128 insertions(+) create mode 100644 backend/internal/app/cloud_agent_pi_awaiting_approval_test.go diff --git a/backend/agent-runtime/pi/agent-runtime.mjs b/backend/agent-runtime/pi/agent-runtime.mjs index 7a98238e9..4d5215c45 100644 --- a/backend/agent-runtime/pi/agent-runtime.mjs +++ b/backend/agent-runtime/pi/agent-runtime.mjs @@ -138,6 +138,14 @@ async function run() { }; void (async () => { stream.push({ type: "start", partial }); + // 审批暂停时 session.abort() 是异步生效的，SDK 可能在它生效前就开下一轮。 + // 这一轮不能发给服务端：运行正在等审批，再跑一步会多计一次费， + // 新的工具调用还会和待审批的调用对不上，让整轮以检查点校验失败收场。 + if (pausedForApproval) { + stream.push({ type: "error", reason: "aborted", error: { ...partial, stopReason: "aborted", errorMessage: "paused for approval" } }); + stream.end(); + return; + } try { const purpose = activeCompaction?.phase === "summarizing" ? "compaction" : "conversation"; // Deliver lifecycle events/snapshots before scheduling the next request. diff --git a/backend/internal/app/cloud_agent_pi_awaiting_approval_test.go b/backend/internal/app/cloud_agent_pi_awaiting_approval_test.go new file mode 100644 index 000000000..4480dc1c4 --- /dev/null +++ b/backend/internal/app/cloud_agent_pi_awaiting_approval_test.go @@ -0,0 +1,105 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "gorm.io/gorm" + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +// approvalPauseRoot 建立 Pi 模式的运行（root 任务不占活动任务位，模型步走 /model 桥）。 +func approvalPauseRoot(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 +} + +// approvalPauseStage 把运行摆进等待审批状态：挂起的工具调用与审批一一对应 +// （检查点校验要求 Approval.Call 与 Calls[CallIndex] 一致）。 +func approvalPauseStage(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{} + } + approvalPauseSaveState(t, s, db, run, state) +} + +// approvalPauseSaveState 用生产同款 CAS 路径落库改过的运行态（直接 db.Save 会把 +// Transcript 关联行追加而不是替换，污染下一次 decode 的 canonical）。 +func approvalPauseSaveState(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) + } +} + +// approvalPauseModelPayload 组装运行时风格的 /model 请求（conversation）。 +func approvalPauseModelPayload(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"`), + } +} + +// 运行时在审批暂停后中止会话是异步的，可能抢在中止生效前再请求一步模型。 +// 这一步不能入队：否则多计一次费，返回的新工具调用还会顶掉待审批的调用， +// 让整轮以「approval does not match current call」的检查点校验失败收场。 +func TestCloudAgentPiModelStepRefusedWhileAwaitingApproval(t *testing.T) { + s, db, root := approvalPauseRoot(t) + run, state := agentInterjectionState(t, s, root.ID) + approvalPauseStage(t, s, db, run, &state) + before, stateBefore := agentInterjectionState(t, s, root.ID) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + messages := []map[string]any{{"role": "user", "content": []any{map[string]any{"type": "text", "text": "继续"}}}} + _, err := s.cloudAgentPiModel(ctx, "user", root.ID, approvalPauseModelPayload(t, messages)) + if !errors.Is(err, errCloudAgentAwaitingApproval) { + t.Fatalf("model step during approval wait: err=%v, want errCloudAgentAwaitingApproval", err) + } + + after, stateAfter := agentInterjectionState(t, s, root.ID) + if after.Status != "waiting_approval" || stateAfter.Approval == nil || stateAfter.Approval.ID != stateBefore.Approval.ID { + t.Fatalf("approval was disturbed: status=%s approval=%#v", after.Status, stateAfter.Approval) + } + if stateAfter.Step != stateBefore.Step || stateAfter.ActiveTaskID != "" || after.Revision != before.Revision { + t.Fatalf("a model step was scheduled: step %d→%d activeTask=%q revision %d→%d", stateBefore.Step, stateAfter.Step, stateAfter.ActiveTaskID, before.Revision, after.Revision) + } + var steps int64 + if err := db.Model(&model.Task{}).Where("agent_run_id = ? AND operation = ?", root.ID, cloudAgentStepOperation).Count(&steps).Error; err != nil { + t.Fatal(err) + } + if steps != 0 { + t.Fatalf("%d billable model step task(s) were created while awaiting approval", steps) + } +} diff --git a/backend/internal/app/cloud_agent_pi_coordinator.go b/backend/internal/app/cloud_agent_pi_coordinator.go index 49b86d9dd..0f397b7ad 100644 --- a/backend/internal/app/cloud_agent_pi_coordinator.go +++ b/backend/internal/app/cloud_agent_pi_coordinator.go @@ -356,6 +356,12 @@ func (s *Service) runCloudAgentPiSession(ctx context.Context, userID, runID stri }, }) if err != nil { + // 已经因审批暂停的会话，收尾阶段再报错不影响运行：状态停在等审批， + // 由审批结论接管。把它当失败处理会让用户还没来得及审批就整轮中断。 + if pausedForApproval.Load() && !errors.Is(err, context.Canceled) { + log.Printf("[Agent] session error after approval pause ignored run=%s: %v", runID, err) + 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 7bb108735..d98235a02 100644 --- a/backend/internal/app/cloud_agent_pi_model.go +++ b/backend/internal/app/cloud_agent_pi_model.go @@ -161,6 +161,9 @@ func (s *Service) broadcastCloudAgentPiContextPressure(userID, runID string, usa // cloudAgentModelStepRetries 是单步模型调用首次失败后的最大重试次数。 const cloudAgentModelStepRetries = 3 +// errCloudAgentAwaitingApproval 运行正在等用户审批，本步模型请求被拒绝（未入队、未计费）。 +var errCloudAgentAwaitingApproval = errors.New("run is waiting for approval") + // errCloudAgentTruncatedToolArguments 标记"模型返回的工具参数不是完整 JSON"： // 调用未执行，可以带纠偏上下文重做同一步。 var errCloudAgentTruncatedToolArguments = errors.New("truncated tool arguments") @@ -186,6 +189,12 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri if cloudAgentRunTerminal(run.Status) { return nil, false, fmt.Errorf("run already terminated") } + // 等审批期间不接受新的模型步骤。运行时在审批暂停后中止会话是异步的， + // 可能抢在中止生效前再请求一步；放它过去会多计一次费，返回的新工具调用 + // 还会顶掉待审批的调用。这里不入队、不改状态，审批结论照常接管运行。 + if run.Status == "waiting_approval" && state.Approval != nil { + return nil, false, errCloudAgentAwaitingApproval + } if cloudAgentStepBudgetExhausted(&state) { // 最终回答之后的插话续步会在这里撞上限；若确实停在最终回答边界， // 按完成收尾而不是把整轮判失败（见 settleCloudAgentPiFinalAnswer）。