Acknowledge automatic material intake from task state
Why: The first material reply could stream an unsupported numeric relationship before the sourced background card had any verified finding. Evidence: Task 2 red/green SQLite run tests cover parent provider suppression, queue/reuse, completed source binding, focus retention, squad entry, failed/unknown state, and attempt ordering; full Backend vet/Go/race SELF receipt /tmp/fast-slow-u3-r05-task2-backend-final-v2.json exited 0. Impact: Prestarted material sends receive one server-owned status message with zero parent model calls; small-text and explicit completed follow-up paths retain their existing runner. No public contract or model budget increases. Refs: U3 r0.5 approved design and plan; Backend PR #193
This commit is contained in:
@@ -139,18 +139,28 @@ func buildMaterialTask(eff core.EffectiveSettings, docs []core.ParsedDocument) s
|
||||
// materialIntake 预启动决策:附件即保证的「发送瞬间」分支。
|
||||
// 返回注入第 0 回合提示的系统说明;空串 = 本回合不预启动(留给 stop 兜底)。
|
||||
func (m *Manager) materialIntake(rc *runCtx) string {
|
||||
return m.materialIntakeDecision(rc).Note
|
||||
}
|
||||
|
||||
func (m *Manager) materialIntakeDecision(rc *runCtx) materialIntakeDecision {
|
||||
if !rc.eff.GetBool("runtime.materialIntake", true) || len(rc.attachments) == 0 {
|
||||
return ""
|
||||
return materialIntakeDecision{State: materialNone}
|
||||
}
|
||||
if note, existing := m.materialExisting(rc, rc.materialRevision()); existing {
|
||||
return note
|
||||
key := rc.materialRevision()
|
||||
if note, existing := m.materialExisting(rc, key); existing {
|
||||
decision := m.materialDecisionForNote(rc, key, note)
|
||||
if decision.State == materialRunning {
|
||||
decision.State = materialReusedRunning
|
||||
decision.Note = materialPendingNote
|
||||
}
|
||||
return decision
|
||||
}
|
||||
userText := messageText(rc.userMessage)
|
||||
threshold := int(rc.eff.GetNumber(materialCharsKey, materialCharsDefaut))
|
||||
if userText != "" && materialChars(rc.attachments) < threshold {
|
||||
return "" // 文字+小材料:模型先答,回合结束兜底
|
||||
return materialIntakeDecision{State: materialNone} // 文字+小材料:模型先答,回合结束兜底
|
||||
}
|
||||
return m.materialSpawn(rc, rc.materialRevision())
|
||||
return m.materialSpawnDecision(rc, key)
|
||||
}
|
||||
|
||||
// ensureMaterial stop 兜底入口(幂等:只在首个调用实际派发并返回说明)。
|
||||
@@ -238,6 +248,10 @@ func materialKVOf(card core.ActivityCard) string {
|
||||
// materialSpawn 启动材料整理后台任务(建卡 + 注册内容键)。
|
||||
// 派发失败(如容量满,I3 前为硬拒绝)返回空串,自然延后到下一次兜底机会。
|
||||
func (m *Manager) materialSpawn(rc *runCtx, key string) string {
|
||||
return m.materialSpawnDecision(rc, key).Note
|
||||
}
|
||||
|
||||
func (m *Manager) materialSpawnDecision(rc *runCtx, key string) materialIntakeDecision {
|
||||
short := key
|
||||
if len(short) > 12 {
|
||||
short = short[:12]
|
||||
@@ -256,9 +270,9 @@ func (m *Manager) materialSpawn(rc *runCtx, key string) string {
|
||||
}
|
||||
if _, err := m.spawnBackgroundOpts(rc.run, opts); err != nil {
|
||||
log.Printf("[runmgr] material intake spawn: %v", err)
|
||||
return ""
|
||||
return materialIntakeDecision{State: materialFailed}
|
||||
}
|
||||
return materialStartedNote
|
||||
return m.materialDecisionForNote(rc, key, materialStartedNote)
|
||||
}
|
||||
|
||||
// materialKeyRegister/RegisterRelease 内容键注册表(内存档;卡片 KV 兜重启)。
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
package runmgr
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"agent-system-backend/internal/runtime"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
type materialIntakeState string
|
||||
|
||||
const (
|
||||
materialNone materialIntakeState = "none"
|
||||
materialQueued materialIntakeState = "queued"
|
||||
materialRunning materialIntakeState = "running"
|
||||
materialReusedRunning materialIntakeState = "reused_running"
|
||||
materialCompleted materialIntakeState = "completed"
|
||||
materialFailed materialIntakeState = "failed"
|
||||
materialUnknown materialIntakeState = "unknown"
|
||||
)
|
||||
|
||||
type materialIntakeDecision struct {
|
||||
Note string
|
||||
State materialIntakeState
|
||||
CardID core.ID
|
||||
}
|
||||
|
||||
func (d materialIntakeDecision) needsAcknowledgement(userText string) bool {
|
||||
return d.State == materialRunning || d.State == materialReusedRunning || d.State == materialQueued ||
|
||||
d.State == materialFailed || d.State == materialUnknown ||
|
||||
(d.State == materialCompleted && userText == "")
|
||||
}
|
||||
|
||||
// materialDecisionForNote reads server-owned task state after intake or reuse;
|
||||
// an absent card is unknown rather than proof that work has started.
|
||||
func (m *Manager) materialDecisionForNote(rc *runCtx, key, note string) materialIntakeDecision {
|
||||
decision := materialIntakeDecision{Note: note, State: materialUnknown}
|
||||
if m.cards == nil {
|
||||
return decision
|
||||
}
|
||||
cards, err := m.cards.List(rc.run.Owner, rc.run.ConversationID)
|
||||
if err != nil {
|
||||
return decision
|
||||
}
|
||||
var current *core.ActivityCard
|
||||
for i := range cards {
|
||||
card := &cards[i]
|
||||
if card.Kind != core.CardKindProgress || card.ProjectID != rc.run.ProjectID ||
|
||||
card.Task == nil || card.Task.InputRevision != key {
|
||||
continue
|
||||
}
|
||||
if current == nil || card.CreatedAt.After(current.CreatedAt) ||
|
||||
(card.CreatedAt.Equal(current.CreatedAt) && current.Task.Terminal() && !card.Task.Terminal()) {
|
||||
current = card
|
||||
}
|
||||
}
|
||||
if current != nil {
|
||||
decision.CardID = current.ID
|
||||
if current.Task.ExecutionState == "completed" {
|
||||
if m.taskSourcesAvailable(rc.run.Owner, rc.run.ProjectID, current.Task) {
|
||||
decision.State = materialCompleted
|
||||
decision.Note = materialDoneNote
|
||||
} else {
|
||||
decision.State = materialUnknown
|
||||
}
|
||||
} else if current.Task.ExecutionState == "queued" {
|
||||
decision.State = materialQueued
|
||||
} else if current.Task.Terminal() {
|
||||
decision.State = materialFailed
|
||||
decision.Note = ""
|
||||
} else {
|
||||
decision.State = materialRunning
|
||||
}
|
||||
}
|
||||
return decision
|
||||
}
|
||||
|
||||
type materialAckRunner struct {
|
||||
decision materialIntakeDecision
|
||||
attachmentCount int
|
||||
hasUserFocus bool
|
||||
}
|
||||
|
||||
func (materialAckRunner) Name() string { return "material-acknowledgement" }
|
||||
|
||||
func (r materialAckRunner) Execute(rc runtime.RunCtx) error {
|
||||
if err := rc.Context().Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
text := r.reply()
|
||||
if err := rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseResponding}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: 0, Text: text}); err != nil {
|
||||
return err
|
||||
}
|
||||
return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: text}})
|
||||
}
|
||||
|
||||
func (r materialAckRunner) reply() string {
|
||||
lead := ""
|
||||
if r.decision.State == materialCompleted {
|
||||
lead = "这批材料已整理完成,可在“材料整理”卡查看结果。"
|
||||
} else if r.decision.State == materialQueued {
|
||||
lead = fmt.Sprintf("已收到 %d 份材料,整理任务正在排队。", r.attachmentCount)
|
||||
} else if r.decision.State == materialReusedRunning {
|
||||
lead = "这批材料仍在整理,已有任务会继续更新“材料整理”卡。"
|
||||
} else if r.decision.State == materialFailed {
|
||||
if r.decision.CardID == "" {
|
||||
lead = "材料已收到,但后台整理未能启动;请稍后重新提出请求。"
|
||||
} else {
|
||||
lead = "后台整理未能完成;请查看“材料整理”卡中的已保存成果和未完成范围。"
|
||||
}
|
||||
} else if r.decision.State == materialUnknown {
|
||||
if r.decision.CardID == "" {
|
||||
return "材料已收到,但暂时无法确认整理状态;当前没有可核对的工作卡,请稍后刷新会话或重新提出请求。数字、分期和归属判断待确认。"
|
||||
}
|
||||
lead = "材料已收到,但暂时无法确认整理状态;请查看工作卡。"
|
||||
} else {
|
||||
lead = fmt.Sprintf("已收到 %d 份材料,正在整理。", r.attachmentCount)
|
||||
}
|
||||
if r.hasUserFocus {
|
||||
lead += "你提出的工作重点已带入整理任务。"
|
||||
}
|
||||
return lead + "材料中的数字、分期和归属判断在核对依据前均标为待确认;有依据的发现会显示在“材料整理”卡。"
|
||||
}
|
||||
@@ -0,0 +1,408 @@
|
||||
package runmgr
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-system-backend/internal/runtime"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
type materialParentSpy struct{ calls atomic.Int32 }
|
||||
|
||||
type materialFocusRunner struct {
|
||||
tasks chan string
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func (*materialFocusRunner) Name() string { return "material-focus" }
|
||||
|
||||
func (r *materialFocusRunner) Execute(rc runtime.RunCtx) error {
|
||||
r.tasks <- rc.Subtask()
|
||||
<-r.release
|
||||
return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "整理完成"}})
|
||||
}
|
||||
|
||||
func (*materialParentSpy) Name() string { return "material-parent-spy" }
|
||||
|
||||
func (s *materialParentSpy) Execute(rc runtime.RunCtx) error {
|
||||
s.calls.Add(1)
|
||||
return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "剩余 4 项(含 3 项依赖接口)是否纳入第二期?"}})
|
||||
}
|
||||
|
||||
func startMaterialAckRun(t *testing.T, c *runCtx, docs []core.ParsedDocument, userText string) core.Run {
|
||||
t.Helper()
|
||||
conv, err := c.mgr.st.GetConversation(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
user := core.Message{ConversationID: conv.ID, Role: core.RoleUser}
|
||||
if userText != "" {
|
||||
user.Parts = []core.Part{{Type: core.PartText, Text: userText}}
|
||||
}
|
||||
if err := c.mgr.st.AppendMessage(&user); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run, err := c.mgr.StartConversationRun(conv, user, docs, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitUntil(t, func() bool {
|
||||
current, err := c.mgr.st.GetRunByID(run.ID)
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[conv.ID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return err == nil && current.Status == core.RunCompleted && !busy
|
||||
})
|
||||
return run
|
||||
}
|
||||
|
||||
func materialAckTextForRun(t *testing.T, c *runCtx, runID core.ID) string {
|
||||
t.Helper()
|
||||
msgs, err := c.mgr.st.ListMessages(c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
visible := ""
|
||||
for _, msg := range msgs {
|
||||
if msg.RunID != runID || msg.Role != core.RoleAssistant {
|
||||
continue
|
||||
}
|
||||
for _, part := range msg.Parts {
|
||||
if part.Type == core.PartText && !part.Process {
|
||||
visible += part.Text
|
||||
}
|
||||
}
|
||||
}
|
||||
return visible
|
||||
}
|
||||
|
||||
func waitMaterialAckTestIdle(t *testing.T, c *runCtx) {
|
||||
t.Helper()
|
||||
waitUntil(t, func() bool {
|
||||
convID := c.run.ConversationID
|
||||
c.mgr.noticeMu.Lock()
|
||||
active := c.mgr.bgInFlight[convID]
|
||||
queued := len(c.mgr.bgQueue[convID])
|
||||
delivering := c.mgr.delivering[convID]
|
||||
pending := len(c.mgr.notices[convID])
|
||||
c.mgr.noticeMu.Unlock()
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[convID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return active == 0 && queued == 0 && !delivering && pending == 0 && !busy
|
||||
})
|
||||
}
|
||||
|
||||
func TestMaterialAckAttachmentOnlyDoesNotCallParentModel(t *testing.T) {
|
||||
hold := &holdRunner{release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, hold)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
t.Cleanup(func() {
|
||||
close(hold.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
run := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
if got := parent.calls.Load(); got != 0 {
|
||||
t.Fatalf("parent provider path called %d times", got)
|
||||
}
|
||||
visible := materialAckTextForRun(t, c, run.ID)
|
||||
if !strings.Contains(visible, "待确认") || strings.Contains(visible, "剩余 4 项(含 3 项依赖接口)") {
|
||||
t.Fatalf("first visible response is not a truthful host acknowledgement: %q", visible)
|
||||
}
|
||||
childIDs, err := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
||||
if err != nil || len(childIDs) != 1 {
|
||||
t.Fatalf("material children = %v, err = %v", childIDs, err)
|
||||
}
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil || len(cards) != 1 || cards[0].Task == nil || cards[0].Task.CheckpointSeq != 0 {
|
||||
t.Fatalf("acknowledgement changed task checkpoints: cards=%+v err=%v", cards, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckQueueAndReuse(t *testing.T) {
|
||||
waiter := &gateRunner{release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, waiter)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
t.Cleanup(func() {
|
||||
close(waiter.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
for i := 0; i < 3; i++ {
|
||||
if _, err := c.SpawnBackground("agent.sub.general", fmt.Sprintf("占位%d", i), "等待", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
first := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
second := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
for _, run := range []core.Run{first, second} {
|
||||
text := materialAckTextForRun(t, c, run.ID)
|
||||
if !strings.Contains(text, "排队") || strings.Contains(text, "正在整理") {
|
||||
t.Fatalf("queued task was described as active work: %q", text)
|
||||
}
|
||||
}
|
||||
if got := parent.calls.Load(); got != 0 {
|
||||
t.Fatalf("queued material triggered %d parent provider calls", got)
|
||||
}
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
materialCards := 0
|
||||
for _, card := range cards {
|
||||
if card.Title == materialCardTitle {
|
||||
materialCards++
|
||||
if card.Task == nil || card.Task.ExecutionState != "queued" {
|
||||
t.Fatalf("material card not queued: %+v", card.Task)
|
||||
}
|
||||
}
|
||||
}
|
||||
if materialCards != 1 {
|
||||
t.Fatalf("same material was enqueued %d times", materialCards)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckActiveReuseDoesNotClaimNewStart(t *testing.T) {
|
||||
hold := &holdRunner{release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, hold)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
t.Cleanup(func() {
|
||||
close(hold.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
_ = startMaterialAckRun(t, c, materialDocs(), "")
|
||||
second := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
text := materialAckTextForRun(t, c, second.ID)
|
||||
if !strings.Contains(text, "仍在整理") || strings.Contains(text, "已重新启动") || parent.calls.Load() != 0 {
|
||||
t.Fatalf("active material reuse was announced as new work: %q calls=%d", text, parent.calls.Load())
|
||||
}
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil || len(cards) != 1 {
|
||||
t.Fatalf("same active material created another task: cards=%d err=%v", len(cards), err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckCompletedReuseNoTextAndUserFollowup(t *testing.T) {
|
||||
c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{{Type: core.PartText, Text: "整理完成:需要核对验收条件。"}}})
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
bindTrustedMaterialNoticeInput(t, c)
|
||||
docs := c.attachments
|
||||
_ = startMaterialAckRun(t, c, docs, "")
|
||||
key := c.materialRevision()
|
||||
waitUntil(t, func() bool {
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil || len(cards) != 1 || cards[0].Task == nil || cards[0].Task.ExecutionState != "completed" {
|
||||
return false
|
||||
}
|
||||
c.mgr.materialMu.Lock()
|
||||
active := c.mgr.materialActive[string(c.run.ConversationID)][key]
|
||||
c.mgr.materialMu.Unlock()
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[c.run.ConversationID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return !active && !busy && len(systemTexts(t, c)) > 0
|
||||
})
|
||||
c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
|
||||
decision := c.mgr.materialIntakeDecision(secondRunCtx(c, docs, ""))
|
||||
c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
|
||||
if decision.State != materialCompleted || decision.Note != materialDoneNote {
|
||||
t.Fatalf("completed card was described as still running: %+v", decision)
|
||||
}
|
||||
before := parent.calls.Load()
|
||||
reused := startMaterialAckRun(t, c, docs, "")
|
||||
reply := materialAckTextForRun(t, c, reused.ID)
|
||||
if !strings.Contains(reply, "已整理完成") || strings.Contains(reply, "正在整理") ||
|
||||
parent.calls.Load() != before {
|
||||
t.Fatalf("completed no-text reuse was not a safe acknowledgement: reply=%q before=%d after=%d", reply, before, parent.calls.Load())
|
||||
}
|
||||
newQuestion := startMaterialAckRun(t, c, docs, "请回答新的具体问题")
|
||||
if parent.calls.Load() != before+1 || !strings.Contains(materialAckTextForRun(t, c, newQuestion.ID), "剩余 4 项") {
|
||||
t.Fatalf("explicit follow-up did not retain the ordinary conversation path")
|
||||
}
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil || len(cards) != 1 {
|
||||
t.Fatalf("completed material was respawned: cards=%d err=%v", len(cards), err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckCompletedReuseRequiresTrustedSources(t *testing.T) {
|
||||
c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{{Type: core.PartText, Text: "整理完成"}}})
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
c.mgr.runner = &materialParentSpy{}
|
||||
docs := materialDocs() // parsed-only legacy material has no artifact/hash binding
|
||||
c.attachments = docs
|
||||
_ = startMaterialAckRun(t, c, docs, "")
|
||||
waitUntil(t, func() bool {
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
return err == nil && len(cards) == 1 && cards[0].Task != nil && cards[0].Task.ExecutionState == "completed"
|
||||
})
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
key := c.materialRevision()
|
||||
c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
|
||||
defer c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
|
||||
decision := c.mgr.materialIntakeDecision(secondRunCtx(c, docs, ""))
|
||||
if decision.State == materialCompleted || strings.Contains(materialAckRunner{decision: decision, attachmentCount: len(docs)}.reply(), "已整理完成") {
|
||||
t.Fatalf("unbound completed card was reused as verified: %+v", decision)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckPrefersCurrentAttemptOverRecentlyUpdatedHistory(t *testing.T) {
|
||||
c, _ := newCardSpawnEnv(t, &captureRunner{})
|
||||
c.attachments = materialDocs()
|
||||
key := c.materialRevision()
|
||||
opts := bgSpawnOpts{name: materialCardTitle, material: true, taskKey: key, documents: c.attachments}
|
||||
oldBinding, oldCard, err := c.mgr.registerBackgroundTask(c.run, opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := c.mgr.finishTaskProjection(oldBinding, core.AgentResultPayload{}, errors.New("old attempt failed")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
newBinding, newCard, err := c.mgr.registerBackgroundTask(c.run, opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := c.mgr.startTaskProjection(newBinding, "run-new"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.mgr.cards.SetPinned(c.run.Owner, oldCard.ID, true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
||||
if err != nil || len(cards) != 2 || cards[0].ID != oldCard.ID {
|
||||
t.Fatalf("fixture did not put old attempt first: %+v err=%v", cards, err)
|
||||
}
|
||||
decision := c.mgr.materialDecisionForNote(c, key, materialPendingNote)
|
||||
if decision.CardID != newCard.ID || decision.State != materialRunning {
|
||||
t.Fatalf("older updated attempt masked current work: %+v", decision)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckFailedCardDoesNotClaimStartFailure(t *testing.T) {
|
||||
c, _ := newCardSpawnEnv(t, &captureRunner{})
|
||||
c.attachments = materialDocs()
|
||||
key := c.materialRevision()
|
||||
binding, card, err := c.mgr.registerBackgroundTask(c.run, bgSpawnOpts{
|
||||
name: materialCardTitle, material: true, taskKey: key, documents: c.attachments,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := c.mgr.finishTaskProjection(binding, core.AgentResultPayload{}, errors.New("worker stopped after starting")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
|
||||
defer c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
|
||||
decision := c.mgr.materialIntakeDecision(c)
|
||||
text := materialAckRunner{decision: decision, attachmentCount: len(c.attachments)}.reply()
|
||||
if decision.CardID != card.ID || decision.State != materialFailed ||
|
||||
!strings.Contains(text, "未能完成") || strings.Contains(text, "未能启动") {
|
||||
t.Fatalf("failed work was mislabeled as never started: state=%+v reply=%q", decision, text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckPreservesUserFocusAndSmallPath(t *testing.T) {
|
||||
child := &materialFocusRunner{tasks: make(chan string, 1), release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, child)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
t.Cleanup(func() {
|
||||
close(child.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
long := []core.ParsedDocument{{Meta: core.DocMeta{Filename: "长材料.txt", Parser: "txt"},
|
||||
Blocks: []core.DocBlock{{Type: core.BlockPara, Text: strings.Repeat("材料", 6100)}}}}
|
||||
focus := "重点核对预算审批"
|
||||
run := startMaterialAckRun(t, c, long, focus)
|
||||
select {
|
||||
case task := <-child.tasks:
|
||||
if !strings.Contains(task, "用户本次工作重点:\n"+focus) {
|
||||
t.Fatalf("original user focus missing from slow task: %q", task)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("material child did not receive the original user focus")
|
||||
}
|
||||
if parent.calls.Load() != 0 || !strings.Contains(materialAckTextForRun(t, c, run.ID), "工作重点") {
|
||||
t.Fatalf("long material did not use a focus-aware host acknowledgement")
|
||||
}
|
||||
small := []core.ParsedDocument{{Meta: core.DocMeta{Filename: "短说明.txt", Parser: "txt"},
|
||||
Blocks: []core.DocBlock{{Type: core.BlockPara, Text: "这是一条短说明。"}}}}
|
||||
followup := startMaterialAckRun(t, c, small, "请直接回答这份短说明")
|
||||
if parent.calls.Load() != 1 || !strings.Contains(materialAckTextForRun(t, c, followup.ID), "剩余 4 项") {
|
||||
t.Fatalf("text with small attachment did not retain the ordinary runner")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckSkipsSquadEntrypoint(t *testing.T) {
|
||||
hold := &holdRunner{release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, hold)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
direct := &materialParentSpy{}
|
||||
squad := &materialParentSpy{}
|
||||
c.mgr.runner, c.mgr.squad = direct, squad
|
||||
if err := c.mgr.settings.MergeUser(c.run.Owner, map[string]any{"runtime.entry": "squad"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
close(hold.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
run := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
if direct.calls.Load() != 0 || squad.calls.Load() != 0 ||
|
||||
!strings.Contains(materialAckTextForRun(t, c, run.ID), "待确认") {
|
||||
t.Fatalf("automatic material intake reached a model runner: direct=%d squad=%d", direct.calls.Load(), squad.calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialAckFailureAndUnknown(t *testing.T) {
|
||||
t.Run("spawn-failed", func(t *testing.T) {
|
||||
c, _ := newCardSpawnEnv(t, &captureRunner{})
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
c.mgr.shuttingDown.Store(true)
|
||||
run := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
reply := materialAckTextForRun(t, c, run.ID)
|
||||
if parent.calls.Load() != 0 || !strings.Contains(reply, "未能启动") || strings.Contains(reply, "正在整理") {
|
||||
t.Fatalf("failed spawn produced a false acknowledgement: calls=%d reply=%q", parent.calls.Load(), reply)
|
||||
}
|
||||
})
|
||||
t.Run("card-unavailable", func(t *testing.T) {
|
||||
child := &gateRunner{entered: make(chan struct{}), release: make(chan struct{})}
|
||||
c, _ := newCardSpawnEnv(t, child)
|
||||
c.mgr.SetExecutionBudgetFn(nil, 0)
|
||||
parent := &materialParentSpy{}
|
||||
c.mgr.runner = parent
|
||||
cards := c.mgr.cards
|
||||
c.mgr.cards = nil
|
||||
t.Cleanup(func() {
|
||||
c.mgr.cards = cards
|
||||
close(child.release)
|
||||
waitMaterialAckTestIdle(t, c)
|
||||
})
|
||||
run := startMaterialAckRun(t, c, materialDocs(), "")
|
||||
select {
|
||||
case <-child.entered:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("material child never started")
|
||||
}
|
||||
reply := materialAckTextForRun(t, c, run.ID)
|
||||
if parent.calls.Load() != 0 || !strings.Contains(reply, "无法确认整理状态") ||
|
||||
strings.Contains(reply, "正在整理") || strings.Contains(reply, "查看工作卡") {
|
||||
t.Fatalf("unavailable card was described as running: calls=%d reply=%q", parent.calls.Load(), reply)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1409,7 +1409,11 @@ func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Messa
|
||||
// 材料整理(附件即保证):无文字/大材料发送瞬间预启动,说明注入第 0 回合;
|
||||
// 其余带附件回合由 runtime stop 钩子兜底(模型已自行委派时尊重模型判断)。
|
||||
if launch == nil {
|
||||
rc.materialNote = m.materialIntake(rc)
|
||||
decision := m.materialIntakeDecision(rc)
|
||||
rc.materialNote = decision.Note
|
||||
if decision.needsAcknowledgement(messageText(userMessage)) {
|
||||
runner = materialAckRunner{decision: decision, attachmentCount: len(attachments), hasUserFocus: messageText(userMessage) != ""}
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
// 执行许可(U3,R8):排队有界;获取失败(队列满/取消/存储故障)以
|
||||
|
||||
Reference in New Issue
Block a user