fix(agent): 插话 - 独立运行时路径的送达与退回 (#673)

复现：Agent 运行中发送插话，独立 Agent 运行时（Pi）下既不送达模型也不退回用户，插话一直停在待送队列里。

原因与修法：插话送达点原本只有旧 Go 执行循环（advanceCloudAgent 开步前出队）；Pi 运行时接手执行后调度循环不再推进。现把 Pi 路径的送达搬到 /model 模型步：出队、规范历史追加与 delivered 事件同在入队的一次 CAS 写入里，响应携带 steeringMessages 交给运行时 session.steer。完成、失败、取消、审批拒绝与步数耗尽统一退回；最终回答之后的插话在预算富余时续跑一步。完成闸一次带走多条插话时，合并后的续跑提示词也记入指纹，避免被恢复闸当成遗留提示词清空。

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
+		}
+	}
+}

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("恢复闸没有识别多条插话合并后的续跑提示词")
+	}
+}

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
 		}
 	}

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 判断挂着的活动任务能否作为本步结果直接接手。

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.

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)