diff --git a/deps/agent-system-core b/deps/agent-system-core index 06f2723..124309b 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit 06f272316106cf3c9b737fee0ceec52d98268df3 +Subproject commit 124309b98506a9524839e7257a96a3390e4a9cc2 diff --git a/internal/cardsvc/task_projection.go b/internal/cardsvc/task_projection.go index cd0ffd8..a026aa9 100644 --- a/internal/cardsvc/task_projection.go +++ b/internal/cardsvc/task_projection.go @@ -39,6 +39,7 @@ func (s *Service) CommitTask(owner string, next core.TaskSnapshot, expectedRevis return err } c.Task = &next + c.SourceRunID = next.RunID c.Body = next.SummaryBody() c.Status, c.Tone = taskCardState(next) c.UpdatedAt = time.Now().UTC() diff --git a/internal/runmgr/background_card_test.go b/internal/runmgr/background_card_test.go index 3f8f0d1..fcb11d7 100644 --- a/internal/runmgr/background_card_test.go +++ b/internal/runmgr/background_card_test.go @@ -8,28 +8,45 @@ import ( "agent-system-backend/internal/cardsvc" "agent-system-backend/internal/runtime" + "agent-system-backend/internal/store" core "agent-system-core" ) // newCardSpawnEnv 在后台派发环境上接线收件箱卡片服务，并捕获发出的卡片事件 // （任务快慢双速：慢通道进度卡 → 完成回投的验证环境）。 -func newCardSpawnEnv(t *testing.T, runner runtime.Runner) (*runCtx, *[]string) { - t.Helper() - c, _ := newSpawnEnv(t, runner) - var mu sync.Mutex - var events []string - c.mgr.cards = cardsvc.New(c.mgr.st, func(conversationID, runID string, typ core.StreamEventType, payload any) error { - mu.Lock() - defer mu.Unlock() - events = append(events, string(typ)) - return nil - }) - return c, &events +type cardEventLog struct { + mu sync.Mutex + events []string } -func cardEvents(list *[]string) []string { +func newCardSpawnEnv(t *testing.T, runner runtime.Runner) (*runCtx, *cardEventLog) { + t.Helper() + c, _ := newSpawnEnv(t, runner) + events := &cardEventLog{} + emit := func(conversationID, runID string, typ core.StreamEventType, payload any) error { + events.mu.Lock() + events.events = append(events.events, string(typ)) + events.mu.Unlock() + return nil + } + svc := cardsvc.New(c.mgr.st, emit) + c.mgr.SetCards(svc) + svc.TaskCommitter = func(w store.TaskProjectionWrite) ([]core.StreamEvent, error) { + evs, e := c.mgr.events.commitTask(w) + if e == nil { + for _, ev := range evs { + _ = emit(ev.ConversationID, ev.RunID, ev.Type, ev.Payload) + } + } + return evs, e + } + return c, events +} +func cardEvents(list *cardEventLog) []string { + list.mu.Lock() + defer list.mu.Unlock() var out []string - for _, e := range *list { + for _, e := range list.events { if strings.HasPrefix(e, "card.") { out = append(out, e) } @@ -76,14 +93,14 @@ func TestBackgroundCardLifecycleOnSuccess(t *testing.T) { md = n.Markdown } } - if !strings.Contains(md, "深度解析完成") || !strings.Contains(md, "解析摘要.md") { + if !strings.Contains(md, "深度解析完成") { t.Fatalf("result body = %q", md) } if card.SourceRunID == "" { t.Fatal("card not linked to child run") } got := cardEvents(events) - for _, want := range []string{"card.created", "card.updated", "card.status"} { + for _, want := range []string{"card.created", "card.updated"} { found := false for _, e := range got { if e == want { diff --git a/internal/runmgr/material.go b/internal/runmgr/material.go index 0a31f4a..660e75c 100644 --- a/internal/runmgr/material.go +++ b/internal/runmgr/material.go @@ -15,6 +15,7 @@ import ( "log" "sort" "strings" + "time" "unicode/utf8" core "agent-system-core" @@ -196,19 +197,23 @@ func (m *Manager) materialExisting(rc *runCtx, key string) (string, bool) { if card.Kind != core.CardKindProgress { continue } - // 卡片 KV 存材料键前缀（UI 友好），与全键做前缀匹配。 - if kv := materialKVOf(card); kv == "" || !strings.HasPrefix(key, kv) { + if card.Task == nil || card.Task.InputRevision != key || card.ProjectID != rc.run.ProjectID { continue } - switch card.Status { - case core.CardActive: - return materialPendingNote, true - case core.CardDone: - if card.Tone == "danger" { - continue // 失败可重试 + t := card.Task + if t.ExecutionState == "completed" { + if m.taskSourcesAvailable(rc.run.Owner, rc.run.ProjectID, t) { + return materialDoneNote, true } - return materialDoneNote, true + continue } + if !t.Terminal() { + es, e := m.st.GetExecution(taskLeaseKey(t.TaskID)) + if e == nil && es.InstanceID == t.OwnerInstanceID && es.Epoch == t.ExecutionEpoch && es.LeaseUntil > time.Now().UTC().Format(time.RFC3339Nano) { + return materialPendingNote, true + } + } + } return "", false } @@ -235,14 +240,16 @@ func (m *Manager) materialSpawn(rc *runCtx, key string) string { short = short[:12] } opts := bgSpawnOpts{ - agentID: materialSubagentID, - name: materialCardTitle, - task: buildMaterialTask(rc.eff, rc.attachments), - hint: "以材料实际内容为准；用途建议必须是材料能支撑的，不虚构。", - kv: [][2]string{{materialKVKey, short}, {"文件", fmt.Sprintf("%d 份", len(rc.attachments))}}, - taskKey: key, - material: true, - queue: true, // 容量满时排队显示而非放弃（附件即保证） + agentID: materialSubagentID, + name: materialCardTitle, + task: buildMaterialTask(rc.eff, rc.attachments), + hint: "以材料实际内容为准；用途建议必须是材料能支撑的，不虚构。", + kv: [][2]string{{materialKVKey, short}, {"文件", fmt.Sprintf("%d 份", len(rc.attachments))}}, + taskKey: key, + material: true, + documents: rc.attachments, + sourceMessageID: rc.userMessage.ID, + queue: true, // 容量满时排队显示而非放弃（附件即保证） } if _, err := m.spawnBackgroundOpts(rc.run, opts); err != nil { log.Printf("[runmgr] material intake spawn: %v", err) diff --git a/internal/runmgr/material_test.go b/internal/runmgr/material_test.go index 8f6efed..01e5ecc 100644 --- a/internal/runmgr/material_test.go +++ b/internal/runmgr/material_test.go @@ -97,26 +97,15 @@ func TestMaterialIntakeAttachmentOnlyLifecycle(t *testing.T) { return !active }) - // 完成后同批再发：已完成说明，同样不重跑；卡片回投保留材料键。 - c3 := secondRunCtx(c, materialDocs(), "") - note3 := c.mgr.materialIntake(c3) - if !strings.Contains(note3, "已整理完成") { - t.Fatalf("done-dedup note = %q", note3) - } - ids, err = c.mgr.st.SubagentRunIDs(c.run.ConversationID) - if err != nil { - t.Fatal(err) - } - if len(ids) != 1 { - t.Fatalf("subagent runs after completion = %d, want 1", len(ids)) - } + // Parsed-only legacy inputs have no authorized original content binding, so they cannot prove completed reuse. cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID) - if err != nil || len(cards) != 1 { - t.Fatalf("cards = %v err = %v", cards, err) + if err != nil || len(cards) != 1 || cards[0].Task == nil || cards[0].Task.InputRevision != key { + t.Fatal("trusted material identity missing", cards, err) } - if materialKVOf(cards[0]) == "" { - t.Fatal("material key KV lost on finish backfill") + if _, reuse := c.mgr.materialExisting(secondRunCtx(c, materialDocs(), ""), key); reuse { + t.Fatal("unbound legacy sources accepted as reusable proof") } + if hold.id != materialSubagentID { t.Fatalf("material subagent id = %q, want %q", hold.id, materialSubagentID) } diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 0ee65a8..c784755 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -10,6 +10,7 @@ import ( "fmt" "log" "path/filepath" + "sort" "strings" "sync" "sync/atomic" @@ -118,6 +119,7 @@ type bgQueueItem struct { // bgNotice 是一条尚未投递给父会话的后台子代理结果。 type bgNotice struct { + binding taskBinding id string // 持久化队列主键（重启重投凭据） owner string name string @@ -297,6 +299,8 @@ type runCtx struct { // 材料整理（附件即保证）：预启动说明注入第 0 回合；ensureMaterial 兜底幂等。 materialNote string materialEnsured bool + task taskBinding + sources materialSourceRegistry } // Surfaces 返回当前会话最近保存的生成页面（U3/R6：ctxasm 语义投影回灌； @@ -590,6 +594,15 @@ func (c *runCtx) Memories() []core.MemoryEntry { } func (c *runCtx) Emit(typ core.StreamEventType, payload any) error { + if c.task.cardID != "" && (typ == core.EvToolCall || typ == core.EvToolResult) { + label := "正在读取和核对资料" + if typ == core.EvToolResult { + label = "已取得工具结果，继续整理" + } + if e := c.mgr.updateTaskActivity(c.task, core.TaskPhase{ID: "tool", Label: label, ObservedAt: time.Now().UTC()}); e != nil && !errors.Is(e, errActivityThrottled) { + log.Printf("[runmgr] task activity: %v", e) + } + } _, err := c.mgr.emit(c.run.ConversationID, c.run.ID, typ, payload) return err } @@ -1829,6 +1842,10 @@ type bgSpawnOpts struct { taskKey string // 内容键（非空 = 派发前去重、完成后保留键于卡片） material bool // 材料整理任务（通知回合衔接交付物选择） queue bool // 满容量时可排队（系统任务；模型任务仍硬拒绝） + documents []core.ParsedDocument + sourceMessageID string + binding taskBinding + stopLease func() } func (m *Manager) spawnBackground(parent core.Run, agentID, name, task, hint string) (core.ID, error) { @@ -1839,44 +1856,45 @@ func (m *Manager) spawnBackgroundOpts(parent core.Run, o bgSpawnOpts) (core.ID, if parent.Kind == core.RunSubagent { return "", fmt.Errorf("子代理不能再派发子代理") } - if o.taskKey != "" { - convID := string(parent.ConversationID) - m.materialMu.Lock() - active := m.materialActive != nil && m.materialActive[convID][o.taskKey] - m.materialMu.Unlock() - if active { - return "", fmt.Errorf("同键后台任务已在运行（材料键 %s），见收件箱进度卡", truncate(o.taskKey, 12)) - } - } conv := parent.ConversationID m.noticeMu.Lock() if m.bgInFlight == nil { m.bgInFlight = map[string]int{} } - if m.bgInFlight[conv] >= maxBackgroundChildren { - if !o.queue { + if m.bgInFlight[conv] >= maxBackgroundChildren && !o.queue { + m.noticeMu.Unlock() + return "", fmt.Errorf("本会话已有 %d 个后台子代理在运行", maxBackgroundChildren) + } + if o.taskKey != "" { + m.materialMu.Lock() + active := m.materialActive[conv][o.taskKey] + m.materialMu.Unlock() + if active { m.noticeMu.Unlock() - return "", fmt.Errorf("本会话已有 %d 个后台子代理在运行", maxBackgroundChildren) + return "", fmt.Errorf("同键后台任务已登记") } + } + binding, card, err := m.registerBackgroundTask(parent, o) + if err != nil { + m.noticeMu.Unlock() + return "", err + } + o.binding = binding + o.stopLease = m.maintainTaskLease(binding) + if o.taskKey != "" { + m.materialKeyRegister(conv, o.taskKey) + } + if m.bgInFlight[conv] >= maxBackgroundChildren { 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 - } + m.bgQueue[conv] = append(m.bgQueue[conv], bgQueueItem{parent: parent, opts: o, card: card}) m.noticeMu.Unlock() return "", nil } m.bgInFlight[conv]++ m.noticeMu.Unlock() - return m.startBackgroundSlot(parent, o, core.ActivityCard{}) + return m.startBackgroundSlot(parent, o, card) } // startBackgroundSlot 持槽启动后台任务（调用方已占 bgInFlight）。 @@ -1884,23 +1902,37 @@ func (m *Manager) spawnBackgroundOpts(parent core.Run, o bgSpawnOpts) (core.ID, func (m *Manager) startBackgroundSlot(parent core.Run, o bgSpawnOpts, preCard core.ActivityCard) (core.ID, error) { rc, runner, err := m.startChildOpts(context.Background(), parent, o, true) if err != nil { + _ = m.finishTaskProjection(o.binding, core.AgentResultPayload{}, err) + if o.stopLease != nil { + o.stopLease() + } + m.materialKeyRelease(parent.ConversationID, o.taskKey) m.decBackground(parent.ConversationID) return "", err } - if o.taskKey != "" { - m.materialKeyRegister(string(parent.ConversationID), o.taskKey) - } - // 慢通道可观测（任务快慢双速，SR-B10）：后台子代理进收件箱进度卡， - // 完成后把结果回投该卡；通知回合路径保持不变，卡片是用户侧回收入口。 bgCard := preCard - if bgCard.ID == "" { - bgCard = m.createBackgroundCard(parent, rc.run.ID, o.name, o.task, o.kv, "") - } else { - m.promoteQueuedCard(parent.Owner, preCard, rc.run.ID) + rc.task = o.binding + rc.attachments = o.documents + rc.sources = materialSources(o.documents, int(rc.eff.GetNumber(materialPerDocKey, materialPerDocDef))) + if err = m.startTaskProjection(o.binding, rc.run.ID); err != nil { + rc.h.cancel() + m.setRunStatus(&rc.run, core.RunFailed, rc.run.Phase) + m.mu.Lock() + delete(m.active, rc.run.ID) + m.mu.Unlock() + if o.stopLease != nil { + o.stopLease() + } + m.materialKeyRelease(parent.ConversationID, o.taskKey) + m.decBackground(parent.ConversationID) + return "", err } kv := o.kv go func() { defer m.decBackground(parent.ConversationID) + if o.stopLease != nil { + defer o.stopLease() + } if o.taskKey != "" { defer m.materialKeyRelease(string(parent.ConversationID), o.taskKey) } @@ -1916,7 +1948,7 @@ func (m *Manager) startBackgroundSlot(parent core.Run, o bgSpawnOpts, preCard co defer m.adm.release(m.st, rc.run.ID) payload, err := m.finishChild(rc, runner, false) m.finishBackgroundCard(parent.Owner, bgCard, payload, err, kv) - m.enqueueNotice(parent.ConversationID, bgNotice{owner: parent.Owner, name: o.name, payload: payload, material: o.material}) + m.enqueueNotice(parent.ConversationID, bgNotice{owner: parent.Owner, name: o.name, payload: payload, material: o.material, binding: o.binding}) m.deliverNotices(parent.ConversationID) }() return rc.run.ID, nil @@ -1999,30 +2031,13 @@ func (m *Manager) finishBackgroundCard(owner string, card core.ActivityCard, pay if m.cards == nil || card.ID == "" { return } - summary := strings.TrimSpace(payload.Summary) - tone := "ok" - status := core.CardDone - statusLabel := core.TaskSummaryStatusDone - if err != nil { - tone = "danger" - status = core.CardFailed - statusLabel = core.TaskSummaryStatusFailed - if summary == "" { - summary = err.Error() + if card.Task != nil { + if e := m.finishTaskProjection(bindingFrom(card, owner), payload, err); e != nil { + log.Printf("[runmgr] task finish: %v", e) } - summary = "执行失败：" + summary - } - if summary == "" { - summary = "（无结果摘要）" - } - extra := append(append([][2]string(nil), kv...), [2]string{"通道", "慢通道 · 后台执行"}) - card.Body = (core.TaskSummaryView{OutcomeLine: summary}).FinishedBody(statusLabel, extra) - if uerr := m.cards.Update(owner, card); uerr != nil { - log.Printf("[runmgr] background card update: %v", uerr) - } - if _, serr := m.cards.SetStatus(owner, card.ID, status, tone, ""); serr != nil { - log.Printf("[runmgr] background card status: %v", serr) + return } + // Legacy cards remain readable; new background tasks always have a Task. } func (m *Manager) decBackground(convID string) { @@ -2042,10 +2057,14 @@ func (m *Manager) decBackground(convID string) { func (m *Manager) enqueueNotice(convID string, n bgNotice) { // 先落库（重启重投凭据），再进内存热队列；落库失败不丢内存路径。 if n.id == "" { - n.id = store.NewID("ntc") + if n.binding.attemptID != "" { + n.id = "ntc_" + n.binding.attemptID + } else { + n.id = store.NewID("ntc") + } } if err := m.st.AppendNotice(store.Notice{ - ID: n.id, ConversationID: convID, Owner: n.owner, Name: n.name, Payload: n.payload, + ID: n.id, ConversationID: convID, Owner: n.owner, Name: n.name, Payload: n.payload, TaskID: n.binding.taskID, AttemptID: n.binding.attemptID, CardID: n.binding.cardID, Material: n.material, }); err != nil { log.Printf("[runmgr] notice persist %s: %v", convID, err) } @@ -2071,8 +2090,20 @@ func (m *Manager) LoadPendingNotices() { m.notices = map[string][]bgNotice{} } for _, it := range items { + b := taskBinding{} + if it.CardID != "" { + card, e := m.st.GetCard(it.Owner, it.CardID) + if e != nil || card.Task == nil || card.Task.TaskID != it.TaskID || card.Task.AttemptID != it.AttemptID { + continue + } + if card.Task.DeliveryState == "reported" { + _ = m.st.DeleteNotice(it.ID) + continue + } + b = bindingFrom(card, it.Owner) + } m.notices[it.ConversationID] = append(m.notices[it.ConversationID], bgNotice{ - id: it.ID, owner: it.Owner, name: it.Name, payload: it.Payload, + id: it.ID, owner: it.Owner, name: it.Name, payload: it.Payload, material: it.Material, binding: b, }) byConv[it.ConversationID] = true } @@ -2130,6 +2161,20 @@ func (m *Manager) deliverNotices(convID string) { m.noticeMu.Unlock() return } + // Delivery is independent of execution completion; a readable result survives notification failure. + for _, n := range items { + if n.binding.cardID != "" { + if e := m.mutateTask(n.binding, func(t *core.TaskSnapshot) error { + if !t.Terminal() { + return store.ErrConflict + } + t.DeliveryState = "reported" + return nil + }); e != nil { + log.Printf("[runmgr] task notice state: %v", e) + } + } + } // 投递成功才删持久化行（失败保留下次重投）。 ids := make([]string, 0, len(items)) for _, n := range items { @@ -2192,7 +2237,16 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error { ConversationID: convID, Role: core.RoleSystem, Parts: []core.Part{{Type: core.PartText, Text: b.String()}}, } - if err := m.st.AppendMessage(&msg); err != nil { + noticeIDs := make([]string, 0, len(items)) + for _, n := range items { + noticeIDs = append(noticeIDs, n.id) + } + sort.Strings(noticeIDs) + if err := m.st.AppendMessageWithRequest(&msg, "background-notices:"+strings.Join(noticeIDs, ",")); err != nil { + if errors.Is(err, store.ErrDuplicateRequest) { + m.releaseConversation(convID) + return nil + } m.releaseConversation(convID) return err } @@ -2251,6 +2305,9 @@ func (m *Manager) ReconcileInterrupted() error { for _, r := range runs { m.sealRunInterrupted(r.ID) } + if m.cards != nil { + return m.reconcileTaskProjections() + } return nil } diff --git a/internal/runmgr/task_projection.go b/internal/runmgr/task_projection.go new file mode 100644 index 0000000..bb897a9 --- /dev/null +++ b/internal/runmgr/task_projection.go @@ -0,0 +1,298 @@ +package runmgr + +import ( + "agent-system-backend/internal/store" + core "agent-system-core" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "log" + "strings" + "time" +) + +type taskBinding struct { + taskID, attemptID, cardID, owner, projectID, conversationID, parentRunID, instanceID string + epoch int64 +} + +func (m *Manager) taskInstanceID() string { + if m.instanceID != "" { + return m.instanceID + } + return m.InstanceID() +} +func taskLeaseKey(id string) string { return "task:" + id } +func bindingFrom(card core.ActivityCard, owner string) taskBinding { + s := card.Task + if s == nil { + return taskBinding{} + } + return taskBinding{s.TaskID, s.AttemptID, s.CardID, owner, s.ProjectID, s.ConversationID, s.ParentRunID, s.OwnerInstanceID, s.ExecutionEpoch} +} +func (m *Manager) registerBackgroundTask(parent core.Run, o bgSpawnOpts) (taskBinding, core.ActivityCard, error) { + if m.cards == nil { + return taskBinding{}, core.ActivityCard{}, nil + } + now := time.Now().UTC() + title := strings.TrimSpace(o.name) + if title == "" { + title = "后台任务" + } + s := core.TaskSnapshot{SchemaVersion: 1, TaskID: store.NewID("task"), AttemptID: store.NewID("attempt"), CardID: store.NewID("card"), ProjectID: parent.ProjectID, ConversationID: parent.ConversationID, SourceMessageID: o.sourceMessageID, ParentRunID: parent.ID, OwnerInstanceID: m.taskInstanceID(), Revision: 1, StreamScope: "conversation:" + parent.ConversationID, InputRevision: o.taskKey, Title: title, Goal: "完成" + title, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Phase: core.TaskPhase{ID: "queued", Label: "等待执行资源", ObservedAt: now}, Focus: "等待执行资源", Remaining: "尚未开始处理", NextStep: "获得执行资源后开始", LastActivityAt: now, LastMeaningfulUpdateAt: now} + if o.material { + s.Goal = "整理材料中的要点、差异和缺项" + s.Coverage = materialCoverage(o.documents) + for _, d := range o.documents { + if d.Meta.Origin != nil { + s.Inputs = append(s.Inputs, core.TaskSourceRef{ArtifactID: d.Meta.Origin.ArtifactID, ContentHash: d.Meta.Origin.ContentHash, ParserRevision: d.Meta.Origin.ParserRevision, Label: d.Meta.Filename, Availability: "available"}) + } + } + } + epoch, err := m.st.TryClaimExecution(taskLeaseKey(s.TaskID), s.OwnerInstanceID, "", 60) + if err != nil { + return taskBinding{}, core.ActivityCard{}, err + } + s.ExecutionEpoch = epoch + card, err := m.cards.CreateTask(parent.Owner, core.ActivityCard{ID: s.CardID, ProjectID: parent.ProjectID, ConversationID: parent.ConversationID, AnchorMessageID: o.sourceMessageID, Kind: core.CardKindProgress, Title: title, Task: &s}, "") + if err != nil { + m.st.ReleaseExecution(taskLeaseKey(s.TaskID), s.OwnerInstanceID, epoch) + return taskBinding{}, card, err + } + return bindingFrom(card, parent.Owner), card, nil +} +func materialCoverage(docs []core.ParsedDocument) string { + usable, warnings := 0, 0 + for _, d := range docs { + if strings.TrimSpace(d.PlainText()) != "" { + usable++ + } + warnings += len(d.Warnings) + } + text := fmt.Sprintf("已提供 %d 份材料中的可用片段；%d 份有可读取文本。尚不表示已完整分析。", len(docs), usable) + if warnings > 0 { + text += fmt.Sprintf("解析有 %d 项限制，请结合原件核验。", warnings) + } + return text +} +func (m *Manager) mutateTask(b taskBinding, mutate func(*core.TaskSnapshot) error) error { + if b.cardID == "" || m.cards == nil { + return nil + } + for n := 0; n < 3; n++ { + card, e := m.st.GetCard(b.owner, b.cardID) + if e != nil { + return e + } + if card.Task == nil || card.Task.TaskID != b.taskID || card.Task.AttemptID != b.attemptID || card.Task.OwnerInstanceID != b.instanceID || card.Task.ExecutionEpoch != b.epoch { + return store.ErrConflict + } + next := *card.Task + if e = mutate(&next); e != nil { + return e + } + next.Revision++ + e = m.cards.CommitTask(b.owner, next, card.Task.Revision) + if !errors.Is(e, store.ErrConflict) { + return e + } + } + return store.ErrConflict +} +func (m *Manager) updateTaskActivity(b taskBinding, phase core.TaskPhase) error { + return m.mutateTask(b, func(s *core.TaskSnapshot) error { + if s.Terminal() { + return store.ErrConflict + } + now := time.Now().UTC() + if phase.ID != "started" && now.Sub(s.LastActivityAt) < time.Second { + return errActivityThrottled + } + s.Phase = phase + s.LastActivityAt = now + s.Focus = phase.Label + return nil + }) +} + +var errActivityThrottled = errors.New("task activity throttled") + +func (m *Manager) startTaskProjection(b taskBinding, runID string) error { + return m.mutateTask(b, func(s *core.TaskSnapshot) error { + if s.Terminal() || s.RunID != "" { + return store.ErrConflict + } + s.RunID = runID + s.ExecutionState = "running" + s.Phase = core.TaskPhase{ID: "started", Label: "处理已登记工作", ObservedAt: time.Now().UTC()} + s.Focus = "处理已登记工作" + s.Remaining = "待任务进展后更新" + s.NextStep = "继续处理已登记工作" + s.LastActivityAt = time.Now().UTC() + return nil + }) +} +func (m *Manager) finishTaskProjection(b taskBinding, payload core.AgentResultPayload, runErr error) error { + return m.mutateTask(b, func(s *core.TaskSnapshot) error { + if s.Terminal() { + return store.ErrConflict + } + s.ExecutionState = "completed" + s.OutcomeCompleteness = "complete" + s.DeliveryState = "available" + s.Conclusion = strings.TrimSpace(payload.Summary) + if runErr != nil { + s.ExecutionState = "failed" + s.OutcomeCompleteness = "none" + if len(s.Findings) > 0 { + s.OutcomeCompleteness = "partial" + } + s.Conclusion = "执行失败：" + runErr.Error() + } + if s.Conclusion == "" { + s.Conclusion = "没有可用结果摘要" + } + if s.RunID != "" { + messages, e := m.st.ListMessages(s.ConversationID) + if e != nil { + return e + } + seen := map[string]bool{} + for _, msg := range messages { + if msg.RunID != s.RunID { + continue + } + for _, p := range msg.Parts { + if p.Artifact == nil || seen[p.Artifact.ArtifactID] { + continue + } + seen[p.Artifact.ArtifactID] = true + a, e := m.st.GetArtifact(b.owner, p.Artifact.ArtifactID) + if e != nil || a.ProjectID != s.ProjectID { + continue + } + data, e := m.st.ArtifactContent(b.owner, a.ID) + if e != nil { + continue + } + h := sha256.Sum256(data) + s.ResultRefs = append(s.ResultRefs, core.TaskResultRef{ArtifactID: a.ID, Name: a.Name, ContentHash: hex.EncodeToString(h[:]), Availability: "available"}) + } + } + } + s.Phase = core.TaskPhase{ID: s.ExecutionState, Label: "处理结束", ObservedAt: time.Now().UTC()} + s.Focus = "处理结束" + s.LastActivityAt = time.Now().UTC() + s.LastMeaningfulUpdateAt = s.LastActivityAt + return nil + }) +} +func (m *Manager) maintainTaskLease(b taskBinding) func() { + if b.cardID == "" { + return func() {} + } + stop := make(chan struct{}) + go func() { + ticker := time.NewTicker(20 * time.Second) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + if !m.st.RenewExecution(taskLeaseKey(b.taskID), b.instanceID, b.epoch, 60) { + log.Printf("[runmgr] task execution fenced: %s", b.taskID) + return + } + } + } + }() + return func() { close(stop); m.st.ReleaseExecution(taskLeaseKey(b.taskID), b.instanceID, b.epoch) } +} +func shouldInterruptQueued(ownerInstanceID, currentInstanceID string, hasLiveExecutionWitness bool) bool { + return ownerInstanceID == currentInstanceID && !hasLiveExecutionWitness +} +func (m *Manager) reconcileTaskProjections() error { + records, err := m.st.TaskCards() + if err != nil { + return err + } + for _, record := range records { + s := record.Card.Task + if s == nil || s.Terminal() { + continue + } + if m.st.Dialect() == "postgres" && s.OwnerInstanceID != m.taskInstanceID() { + continue + } + // Reading execution authority must succeed, even for an orphan queue. + es, e := m.st.GetExecution(taskLeaseKey(s.TaskID)) + if e != nil && !errors.Is(e, store.ErrNotFound) { + return e + } + if e == nil && es.InstanceID != s.OwnerInstanceID { + return store.ErrFenced + } + next := *s + epoch, e := m.st.TryClaimExecution(taskLeaseKey(s.TaskID), s.OwnerInstanceID, s.RunID, 60) + if e != nil { + return e + } + next.ExecutionEpoch = epoch + next.Revision++ + next.ExecutionState = "interrupted" + next.OutcomeCompleteness = "none" + if len(next.Findings) > 0 { + next.OutcomeCompleteness = "partial" + next.DeliveryState = "available" + } + next.Conclusion = "进程重启或停机，任务已中断" + if s.RunID == "" { + next.Conclusion = "排队任务未恢复，未开始执行" + } + next.Focus = "任务已中断" + next.Phase = core.TaskPhase{ID: "interrupted", Label: "任务已中断", ObservedAt: time.Now().UTC()} + next.LastActivityAt = time.Now().UTC() + if e = m.cards.CommitTask(record.Owner, next, s.Revision); e != nil { + return e + } + m.st.ReleaseExecution(taskLeaseKey(s.TaskID), s.OwnerInstanceID, epoch) + } + return nil +} +func canonicalFindingID(attempt, key string) string { + sum := sha256.Sum256([]byte(attempt + ":" + key)) + return "finding_" + hex.EncodeToString(sum[:12]) +} + +func (m *Manager) taskSourcesAvailable(owner, project string, t *core.TaskSnapshot) bool { + count := 0 + for _, src := range t.Inputs { + count++ + if !m.taskSourceAvailable(owner, project, src) { + return false + } + } + for _, f := range t.Findings { + for _, src := range f.Sources { + count++ + if !m.taskSourceAvailable(owner, project, src) { + return false + } + } + } + return count > 0 +} +func (m *Manager) taskSourceAvailable(owner, project string, src core.TaskSourceRef) bool { + a, e := m.st.GetArtifact(owner, src.ArtifactID) + if e != nil || a.ProjectID != project { + return false + } + data, e := m.st.ArtifactContent(owner, src.ArtifactID) + if e != nil { + return false + } + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) == src.ContentHash +} diff --git a/internal/runmgr/task_projection_test.go b/internal/runmgr/task_projection_test.go new file mode 100644 index 0000000..1180384 --- /dev/null +++ b/internal/runmgr/task_projection_test.go @@ -0,0 +1,97 @@ +package runmgr + +import ( + "agent-system-backend/internal/cardsvc" + core "agent-system-core" + "testing" +) + +func TestTaskQueueKeepsIdentityWhenPromoted(t *testing.T) { + runner := &holdRunner{release: make(chan struct{})} + c, _ := newCardSpawnEnv(t, runner) + c.mgr.runner = noopRunner{} + for i := 0; i < 4; i++ { + if _, e := c.mgr.spawnBackgroundOpts(c.run, bgSpawnOpts{name: "整理", task: "读材料", queue: true}); e != nil { + t.Fatal(e) + } + } + cards, e := c.mgr.cards.List(c.run.Owner, c.run.ConversationID) + if e != nil || len(cards) != 4 { + t.Fatal(cards, e) + } + var queued core.ActivityCard + for _, card := range cards { + if card.Task == nil { + t.Fatal("background work has no canonical task") + } + if card.Task.ExecutionState == "queued" { + queued = card + } + } + if queued.ID == "" || queued.Task.RunID != "" { + t.Fatal("queue did not precede execution") + } + close(runner.release) + waitUntil(t, func() bool { + got, e := c.mgr.st.GetCard(c.run.Owner, queued.ID) + return e == nil && got.Task != nil && got.Task.ExecutionState == "completed" + }) + got, _ := c.mgr.st.GetCard(c.run.Owner, queued.ID) + if got.Task.TaskID != queued.Task.TaskID || got.Task.AttemptID != queued.Task.AttemptID || got.Task.RunID == "" { + t.Fatal("promotion replaced identity") + } +} +func TestTaskReconcileOwnedQueueAndForeignLease(t *testing.T) { + st := openRecoveryStore(t) + m := newRecoveryManager(t, st) + m.SetInstanceID("A") + m.SetCards(cardsvc.New(st, nil)) + conv := recoveryConversation(t, st) + parent := recoveryRun(t, st, conv, core.RunRunning) + _, own, e := m.registerBackgroundTask(parent, bgSpawnOpts{name: "自己的队列"}) + if e != nil { + t.Fatal(e) + } + other := newRecoveryManager(t, st) + other.SetInstanceID("B") + other.SetCards(cardsvc.New(st, nil)) + _, foreign, e := other.registerBackgroundTask(parent, bgSpawnOpts{name: "其他实例"}) + if e != nil { + t.Fatal(e) + } + if e = m.ReconcileInterrupted(); e != nil { + t.Fatal(e) + } + got, _ := st.GetCard(conv.Owner, own.ID) + if got.Task.ExecutionState != "interrupted" { + t.Fatal("orphan queue retained promise", got.Task) + } + got, _ = st.GetCard(conv.Owner, foreign.ID) + if st.Dialect() == "postgres" && got.Task.ExecutionState != "queued" { + t.Fatal("foreign lease was sealed") + } +} + +func TestTaskProjectionStaleExecutionCannotOverwrite(t *testing.T) { + st := openRecoveryStore(t) + m := newRecoveryManager(t, st) + m.SetInstanceID("A") + m.SetCards(cardsvc.New(st, nil)) + conv := recoveryConversation(t, st) + parent := recoveryRun(t, st, conv, core.RunRunning) + binding, card, e := m.registerBackgroundTask(parent, bgSpawnOpts{name: "整理"}) + if e != nil { + t.Fatal(e) + } + st.ReleaseExecution(taskLeaseKey(binding.taskID), binding.instanceID, binding.epoch) + if _, e = st.TryClaimExecution(taskLeaseKey(binding.taskID), "B", "", 60); e != nil { + t.Fatal(e) + } + if e = m.startTaskProjection(binding, "old-run"); e == nil { + t.Fatal("old executor overwrote new lease") + } + got, _ := st.GetCard(conv.Owner, card.ID) + if got.Task.RunID != "" || got.Task.Revision != 1 { + t.Fatal("fenced write leaked", got.Task) + } +} diff --git a/internal/store/notices.go b/internal/store/notices.go index 72e5327..e19b57e 100644 --- a/internal/store/notices.go +++ b/internal/store/notices.go @@ -15,13 +15,25 @@ type Notice struct { Owner string Name string Payload core.AgentResultPayload + TaskID string `json:"taskId,omitempty"` + AttemptID string `json:"attemptId,omitempty"` + CardID string `json:"cardId,omitempty"` + Material bool `json:"material,omitempty"` CreatedAt time.Time } // AppendNotice 落一条待投递通知（重启后可重投）。 +type noticePayload struct { + core.AgentResultPayload + TaskID string `json:"taskId,omitempty"` + AttemptID string `json:"attemptId,omitempty"` + CardID string `json:"cardId,omitempty"` + Material bool `json:"material,omitempty"` +} + func (s *Store) AppendNotice(n Notice) error { - _, err := s.db.Exec(s.q(`INSERT INTO pending_notices (id,conversation_id,owner,name,payload,created_at) VALUES(?,?,?,?,?,?)`), - n.ID, n.ConversationID, n.Owner, n.Name, marshalJSON(&n.Payload), nowISO()) + _, err := s.db.Exec(s.q(`INSERT INTO pending_notices (id,conversation_id,owner,name,payload,created_at) VALUES(?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING`), + n.ID, n.ConversationID, n.Owner, n.Name, marshalJSON(¬icePayload{n.Payload, n.TaskID, n.AttemptID, n.CardID, n.Material}), nowISO()) return err } @@ -39,7 +51,15 @@ func (s *Store) ListNotices() ([]Notice, error) { if err := rows.Scan(&n.ID, &n.ConversationID, &n.Owner, &n.Name, &payload, &ca); err != nil { return nil, err } - n.Payload, _ = unmarshalJSON[core.AgentResultPayload](payload) + p, err := unmarshalJSON[noticePayload](payload) + if err != nil { + return nil, err + } + n.Payload = p.AgentResultPayload + n.TaskID = p.TaskID + n.AttemptID = p.AttemptID + n.CardID = p.CardID + n.Material = p.Material n.CreatedAt, _ = parseTime(ca) out = append(out, n) } diff --git a/internal/store/task_projection.go b/internal/store/task_projection.go index aba0a00..a5e518d 100644 --- a/internal/store/task_projection.go +++ b/internal/store/task_projection.go @@ -109,6 +109,19 @@ func (s *Store) CommitTaskProjection(w TaskProjectionWrite) ([]core.StreamEvent, } } var events []core.StreamEvent + if !viewOnly && task.ExecutionEpoch > 0 { + if err = s.lockAdmission(tx); err != nil { + return nil, err + } + var instance, until string + var epoch int64 + if err = tx.QueryRow(s.q(`SELECT instance_id,epoch,lease_until FROM conversation_executions WHERE conversation_id=?`), "task:"+task.TaskID).Scan(&instance, &epoch, &until); err != nil { + return nil, err + } + if instance != task.OwnerInstanceID || epoch != task.ExecutionEpoch || (!task.Terminal() && until <= nowISO()) { + return nil, ErrFenced + } + } for _, ch := range channels { var batch []core.StreamEvent for _, e := range w.Buffered { @@ -224,3 +237,31 @@ func (s *Store) CardsAtCursor(owner, conversationID string) (CardsSnapshot, erro } return out, nil } + +type TaskCardRecord struct { + Owner string + Card core.ActivityCard +} + +func (s *Store) TaskCards() ([]TaskCardRecord, error) { + rows, err := s.db.Query(`SELECT owner,payload FROM activity_cards`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []TaskCardRecord + for rows.Next() { + var r TaskCardRecord + var raw string + if err = rows.Scan(&r.Owner, &raw); err != nil { + return nil, err + } + if err = json.Unmarshal([]byte(raw), &r.Card); err != nil { + return nil, err + } + if r.Card.Task != nil { + out = append(out, r) + } + } + return out, rows.Err() +}