Files
Assistant/tests/agent_execution/graph_runtime_test.go
m da251e1aa8
CI / test (push) Successful in 1m8s
CI / test (pull_request) Successful in 1m11s
feat(work): persist orchestration snapshot for replan
2026-08-23 18:13:07 +03:00

135 lines
8.6 KiB
Go

package agent_execution_test
import (
"context"
"encoding/json"
"strconv"
"strings"
"testing"
"git.cin.su/m/Assistant/contracts/work"
"git.cin.su/m/Assistant/modules/work/application"
)
type graphStore struct {
graphs map[string]work.WorkGraph
queued []string
intents []string
candidates []work.ResultCandidate
syntheses int
}
func (s *graphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error {
if s.graphs == nil {
s.graphs = map[string]work.WorkGraph{}
}
s.graphs[graph.ID+":"+itoa(graph.Revision)] = graph
return nil
}
func (s *graphStore) LoadWorkGraph(_ context.Context, id string, revision int64) (work.WorkGraph, error) {
return s.graphs[id+":"+itoa(revision)], nil
}
func (s *graphStore) SaveDispatchIntent(_ context.Context, intent work.AgentDispatchIntent) error {
s.intents = append(s.intents, intent.DispatchID)
return nil
}
func (s *graphStore) EnqueueDispatch(_ context.Context, item work.AgentDispatchQueueItem) error {
s.queued = append(s.queued, item.DispatchID)
return nil
}
func (s *graphStore) SaveGraphResultCandidate(_ context.Context, candidate work.ResultCandidate) error {
s.candidates = append(s.candidates, candidate)
return nil
}
func (s *graphStore) SaveGraphSynthesis(_ context.Context, _ work.ConflictSet, _ []work.ResultCandidate, _ work.GraphSynthesisCandidate) error {
s.syntheses++
return nil
}
func itoa(value int64) string {
return strconv.FormatInt(value, 10)
}
func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) {
store := &graphStore{}
graph, ready, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "g", Revision: 1, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "a", DispatchID: "d-a", IdempotencyKey: "a", State: work.GraphNodeReady}, {ID: "b", DispatchID: "d-b", IdempotencyKey: "b", State: work.GraphNodeReady}, {ID: "join", DispatchID: "d-join", IdempotencyKey: "join", State: work.GraphNodeBlocked}}, Edges: []work.GraphEdge{{From: "a", To: "join"}, {From: "b", To: "join"}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "d-a"}, {DispatchID: "d-b"}, {DispatchID: "d-join"}}, Now: 1})
if err != nil || len(ready) != 2 || len(store.queued) != 2 {
t.Fatalf("graph=%+v ready=%v queued=%v err=%v", graph, ready, store.queued, err)
}
if _, err := application.CompleteGraphNode(context.Background(), store, "g", 2, "a", true); err != nil {
t.Fatal(err)
}
if _, err := application.CompleteGraphNode(context.Background(), store, "g", 3, "b", true); err != nil {
t.Fatal(err)
}
if len(store.queued) != 3 || store.queued[2] != "d-join" {
t.Fatalf("fan-in dispatch was not enqueued: %v", store.queued)
}
completed, err := application.CompleteGraphNode(context.Background(), store, "g", 4, "join", true)
if err != nil || completed.State != work.GraphCompleted {
t.Fatalf("completed=%+v err=%v", completed, err)
}
}
type graphResultStoreStub struct {
conflict work.ConflictSet
count int
}
func (s *graphResultStoreStub) SaveGraphSynthesis(_ context.Context, conflict work.ConflictSet, candidates []work.ResultCandidate, _ work.GraphSynthesisCandidate) error {
s.conflict = conflict
s.count = len(candidates)
return nil
}
func TestSynthesizeGraphPersistsVerifiedDissent(t *testing.T) {
store := &graphResultStoreStub{}
conflict, synthesis, err := application.SynthesizeGraph(context.Background(), store, application.GraphSynthesisRequest{
Conflict: work.ConflictSet{ID: "conflict-1", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", CandidateIDs: []string{"candidate-a", "candidate-b"}},
Candidates: []work.ResultCandidate{{ID: "candidate-a", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", AuthorRunID: "producer-a", EvidenceSummary: "a", Summary: "a"}, {ID: "candidate-b", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", AuthorRunID: "producer-b", EvidenceSummary: "b", Summary: "b"}},
SelectedCandidate: "candidate-a", VerifierRunID: "verifier-1", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"tests:1"}, Summary: "verified", Verified: true,
})
if err != nil || !conflict.Resolved || synthesis.State != work.GraphSynthesisVerified || store.count != 2 || !store.conflict.Resolved {
t.Fatalf("unexpected runtime synthesis: conflict=%+v synthesis=%+v stored=%+v err=%v", conflict, synthesis, store.conflict, err)
}
}
func TestGraphResultCallbackRequiresIndependentSynthesisBeforeSuccess(t *testing.T) {
store := &graphStore{}
graph, _, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "result-graph", Revision: 1, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "node-1", DispatchID: "dispatch-1", IdempotencyKey: "node-1", State: work.GraphNodeReady}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "dispatch-1"}}, Now: 1})
if err != nil {
t.Fatal(err)
}
returned, err := application.RecordGraphResultCandidate(context.Background(), store, store, work.GraphResultCallback{GraphID: graph.ID, Revision: graph.Revision, Candidate: work.ResultCandidate{ID: "candidate-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-1", EvidenceSummary: "evidence", Summary: "candidate"}})
if err != nil || len(store.candidates) != 1 || returned.State != work.GraphRunning {
t.Fatalf("candidate callback changed completion state: returned=%+v candidates=%d err=%v", returned, len(store.candidates), err)
}
conflict, synthesis, completed, err := application.SynthesizeAndCompleteGraph(context.Background(), store, store, application.GraphSynthesisRequest{Conflict: work.ConflictSet{ID: "conflict-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", CandidateIDs: []string{"candidate-1", "candidate-2"}}, Candidates: []work.ResultCandidate{{ID: "candidate-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-1", EvidenceSummary: "evidence", Summary: "candidate"}, {ID: "candidate-2", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-2", EvidenceSummary: "evidence-2", Summary: "candidate-2"}}, SelectedCandidate: "candidate-1", VerifierRunID: "verifier-1", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"test:evidence"}, Summary: "verified", Verified: true})
if err != nil || !conflict.Resolved || synthesis.State != work.GraphSynthesisVerified || completed.State != work.GraphCompleted {
t.Fatalf("verified fan-in did not complete graph: conflict=%+v synthesis=%+v graph=%+v err=%v", conflict, synthesis, completed, err)
}
}
func TestCompletedOrchestrationGraphReturnsDurableNextWaveDecision(t *testing.T) {
store := &graphStore{}
graph, _, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "replan-graph", Revision: 1, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, SearchInputBound: true, SearchInput: work.SolutionSearchInput{GoalID: "goal-1", GoalState: work.GoalInProgress, SearchRound: 1, MaxRound: 3}, Nodes: []work.GraphNode{{ID: "node-1", DispatchID: "dispatch-1", IdempotencyKey: "node-1", State: work.GraphNodeReady}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "dispatch-1"}}, Now: 1})
if err != nil {
t.Fatal(err)
}
result, err := application.CompleteGraphNodeWithReplan(context.Background(), store, application.GraphCompleteRequest{GraphID: graph.ID, Revision: graph.Revision, NodeID: "node-1", Succeeded: true})
if err != nil || !result.ReplanRequired || !result.NextSearchDecision.ContinueSearching || result.NextSearchDecision.NextSearchRound != 2 {
t.Fatalf("missing next-wave decision: result=%+v err=%v", result, err)
}
}
func TestGraphSynthesisJSONUsesStableSnakeCaseContract(t *testing.T) {
payload, err := json.Marshal(application.GraphSynthesisRequest{Conflict: work.ConflictSet{ID: "c", GraphID: "g", GraphRevision: 1, NodeID: "n", CandidateIDs: []string{"a", "b"}}, SelectedCandidate: "a", VerifierRunID: "v", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"e"}, Summary: "s", Verified: true})
if err != nil || !strings.Contains(string(payload), `"graph_id":"g"`) || !strings.Contains(string(payload), `"selected_candidate_id":"a"`) {
t.Fatalf("unexpected synthesis JSON: %s err=%v", payload, err)
}
var decoded application.GraphSynthesisRequest
if err := json.Unmarshal(payload, &decoded); err != nil || decoded.Conflict.GraphID != "g" || decoded.SelectedCandidate != "a" || !decoded.Verified {
t.Fatalf("snake_case request did not round-trip: %+v err=%v", decoded, err)
}
}