diff --git a/internal/runmgr/material_test.go b/internal/runmgr/material_test.go index f014f26..416eddd 100644 --- a/internal/runmgr/material_test.go +++ b/internal/runmgr/material_test.go @@ -3,6 +3,7 @@ package runmgr import ( "fmt" "strings" + "sync/atomic" "testing" "agent-system-backend/internal/runtime" @@ -12,14 +13,14 @@ import ( // holdRunner 阻塞到 release 才完成——材料整理去重需要子任务仍在运行的窗口。 type holdRunner struct { release chan struct{} - id string + id atomic.Value } func (h *holdRunner) Name() string { return "hold" } func (h *holdRunner) Execute(rc runtime.RunCtx) error { if src, ok := rc.(interface{ SubagentID() string }); ok { - h.id = src.SubagentID() + h.id.Store(src.SubagentID()) } <-h.release return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "整理完成：三份材料的要点与冲突已汇总。"}}) @@ -117,8 +118,8 @@ func TestMaterialIntakeAttachmentOnlyLifecycle(t *testing.T) { if materialKVOf(cards[0]) == "" { t.Fatal("material key KV lost on finish backfill") } - if hold.id != materialSubagentID { - t.Fatalf("material subagent id = %q, want %q", hold.id, materialSubagentID) + if got := hold.id.Load(); got != materialSubagentID { + t.Fatalf("material subagent id = %q, want %q", got, materialSubagentID) } } @@ -284,6 +285,7 @@ func TestMaterialQueuedWhenFull(t *testing.T) { for _, h := range fills[1:] { close(h.release) } + waitBackgroundIdle(t, c) } func cardBodyText(card core.ActivityCard) string { diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index a2b0a30..3b735f2 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -2235,10 +2235,6 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool } else { _ = m.st.UpdateRun(&child) } - if emitResult && accountDenied == nil { - m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload) - } - m.mu.Lock() subParts := append([]core.Part{}, rc.h.assistant...) m.mu.Unlock() @@ -2257,10 +2253,28 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool } fullText := partsText(subParts) if len(subParts) > 0 { - _ = m.appendMessageForEffectiveAccount(child.Owner, &core.Message{ConversationID: child.ConversationID, RunID: child.ID, Role: core.RoleAssistant, Parts: subParts}) + if persistErr := m.appendMessageForEffectiveAccount(child.Owner, &core.Message{ConversationID: child.ConversationID, RunID: child.ID, Role: core.RoleAssistant, Parts: subParts}); persistErr != nil { + log.Printf("[runmgr] subagent result persistence %s: %v", child.ID, persistErr) + status = core.RunFailed + child.Status = status + payload.Status = status + payload.Summary = "执行记录保存失败；已生成的文字结果保留在工作卡，完整过程暂不可回看。" + child.Error = &core.ErrorPart{Code: "run.result_persistence_failed", Message: payload.Summary} + _ = m.st.UpdateRun(&child) + _ = m.writeSubagentTranscriptChecked(rc.ctx, child, subParts, string(status), transcriptCheck) + if guardErr := m.authorizeRunWork(context.Background(), child.Owner, child.ProjectID); guardErr != nil { + accountDenied = guardErr + subParts = nil + fullText = "" + payload.Summary = guardErr.Error() + } + } } if fullText != "" { payload.Text = capSubagentResult(rc, fullText) + if !emitResult { + payload.Text = finalWorkText(subParts) + } if payload.Summary == "" || payload.Summary == "完成" { payload.Summary = truncate(fullText, 160) } @@ -2268,6 +2282,9 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool if line := artifactSummary(subParts); line != "" && status == core.RunCompleted { payload.Summary = strings.TrimSpace(payload.Summary + "\n" + line) } + if emitResult && accountDenied == nil { + m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload) + } m.mu.Lock() delete(m.active, child.ID) @@ -2278,6 +2295,16 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool return payload, nil } +func finalWorkText(parts []core.Part) string { + var final []core.Part + for _, part := range parts { + if part.Type == core.PartText && !part.Process { + final = append(final, part) + } + } + return strings.TrimSpace(partsText(final)) +} + // bgSpawnOpts 后台派发选项：模型 delegate_task 走零值；系统确定性派发 // （材料整理）携带模型档、卡片 KV、内容键、材料标记与排队许可。 type bgSpawnOpts struct { @@ -2323,16 +2350,23 @@ func (m *Manager) spawnBackgroundOpts(parent core.Run, o bgSpawnOpts) (core.ID, if m.bgQueue == nil { m.bgQueue = map[string][]bgQueueItem{} } - m.bgQueue[conv] = append(m.bgQueue[conv], bgQueueItem{parent: parent, opts: o}) m.noticeMu.Unlock() - // 排队卡先行（用户立即可见排队态）；SourceRunID 与执行态在槽位 - // 释放启动时回填。入队与建卡之间释放槽位的窗口由 pump 兜住： - // 下一次 decBackground 会尝试启动它。 - card := m.createBackgroundCard(parent, "", o.name, o.task, o.kv, fmt.Sprintf("排队中（本会话已有 %d 个后台任务，释放后自动开始）", maxBackgroundChildren)) - m.noticeMu.Lock() - if q := m.bgQueue[conv]; len(q) > 0 { - q[len(q)-1].card = card + card, cardErr := m.createBackgroundCardChecked(parent, "", o.name, o.task, o.kv, fmt.Sprintf("排队中（本会话已有 %d 个后台任务，释放后自动开始）", maxBackgroundChildren)) + if cardErr != nil { + if card.ID != "" { + m.finishBackgroundCard(parent.Owner, card, core.AgentResultPayload{}, cardErr, o.kv) + } + return "", cardErr } + m.noticeMu.Lock() + // A slot may have opened while the card was registered. Never enqueue + // a task without its card, or leave it waiting for a release already past. + if m.bgInFlight[conv] < maxBackgroundChildren { + m.bgInFlight[conv]++ + m.noticeMu.Unlock() + return m.startBackgroundSlot(parent, o, card) + } + m.bgQueue[conv] = append(m.bgQueue[conv], bgQueueItem{parent: parent, opts: o, card: card}) m.noticeMu.Unlock() return "", nil } @@ -2359,12 +2393,32 @@ func (m *Manager) startBackgroundSlot(parent core.Run, o bgSpawnOpts, preCard co // 慢通道可观测（任务快慢双速，SR-B10）：后台子代理进收件箱进度卡， // 完成后把结果回投该卡；通知回合路径保持不变，卡片是用户侧回收入口。 bgCard := preCard + var cardErr error if bgCard.ID == "" { - bgCard = m.createBackgroundCard(parent, rc.run.ID, o.name, o.task, o.kv, "") + bgCard, cardErr = m.createBackgroundCardChecked(parent, rc.run.ID, o.name, o.task, o.kv, "") } else { - m.promoteQueuedCard(parent.Owner, preCard, rc.run.ID) + cardErr = m.promoteQueuedCard(parent.Owner, preCard, rc.run.ID) bgCard.SourceRunID = rc.run.ID } + if cardErr != nil { + rc.h.cancel() + now := time.Now().UTC() + rc.run.Status = core.RunFailed + rc.run.Phase = core.PhaseDone + rc.run.EndedAt = &now + rc.run.Error = &core.ErrorPart{Code: "run.work_card_unavailable", Message: "工作卡登记失败，后台任务未执行。"} + _ = m.st.UpdateRun(&rc.run) + m.emit(parent.ConversationID, rc.run.ID, core.EvAgentResult, core.AgentResultPayload{AgentRun: rc.run.ID, Status: core.RunFailed, Summary: rc.run.Error.Message}) + m.mu.Lock() + delete(m.active, rc.run.ID) + m.mu.Unlock() + m.finishBackgroundCard(parent.Owner, bgCard, core.AgentResultPayload{}, fmt.Errorf("工作卡登记失败，后台任务未执行"), o.kv) + if o.taskKey != "" { + m.materialKeyRelease(parent.ConversationID, o.taskKey) + } + m.decBackground(parent.ConversationID) + return "", fmt.Errorf("background task card registration: %w", cardErr) + } kv := o.kv go func() { defer m.decBackground(parent.ConversationID) @@ -2413,8 +2467,13 @@ func (m *Manager) pumpBackgroundQueue(conv string) { // kv 为系统派发任务的附加键值（材料键等），随完成回投保留；status 非空时 // 覆盖默认状态行（排队态文案）。 func (m *Manager) createBackgroundCard(parent core.Run, childID core.ID, name, task string, kv [][2]string, status string) core.ActivityCard { + card, _ := m.createBackgroundCardChecked(parent, childID, name, task, kv, status) + return card +} + +func (m *Manager) createBackgroundCardChecked(parent core.Run, childID core.ID, name, task string, kv [][2]string, status string) (core.ActivityCard, error) { if m.cards == nil { - return core.ActivityCard{} + return core.ActivityCard{}, nil } title := strings.TrimSpace(name) if title == "" { @@ -2435,16 +2494,24 @@ func (m *Manager) createBackgroundCard(parent core.Run, childID core.ID, name, t }, "") if err != nil { log.Printf("[runmgr] background card create: %v", err) - return core.ActivityCard{} + return card, err } - return card + return card, nil } // promoteQueuedCard 排队任务获得槽位启动：回填 SourceRunID 并把状态行转为执行中。 -func (m *Manager) promoteQueuedCard(owner string, card core.ActivityCard, childID core.ID) { +func (m *Manager) promoteQueuedCard(owner string, card core.ActivityCard, childID core.ID) error { if m.cards == nil || card.ID == "" { - return + return nil } + current, version, err := m.st.GetCardVersion(owner, card.ID) + if err != nil { + return err + } + if current.Status != core.CardActive { + return store.ErrConflict + } + card = current card.SourceRunID = childID for i, node := range card.Body { if node.Type != core.UIKV { @@ -2456,9 +2523,11 @@ func (m *Manager) promoteQueuedCard(owner string, card core.ActivityCard, childI } } } - if err := m.updateCardForEffectiveAccount(owner, card); err != nil { + if err := m.saveWorkCard(owner, card, version); err != nil { log.Printf("[runmgr] queued card promote: %v", err) + return err } + return nil } // finishBackgroundCard 把后台子代理终态写回进度卡：成功=结果摘要（含工件行）， @@ -2486,6 +2555,10 @@ func (m *Manager) finishBackgroundCard(owner string, card core.ActivityCard, pay extra := append(append([][2]string(nil), kv...), [2]string{"通道", "慢通道 · 后台执行"}) completed := time.Now().UTC() result, _, resultErr := m.savedWorkResult(owner, card) + if err != nil && payload.Status != core.RunCompleted && payload.Text != "" && m.authorizeRunWork(context.Background(), owner, card.ProjectID) == nil { + result = payload.Text + resultErr = nil + } if resultErr != nil { if err == nil { log.Printf("[runmgr] saved background result unavailable: %v", resultErr) @@ -2497,7 +2570,11 @@ func (m *Manager) finishBackgroundCard(owner string, card core.ActivityCard, pay } hasTextResult := result != "" if result == "" { - result = "本任务未返回文字结果。请查看已保存的文件与完整执行记录。" + if err != nil { + result = summary + } else { + result = "本任务未返回文字结果。请查看已保存的文件与完整执行记录。" + } } work := &core.WorkCard{Result: result, ResultVersion: 1, CompletedAt: &completed} if err == nil && materialKVOf(card) != "" && hasTextResult { diff --git a/internal/runmgr/workcard.go b/internal/runmgr/workcard.go index 1f6b05c..4dfeff0 100644 --- a/internal/runmgr/workcard.go +++ b/internal/runmgr/workcard.go @@ -46,13 +46,7 @@ func (m *Manager) savedWorkResult(owner string, card core.ActivityCard) (string, if err != nil { return "", core.Run{}, err } - var final []core.Part - for _, part := range parts { - if part.Type == core.PartText && !part.Process { - final = append(final, part) - } - } - return strings.TrimSpace(partsText(final)), child, nil + return finalWorkText(parts), child, nil } func (m *Manager) saveWorkCard(owner string, card core.ActivityCard, version int) error { @@ -122,7 +116,31 @@ func (m *Manager) PresentWorkCards(owner string, cards []core.ActivityCard) []co } continue } - c.Work = &core.WorkCard{Result: result, ResultVersion: 1, ReadVersion: 1, CompletedAt: child.EndedAt} + current, version, getErr := m.st.GetCardVersion(owner, c.ID) + if getErr != nil { + continue + } + if current.Work != nil { + *c = current + continue + } + current.Work = &core.WorkCard{Result: result, ResultVersion: 1, ReadVersion: 1, CompletedAt: child.EndedAt} + if m.authorizeRunWork(context.Background(), owner, current.ProjectID) != nil { + continue + } + // Cache the derived projection once without changing historical times or + // emitting a new-result notification for already existing work. + var saveErr error + if m.accountEffectiveAuthorization != nil { + saveErr = m.st.SaveCardVersionedAccountChecked(owner, ¤t, "", version, m.accountAllowImplicitLocal) + } else { + saveErr = m.st.SaveCardVersioned(owner, ¤t, "", version) + } + if saveErr == nil { + *c = current + } else if latest, err := m.st.GetCard(owner, c.ID); err == nil { + *c = latest + } } return cards } @@ -296,12 +314,15 @@ func (m *Manager) directWorkReference(conv core.Conversation, text string) *Work func materialPurposeRequest(text string) bool { compact := strings.ReplaceAll(strings.TrimSpace(text), " ", "") + if strings.Contains(compact, "是否") || strings.Contains(compact, "为什么") || strings.HasSuffix(strings.TrimRight(compact, "？?。.!！"), "吗") { + return false + } for _, negative := range []string{"不要", "不用", "无需", "不需要", "先不", "还是", "或者", "对比"} { if strings.Contains(compact, negative) { return false } } - return (strings.Contains(compact, "这份材料") || strings.Contains(compact, "这份文档")) && (strings.Contains(compact, "总结") || strings.Contains(compact, "摘要") || strings.Contains(compact, "重点")) + return (strings.Contains(compact, "这份材料") || strings.Contains(compact, "这份文档")) && (strings.Contains(compact, "总结") || (strings.Contains(compact, "生成") && strings.Contains(compact, "摘要"))) } func (m *Manager) bindWorkContext(conv core.Conversation, msg core.Message, ref *WorkReference, runIDs ...core.ID) (string, error) { diff --git a/internal/runmgr/workcard_failures_test.go b/internal/runmgr/workcard_failures_test.go new file mode 100644 index 0000000..58a0609 --- /dev/null +++ b/internal/runmgr/workcard_failures_test.go @@ -0,0 +1,111 @@ +package runmgr + +import ( + "errors" + "strings" + "sync/atomic" + "testing" + + "agent-system-backend/internal/cardsvc" + "agent-system-backend/internal/runtime" + core "agent-system-core" +) + +type observedWorkRunner struct { + calls atomic.Int32 + parts []core.Part +} + +func (r *observedWorkRunner) Name() string { return "observed-work" } +func (r *observedWorkRunner) Execute(rc runtime.RunCtx) error { + r.calls.Add(1) + return rc.SetAssistantParts(r.parts) +} + +func TestBackgroundRegistrationFailureStopsBeforeModel(t *testing.T) { + for _, queued := range []bool{false, true} { + t.Run(map[bool]string{false: "create", true: "promote"}[queued], func(t *testing.T) { + runner := &observedWorkRunner{} + c, _ := newCardSpawnEnv(t, runner) + var preCard core.ActivityCard + if queued { + preCard = c.mgr.createBackgroundCard(c.run, "", "排队任务", "", nil, "排队中") + } + c.mgr.cards = cardsvc.New(c.mgr.st, func(_, _ string, typ core.StreamEventType, _ any) error { + if (!queued && typ == core.EvCardCreated) || (queued && typ == core.EvCardUpdated) { + return errors.New("injected card event persistence failure") + } + return nil + }) + c.mgr.noticeMu.Lock() + if c.mgr.bgInFlight == nil { + c.mgr.bgInFlight = map[string]int{} + } + c.mgr.bgInFlight[c.run.ConversationID]++ + c.mgr.noticeMu.Unlock() + if _, err := c.mgr.startBackgroundSlot(c.run, bgSpawnOpts{agentID: "agent.sub.general", name: "任务", task: "执行", taskKey: "registered-key"}, preCard); err == nil { + t.Fatal("registration failure accepted task") + } + if runner.calls.Load() != 0 { + t.Fatal("model executed despite failed registration") + } + waitBackgroundIdle(t, c) + runs, _ := c.mgr.st.ListRunsByConversation(c.run.ConversationID) + for _, run := range runs { + if run.Kind == core.RunSubagent && (run.Status != core.RunFailed || run.EndedAt == nil) { + t.Fatalf("unregistered child not sealed failed: %+v", run) + } + } + c.mgr.mu.Lock() + active := len(c.mgr.active) + c.mgr.mu.Unlock() + if active != 0 { + t.Fatal("failed registration leaked active handle") + } + c.mgr.materialMu.Lock() + keyActive := c.mgr.materialActive[c.run.ConversationID]["registered-key"] + c.mgr.materialMu.Unlock() + if keyActive { + t.Fatal("failed registration leaked material key") + } + cards, _ := c.mgr.cards.List(c.run.Owner, c.run.ConversationID) + if len(cards) != 1 || cards[0].Status != core.CardFailed { + t.Fatalf("persisted card not visibly failed: %+v", cards) + } + }) + } +} + +func TestBackgroundMessagePersistenceFailureRetainsFullPublicResult(t *testing.T) { + final := strings.Repeat("有来源的材料事实。", 400) + "最后一条来源仍然保留" + runner := &observedWorkRunner{parts: []core.Part{{Type: core.PartThinking, Text: "公开思考仅属于过程", Process: true}, {Type: core.PartText, Text: "中间检索输出", Process: true}, {Type: core.PartText, Text: final}}} + c, _ := newCardSpawnEnv(t, runner) + if c.mgr.st.Dialect() != "sqlite" { + t.Skip("SQLite-only assistant-message fault injection; normal workflow tests cover both dialects") + } + if _, err := c.mgr.st.DB().Exec(`CREATE TRIGGER fail_work_message BEFORE INSERT ON messages WHEN NEW.role='assistant' BEGIN SELECT RAISE(FAIL, 'injected assistant message persistence failure'); END`); err != nil { + t.Fatal(err) + } + if _, err := c.SpawnBackground("agent.sub.general", "材料整理", "整理", ""); err != nil { + t.Fatal(err) + } + waitBackgroundIdle(t, c) + cards, _ := c.mgr.cards.List(c.run.Owner, c.run.ConversationID) + if len(cards) != 1 || cards[0].Status != core.CardFailed || cards[0].Work == nil || cards[0].Work.Result != final || cards[0].Work.Decision != nil { + t.Fatalf("public result lost or falsely completed after persistence failure: %+v", cards) + } + if runner.calls.Load() != 1 { + t.Fatal("persistence failure retried model") + } + run, err := c.mgr.st.GetRun(c.run.Owner, cards[0].SourceRunID) + if err != nil || run.Status != core.RunFailed || run.Error == nil || run.Error.Code != "run.result_persistence_failed" { + t.Fatalf("child falsely claimed complete record: %+v %v", run, err) + } + var explanation strings.Builder + for _, node := range cards[0].Body { + explanation.WriteString(node.Markdown) + } + if !strings.Contains(explanation.String(), "保存失败") { + t.Fatal("missing visible record-persistence explanation") + } +} diff --git a/internal/runmgr/workcard_test.go b/internal/runmgr/workcard_test.go index 555b08d..11dc444 100644 --- a/internal/runmgr/workcard_test.go +++ b/internal/runmgr/workcard_test.go @@ -35,10 +35,20 @@ func TestLegacyWorkProjectionCanBeReferencedAndMarkedRead(t *testing.T) { if err := c.mgr.cards.Update(c.run.Owner, card); err != nil { t.Fatal(err) } + card, _ = c.mgr.st.GetCard(c.run.Owner, card.ID) view := c.mgr.PresentWorkCards(c.run.Owner, []core.ActivityCard{card})[0] if view.Work == nil || view.Work.Result == "" { t.Fatal("historical fullresult missing") } + stored, version, err := c.mgr.st.GetCardVersion(c.run.Owner, card.ID) + if err != nil || stored.Work == nil || !stored.UpdatedAt.Equal(card.UpdatedAt) || stored.Work.ReadVersion != 1 { + t.Fatal("legacy projection was not cached without rewriting historical time") + } + c.mgr.PresentWorkCards(c.run.Owner, []core.ActivityCard{stored}) + _, nextVersion, _ := c.mgr.st.GetCardVersion(c.run.Owner, card.ID) + if version != nextVersion { + t.Fatal("cached projection rewritten on every read") + } if err := c.mgr.ValidateWorkReference(c.run.Owner, c.run.ConversationID, WorkReference{card.ID, 1}); err != nil { t.Fatalf("visible reference rejected: %v", err) } @@ -232,6 +242,17 @@ func TestExplicitWorkReferenceAllowsNewSummaryAfterPreviousChoice(t *testing.T) } } +func TestImplicitWorkBindingDoesNotCaptureQuestionsAboutMaterial(t *testing.T) { + for _, question := range []string{"这份材料的重点适合儿童患者吗？", "这份材料的摘要是否覆盖维护要求？", "这份文档为什么把这个列为重点？"} { + if materialPurposeRequest(question) { + t.Fatalf("ordinary question captured: %s", question) + } + } + if !materialPurposeRequest("总结一下这份材料的摘要吧，我先看看里面有什么重点的信息") { + t.Fatal("explicit summary request no longer recognized") + } +} + func TestButtonAndDirectSummaryRaceKeepsOneExecution(t *testing.T) { c, card := completedMaterialWork(t) hold := &holdRunner{release: make(chan struct{})}