413 lines
14 KiB
Go
413 lines
14 KiB
Go
package runmgr
|
||
|
||
import (
|
||
"agent-system-backend/internal/store"
|
||
core "agent-system-core"
|
||
"context"
|
||
"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)
|
||
left := materialInlineTotal
|
||
for _, d := range o.documents {
|
||
chars := len([]rune(d.PlainText()))
|
||
if chars == 0 {
|
||
s.Limitations = append(s.Limitations, d.Meta.Filename+"：未读取到可用文本")
|
||
}
|
||
if len(d.Warnings) > 0 {
|
||
s.Limitations = append(s.Limitations, d.Meta.Filename+"：存在解析限制，请结合原件核验")
|
||
}
|
||
perDoc := int(m.settings.Effective(parent.Owner, parent.ProjectID).GetNumber(materialPerDocKey, materialPerDocDef))
|
||
if perDoc <= 0 {
|
||
perDoc = materialPerDocDef
|
||
}
|
||
cap := perDoc
|
||
if left < cap {
|
||
cap = left
|
||
}
|
||
if chars > cap {
|
||
s.Limitations = append(s.Limitations, d.Meta.Filename+"：仅提供部分文字，请勿据此认定全文已分析")
|
||
}
|
||
if chars < cap {
|
||
cap = chars
|
||
}
|
||
left -= cap
|
||
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, Work: &core.WorkCard{}}, "")
|
||
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"
|
||
if len(s.Limitations) > 0 || s.Remaining != "" && s.Remaining != "待任务进展后更新" {
|
||
s.OutcomeCompleteness = "partial"
|
||
}
|
||
s.DeliveryState = "available"
|
||
s.Conclusion = taskResultConclusion(payload)
|
||
s.NextStep = "核验已有结果和未决事项，再选择后续工作"
|
||
s.CheckpointKeys = nil
|
||
if runErr != nil {
|
||
s.ExecutionState = "failed"
|
||
if errors.Is(runErr, context.Canceled) {
|
||
s.ExecutionState = "interrupted"
|
||
}
|
||
s.OutcomeCompleteness = "none"
|
||
if len(s.Findings) > 0 {
|
||
s.OutcomeCompleteness = "partial"
|
||
}
|
||
hasFindings := false
|
||
for _, finding := range s.Findings {
|
||
if finding.Status == "active" {
|
||
hasFindings = true
|
||
break
|
||
}
|
||
}
|
||
s.NextStep = "核对已有发现和材料；需要继续处理时重新提出请求"
|
||
s.Remaining = "剩余工作尚未完成，已保留的发现需结合原件核验"
|
||
s.Conclusion = "执行失败，任务已停止；请核验已有发现和材料限制"
|
||
if strings.Contains(runErr.Error(), "budget_exceeded") {
|
||
s.Conclusion = "达到本次处理额度，执行已停止；已有发现已保留"
|
||
}
|
||
if !hasFindings {
|
||
s.OutcomeCompleteness = "none"
|
||
s.Remaining = "尚未形成可用阶段成果"
|
||
s.NextStep = "核对材料和处理范围；需要继续时重新提出请求"
|
||
s.Conclusion = "执行失败，任务已停止；尚无可用阶段发现"
|
||
if strings.Contains(runErr.Error(), "budget_exceeded") {
|
||
s.Conclusion = "达到本次处理额度，执行已停止；尚无可用阶段发现"
|
||
}
|
||
}
|
||
if strings.Contains(runErr.Error(), resultPersistenceFailedSummary) {
|
||
s.Conclusion = resultPersistenceFailedSummary
|
||
}
|
||
}
|
||
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"})
|
||
}
|
||
}
|
||
}
|
||
label := "处理结束"
|
||
if s.ExecutionState == "failed" {
|
||
label = "执行已停止"
|
||
}
|
||
if s.ExecutionState == "interrupted" {
|
||
label = "任务已中断"
|
||
}
|
||
s.Phase = core.TaskPhase{ID: s.ExecutionState, Label: label, ObservedAt: time.Now().UTC()}
|
||
s.Focus = label
|
||
s.LastActivityAt = time.Now().UTC()
|
||
s.LastMeaningfulUpdateAt = s.LastActivityAt
|
||
if e := s.Validate(); e != nil {
|
||
s.Conclusion = "任务已结束，已有发现与限制见本卡；完整结果保留在执行记录中"
|
||
s.Focus = ""
|
||
s.NextStep = ""
|
||
s.ResultRefs = nil
|
||
s.Phase.Label = ""
|
||
if e = s.Validate(); e != nil {
|
||
return e
|
||
}
|
||
}
|
||
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.NextStep = "核对已有发现和材料；需要继续处理时重新提出请求"
|
||
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
|
||
}
|
||
|
||
func taskResultConclusion(payload core.AgentResultPayload) string {
|
||
text := strings.TrimSpace(payload.Text)
|
||
if text == "" {
|
||
return strings.TrimSpace(payload.Summary)
|
||
}
|
||
for _, p := range strings.Split(text, "\n\n") {
|
||
p = strings.TrimSpace(p)
|
||
if p == "" || strings.HasPrefix(p, "#") {
|
||
continue
|
||
}
|
||
if len([]rune(p)) <= 1200 {
|
||
return p
|
||
}
|
||
for _, mark := range []string{"。", "！", "？"} {
|
||
if i := strings.Index(p, mark); i >= 0 && len([]rune(p[:i+len(mark)])) <= 1200 {
|
||
return p[:i+len(mark)]
|
||
}
|
||
}
|
||
break
|
||
}
|
||
return "整理已结束，请结合下面的阶段发现、范围与限制核验结果"
|
||
}
|
||
func (m *Manager) safeTaskNotice(owner string, card core.ActivityCard, payload core.AgentResultPayload) core.AgentResultPayload {
|
||
if card.Task == nil {
|
||
return payload
|
||
}
|
||
visible, e := m.cards.Get(owner, card.ID)
|
||
if e != nil || visible.Task == nil {
|
||
payload.Summary = "任务已结束，当前结果暂不可读取，请核实材料权限"
|
||
payload.Text = ""
|
||
return payload
|
||
}
|
||
payload.Summary = visible.Task.Conclusion
|
||
if visible.Task.OutcomeCompleteness == "partial" {
|
||
payload.Summary += "\n当前成果仍有范围或未决事项，请核对任务卡中的限制。"
|
||
}
|
||
payload.Text = ""
|
||
return payload
|
||
}
|