280 lines
11 KiB
Go
280 lines
11 KiB
Go
package application
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/contracts/authority"
|
|
"git.cin.su/m/Assistant/contracts/capability"
|
|
"git.cin.su/m/Assistant/contracts/identity"
|
|
"git.cin.su/m/Assistant/contracts/work"
|
|
"git.cin.su/m/Assistant/modules/work/domain"
|
|
"git.cin.su/m/Assistant/modules/work/ports"
|
|
)
|
|
|
|
var ErrGraphDispatchBinding = errors.New("graph node dispatch binding is missing")
|
|
var ErrTaskEvaluationCallbackMismatch = errors.New("task evaluation callback does not match graph node")
|
|
var ErrGraphResultCallbackMismatch = errors.New("graph result callback does not match graph node")
|
|
|
|
type GraphScheduleRequest struct {
|
|
Graph work.WorkGraph `json:"graph"`
|
|
Intents []work.AgentDispatchIntent `json:"intents"`
|
|
Now int64 `json:"now_unix_millis"`
|
|
}
|
|
|
|
type TaskEvaluationScheduleRequest struct {
|
|
Plan work.TaskEvaluationGraphRequest `json:"plan"`
|
|
Intents []work.AgentDispatchIntent `json:"intents"`
|
|
Now int64 `json:"now_unix_millis"`
|
|
}
|
|
|
|
type GraphCompleteRequest struct {
|
|
GraphID string `json:"graph_id"`
|
|
Revision int64 `json:"revision"`
|
|
NodeID work.GraphNodeID `json:"node_id"`
|
|
Succeeded bool `json:"succeeded"`
|
|
}
|
|
|
|
type GraphSynthesisRequest struct {
|
|
Conflict work.ConflictSet `json:"conflict"`
|
|
Candidates []work.ResultCandidate `json:"candidates"`
|
|
SelectedCandidate string `json:"selected_candidate_id"`
|
|
VerifierRunID work.AgentRunID `json:"verifier_run_id"`
|
|
VerifierIdentity string `json:"verifier_identity"`
|
|
EvidenceRefs []work.EvidenceRef `json:"evidence_refs"`
|
|
Summary string `json:"summary"`
|
|
Verified bool `json:"verified"`
|
|
}
|
|
|
|
func CompleteTaskEvaluationNode(ctx context.Context, graphStore ports.GraphStore, evidenceSink ports.TaskEvaluationSink, request work.TaskEvaluationEvidenceCallback) (work.WorkGraph, error) {
|
|
if evidenceSink == nil || request.GraphID == "" || request.Revision < 1 || request.NodeID == "" || !request.Succeeded {
|
|
return work.WorkGraph{}, ErrTaskEvaluationCallbackMismatch
|
|
}
|
|
graph, err := graphStore.LoadWorkGraph(ctx, request.GraphID, request.Revision)
|
|
if err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
matched := false
|
|
for _, node := range graph.Nodes {
|
|
if node.ID != request.NodeID {
|
|
continue
|
|
}
|
|
matched = node.Purpose == work.GraphNodePurposeTaskEvaluation && node.TaskProfileID == request.Evaluation.ProfileID && node.BindingID == request.Evaluation.BindingID
|
|
break
|
|
}
|
|
if !matched || request.Evaluation.BindingRevision < 1 || request.Evaluation.CapabilityID == "" {
|
|
return work.WorkGraph{}, ErrTaskEvaluationCallbackMismatch
|
|
}
|
|
if err := evidenceSink.SaveTaskEvaluation(ctx, request.Evaluation); err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
return CompleteGraphNode(ctx, graphStore, request.GraphID, request.Revision, request.NodeID, true)
|
|
}
|
|
|
|
func SynthesizeGraph(ctx context.Context, store ports.GraphResultStore, request GraphSynthesisRequest) (work.ConflictSet, work.GraphSynthesisCandidate, error) {
|
|
conflict, synthesis, err := domain.BuildVerifiedGraphSynthesis(request.Conflict, request.Candidates, request.SelectedCandidate, request.VerifierRunID, request.VerifierIdentity, request.EvidenceRefs, request.Summary, request.Verified)
|
|
if err != nil {
|
|
return work.ConflictSet{}, work.GraphSynthesisCandidate{}, err
|
|
}
|
|
if err := store.SaveGraphSynthesis(ctx, conflict, request.Candidates, synthesis); err != nil {
|
|
return work.ConflictSet{}, work.GraphSynthesisCandidate{}, err
|
|
}
|
|
return conflict, synthesis, nil
|
|
}
|
|
|
|
// RecordGraphResultCandidate persists attributable agent output. The graph is
|
|
// immutable by revision, so candidate recording does not rewrite the node;
|
|
// only verified synthesis may create the next successful graph revision.
|
|
func RecordGraphResultCandidate(ctx context.Context, candidateStore ports.GraphCandidateStore, graphStore ports.GraphStore, request work.GraphResultCallback) (work.WorkGraph, error) {
|
|
candidate := request.Candidate
|
|
if candidateStore == nil || request.GraphID == "" || request.Revision < 1 || candidate.ID == "" || candidate.GraphID != request.GraphID || candidate.GraphRevision != request.Revision || candidate.NodeID == "" || candidate.AuthorRunID == "" || candidate.EvidenceSummary == "" || candidate.Summary == "" {
|
|
return work.WorkGraph{}, ErrGraphResultCallbackMismatch
|
|
}
|
|
graph, err := graphStore.LoadWorkGraph(ctx, request.GraphID, request.Revision)
|
|
if err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
found := false
|
|
for i := range graph.Nodes {
|
|
if graph.Nodes[i].ID != candidate.NodeID {
|
|
continue
|
|
}
|
|
found = true
|
|
if graph.Nodes[i].State == work.GraphNodeSuccess || graph.Nodes[i].State == work.GraphNodeCancel {
|
|
return work.WorkGraph{}, ErrGraphResultCallbackMismatch
|
|
}
|
|
}
|
|
if !found {
|
|
return work.WorkGraph{}, ErrGraphResultCallbackMismatch
|
|
}
|
|
if err := candidateStore.SaveGraphResultCandidate(ctx, candidate); err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
return graph, nil
|
|
}
|
|
|
|
// SynthesizeAndCompleteGraph is the only path that turns a verifying node
|
|
// into success: synthesis must be independently verified first.
|
|
func SynthesizeAndCompleteGraph(ctx context.Context, resultStore ports.GraphResultStore, graphStore ports.GraphStore, request GraphSynthesisRequest) (work.ConflictSet, work.GraphSynthesisCandidate, work.WorkGraph, error) {
|
|
conflict, synthesis, err := SynthesizeGraph(ctx, resultStore, request)
|
|
if err != nil {
|
|
return work.ConflictSet{}, work.GraphSynthesisCandidate{}, work.WorkGraph{}, err
|
|
}
|
|
graph, err := CompleteGraphNode(ctx, graphStore, conflict.GraphID, conflict.GraphRevision, conflict.NodeID, true)
|
|
if err != nil {
|
|
return work.ConflictSet{}, work.GraphSynthesisCandidate{}, work.WorkGraph{}, err
|
|
}
|
|
return conflict, synthesis, graph, nil
|
|
}
|
|
|
|
func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request GraphScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) {
|
|
if request.Now < 1 {
|
|
request.Now = time.Now().UnixMilli()
|
|
}
|
|
admitted, err := domain.AdmitGraph(request.Graph)
|
|
if err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
if err := store.SaveWorkGraph(ctx, admitted); err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
ready, err := domain.ScheduleReadyNodes(admitted, 0)
|
|
if err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
intentByDispatch := make(map[string]work.AgentDispatchIntent, len(request.Intents))
|
|
for _, intent := range request.Intents {
|
|
intentByDispatch[intent.DispatchID] = intent
|
|
}
|
|
for _, node := range admitted.Nodes {
|
|
if node.DispatchID == "" {
|
|
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
|
|
}
|
|
if _, ok := intentByDispatch[node.DispatchID]; !ok {
|
|
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
|
|
}
|
|
}
|
|
for _, intent := range request.Intents {
|
|
if err := store.SaveDispatchIntent(ctx, intent); err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
}
|
|
next := admitted
|
|
next.Revision++
|
|
next.State = work.GraphRunning
|
|
selected := make(map[work.GraphNodeID]bool, len(ready))
|
|
for _, nodeID := range ready {
|
|
selected[nodeID] = true
|
|
}
|
|
for i := range next.Nodes {
|
|
if !selected[next.Nodes[i].ID] {
|
|
continue
|
|
}
|
|
if next.Nodes[i].DispatchID == "" {
|
|
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
|
|
}
|
|
intent := intentByDispatch[next.Nodes[i].DispatchID]
|
|
if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: intent.DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: request.Now, UpdatedAtUnixMillis: request.Now}); err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
next.Nodes[i].State = work.GraphNodeLeased
|
|
}
|
|
if err := store.SaveWorkGraph(ctx, next); err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
return next, ready, nil
|
|
}
|
|
|
|
// AdmitAndScheduleTaskEvaluation creates the durable general-first or
|
|
// bounded-expansion fan-out before handing it to the existing graph scheduler.
|
|
// The request path only persists/enqueues work; specialist execution happens
|
|
// independently through the dispatch workers.
|
|
func AdmitAndScheduleTaskEvaluation(ctx context.Context, store ports.GraphStore, request TaskEvaluationScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) {
|
|
graph, err := domain.BuildTaskEvaluationGraph(request.Plan)
|
|
if err != nil {
|
|
return work.WorkGraph{}, nil, err
|
|
}
|
|
intents := make([]work.AgentDispatchIntent, len(request.Intents))
|
|
copy(intents, request.Intents)
|
|
for i := range intents {
|
|
for _, node := range graph.Nodes {
|
|
if node.DispatchID != intents[i].DispatchID {
|
|
continue
|
|
}
|
|
intents[i].SpecialistPurpose = string(node.Purpose)
|
|
intents[i].TaskEvaluationProfileID = node.TaskProfileID
|
|
intents[i].TaskEvaluationBindingID = node.BindingID
|
|
intents[i].TaskEvaluationCaseIDs = append([]string(nil), node.CaseIDs...)
|
|
if node.TaskProfile != nil {
|
|
profile := *node.TaskProfile
|
|
profile.RequiredCaseIDs = append([]string(nil), node.TaskProfile.RequiredCaseIDs...)
|
|
intents[i].TaskEvaluationProfile = &profile
|
|
}
|
|
intents[i].TaskEvaluationCases = append([]capability.TaskEvaluationCase(nil), node.TaskCases...)
|
|
intents[i].TaskEvaluationExecution = &work.TaskEvaluationDispatchContext{GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: node.ID, BindingRevision: request.Plan.BindingRevision, ActorID: identity.ActorID(request.Plan.ActorID), PrincipalID: identity.PrincipalID(request.Plan.PrincipalID), ScopeID: authority.ScopeID(request.Plan.ScopeID)}
|
|
break
|
|
}
|
|
}
|
|
return AdmitAndSchedule(ctx, store, GraphScheduleRequest{Graph: graph, Intents: intents, Now: request.Now})
|
|
}
|
|
|
|
func CompleteGraphNode(ctx context.Context, store ports.GraphStore, graphID string, revision int64, nodeID work.GraphNodeID, success bool) (work.WorkGraph, error) {
|
|
graph, err := store.LoadWorkGraph(ctx, graphID, revision)
|
|
if err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
found := false
|
|
for i := range graph.Nodes {
|
|
if graph.Nodes[i].ID == nodeID {
|
|
found = true
|
|
if success {
|
|
graph.Nodes[i].State = work.GraphNodeSuccess
|
|
} else {
|
|
graph.Nodes[i].State = work.GraphNodeFailed
|
|
}
|
|
}
|
|
}
|
|
if !found {
|
|
return work.WorkGraph{}, domain.ErrInvalidGraph
|
|
}
|
|
if !success {
|
|
graph.State = work.GraphFailed
|
|
} else {
|
|
ready, err := domain.ReadyGraphNodes(graph)
|
|
if err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
for i := range graph.Nodes {
|
|
for _, candidate := range ready {
|
|
if graph.Nodes[i].ID == candidate && graph.Nodes[i].State == work.GraphNodeBlocked {
|
|
graph.Nodes[i].State = work.GraphNodeReady
|
|
if graph.Nodes[i].DispatchID == "" {
|
|
return work.WorkGraph{}, ErrGraphDispatchBinding
|
|
}
|
|
if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: graph.Nodes[i].DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: time.Now().UnixMilli(), UpdatedAtUnixMillis: time.Now().UnixMilli()}); err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
allSuccess := true
|
|
for _, node := range graph.Nodes {
|
|
if node.State != work.GraphNodeSuccess {
|
|
allSuccess = false
|
|
break
|
|
}
|
|
}
|
|
if allSuccess {
|
|
graph.State = work.GraphCompleted
|
|
} else {
|
|
graph.State = work.GraphPartial
|
|
}
|
|
}
|
|
graph.Revision++
|
|
if err := store.SaveWorkGraph(ctx, graph); err != nil {
|
|
return work.WorkGraph{}, err
|
|
}
|
|
return graph, nil
|
|
}
|