Agent-System-Backend/internal/runmgr/task_checkpoint.go

package runmgr

import (
	"agent-system-backend/internal/store"
	core "agent-system-core"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strings"
	"time"
)

func (c *runCtx) PublishWorkCheckpoint(candidate core.WorkCheckpointCandidate) error {
	if c.run.Kind != core.RunSubagent || c.subAgent != materialSubagentID || c.task.cardID == "" {
		return fmt.Errorf("checkpoint.scope: material task required")
	}
	if !c.prof.Tools.ToolAllowed("publish_work_checkpoint") || c.prof.Tools.NeedsConfirm("publish_work_checkpoint") {
		return fmt.Errorf("checkpoint.policy: denied")
	}
	if err := core.ValidateWorkCheckpoint(candidate); err != nil {
		return err
	}
	raw, _ := json.Marshal(candidate)
	sum := sha256.Sum256(raw)
	digest := hex.EncodeToString(sum[:])
	witnesses := map[string]store.ArtifactWitness{}
	findings := make([]core.TaskFinding, 0, len(candidate.Findings))
	critical := false
	for _, f := range candidate.Findings {
		next := core.TaskFinding{ID: canonicalFindingID(c.task.attemptID, f.FindingKey), Revision: 1, Text: f.Text, Impact: f.Impact, Kind: f.Kind, Status: "active", Assumptions: f.Assumptions}
		if f.Corrects != "" {
			critical = true
			next.Corrects = canonicalFindingID(c.task.attemptID, f.Corrects)
		}
		for _, q := range f.Sources {
			src, err := c.sources.validate(q)
			if err != nil {
				return err
			}
			a, err := c.mgr.st.GetArtifact(c.run.Owner, src.ArtifactID)
			if err != nil {
				return fmt.Errorf("checkpoint.source: unavailable")
			}
			if !c.mgr.taskSourceAvailable(c.run.Owner, c.run.ProjectID, src) {
				return fmt.Errorf("checkpoint.source: unavailable or changed")
			}
			after, err := c.mgr.st.GetArtifact(c.run.Owner, src.ArtifactID)
			if err != nil || !after.UpdatedAt.Equal(a.UpdatedAt) {
				return store.ErrConflict
			}
			witnesses[src.ArtifactID] = store.ArtifactWitness{ArtifactID: src.ArtifactID, UpdatedAt: a.UpdatedAt.Format(time.RFC3339Nano)}
			next.Sources = append(next.Sources, src)
		}
		if f.Kind == "material_record" {
			// A quotation proves what the source says, not an unstated approval
			// or model paraphrase. Keep interpretations in their own finding kinds.
			quotes := make([]string, 0, len(next.Sources))
			seen := map[string]bool{}
			for _, src := range next.Sources {
				line := src.Label + "：「" + src.Quote + "」"
				if !seen[line] {
					quotes = append(quotes, line)
					seen[line] = true
				}
			}
			next.Text = strings.Join(quotes, "\n")
			next.Impact, next.Assumptions = "", ""
		}
		findings = append(findings, next)
	}
	card, err := c.mgr.st.GetCard(c.run.Owner, c.task.cardID)
	if err != nil {
		return err
	}
	s := card.Task
	if s == nil || s.Terminal() || s.TaskID != c.task.taskID || s.AttemptID != c.task.attemptID || s.RunID != c.run.ID || s.OwnerInstanceID != c.task.instanceID || s.ExecutionEpoch != c.task.epoch {
		return store.ErrConflict
	}
	if old, ok := s.CheckpointKeys[candidate.CheckpointID]; ok {
		if old == digest {
			return nil
		}
		return fmt.Errorf("checkpoint.id: payload conflict")
	}
	if !critical && s.CheckpointCount >= 6 {
		return fmt.Errorf("checkpoint.limit: six updates already saved; continue to final result")
	}
	next := *s
	next.Findings = append([]core.TaskFinding(nil), s.Findings...)
	next.CheckpointKeys = map[string]string{}
	for k, v := range s.CheckpointKeys {
		next.CheckpointKeys[k] = v
	}
	for _, f := range findings {
		corrected := f.Corrects == ""
		for i := range next.Findings {
			old := &next.Findings[i]
			if old.ID == f.ID && old.Revision >= f.Revision {
				f.Revision = old.Revision + 1
			}
			if old.Status == "active" && (old.ID == f.ID || old.ID == f.Corrects) {
				if f.Corrects != "" {
					old.Status = "corrected"
					corrected = true
				} else {
					old.Status = "superseded"
				}
			}
		}
		if !corrected {
			return fmt.Errorf("checkpoint.corrects: active finding not found")
		}
		next.Findings = append(next.Findings, f)
	}
	next.CheckpointKeys[candidate.CheckpointID] = digest
	next.CheckpointSeq++
	if !critical {
		next.CheckpointCount++
	}
	next.Revision++
	next.OutcomeCompleteness = "partial"
	next.DeliveryState = "available"
	next.Focus = candidate.Focus
	next.Remaining = candidate.Remaining
	next.NextStep = candidate.NextStep
	now := time.Now().UTC()
	next.Phase = core.TaskPhase{ID: candidate.PhaseID, Label: map[string]string{"reading": "读取材料", "crosscheck": "核对材料", "summarizing": "整理结论", "draft-ready": "形成阶段成果"}[candidate.PhaseID], ObservedAt: now}
	next.LastActivityAt = now
	next.LastMeaningfulUpdateAt = now
	if err = next.Validate(); err != nil {
		if !critical {
			return err
		}
		// Reject an oversized replacement, while immediately withdrawing the known-wrong claim.
		next.Findings = next.Findings[:len(s.Findings)]
		next.Focus = ""
		next.Remaining = "更正说明超过展示上限，原结论已撤回，请核对来源"
		next.NextStep = "结合原件重新核验"
		next.Conclusion = "原结论已撤回，尚无可展示的替代结论"
		next.Phase = core.TaskPhase{ID: "correction", Label: "原结论已撤回", ObservedAt: now}
		for i := range next.Findings {
			if next.Findings[i].Status == "corrected" {
				next.Findings[i].Status = "withdrawn"
			}
		}
		if err = next.Validate(); err != nil {
			return err
		}
	}
	proofs := make([]store.ArtifactWitness, 0, len(witnesses))
	for _, w := range witnesses {
		proofs = append(proofs, w)
	}
	return c.mgr.cards.CommitTaskSources(c.run.Owner, next, s.Revision, proofs)
}

func (c *runCtx) WorkCheckpointAvailable() bool {
	return c.task.cardID != "" && c.subAgent == materialSubagentID && len(c.sources) > 0
}