Stop untracked background execution and retain results on record failures
ci / vet-and-test (pull_request) Successful in 1m2s

Why: Card registration or transcript persistence failures must not orphan paid work or discard a generated result, and ordinary questions must not match summary requests.

Evidence: Registration/promote and message-insert fault regressions, implicit-question checks and scoped race tests pass; legacy projection caching preserves historical times.

Impact: Fail before execution when tracking is unavailable, preserve full public final text with an explicit failure state, cache legacy projections once, and repair the concurrent material test fixture.

Refs: feature/workbench-flow-completion independent review closure
This commit is contained in:
2026-10-15 07:37:10 +00:00
parent 5561df657e
commit 7d6490feee
5 changed files with 267 additions and 35 deletions
+6 -4
View File
@@ -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 {
+99 -22
View File
@@ -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 {
+30 -9
View File
@@ -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, &current, "", version, m.accountAllowImplicitLocal)
} else {
saveErr = m.st.SaveCardVersioned(owner, &current, "", 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) {
+111
View File
@@ -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")
}
}
+21
View File
@@ -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{})}