feat(work): add graph admission scheduler
CI / test (push) Successful in 22s

This commit is contained in:
m
2026-08-22 13:31:39 +03:00
parent b2f922cbbc
commit 4a800df1e1
4 changed files with 108 additions and 2 deletions
+2 -2
View File
@@ -1,5 +1,5 @@
{
"task_file": "dev/tasks/DEV-000131-work-graph-harness.json",
"task_id": "DEV-000131",
"task_file": "dev/tasks/DEV-000132-work-graph-admission-scheduler.json",
"task_id": "DEV-000132",
"task_revision": 1
}
@@ -0,0 +1,23 @@
{
"task_id": "DEV-000132",
"task_revision": 1,
"architecture_revision": "ADR-0018",
"base_repo_revision": "b2f922c",
"risk_tier": "E",
"intent": "Add deterministic Work Graph admission, bounded ready-node scheduling and explicit conflict resolution to the graph harness.",
"requirement_ids": ["WP-400", "Graph admission", "Bounded scheduler", "Conflict resolution"],
"allowed_change_scope": {"modules": ["work-agent-execution"], "paths": ["modules/work/domain/graph_scheduler.go", "tests/agent_execution/graph_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000132-work-graph-admission-scheduler.json"]},
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "database/**", "adapters/**", "cmd/**", "runtime/**", ".gitea/**"]},
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
"contracts": {"consume": ["DEV-000131 graph harness", "ADR-0021 graph lifecycle"], "may_change": ["Pure graph admission/scheduling/conflict domain functions"]},
"semantic_owner": "work-agent-execution",
"canonical_state": ["No persistence; graph values remain deterministic harness state"],
"state_transitions": ["DRAFT -> ADMITTED; READY nodes bounded by MaxParallel; conflict unresolved -> explicitly resolved"],
"side_effects_and_privileges": ["Pure domain logic only"],
"persistence_and_migrations": ["Durable scheduler follows after the harness"],
"failure_retry_restart": ["Budget overflow and unknown conflict candidate reject without mutation"],
"assumptions": ["Runtime scheduler must use Agent Execution queue/leases in a later task"],
"acceptance_scenarios": ["Admission validates attempt budget", "Scheduler respects active/max parallel", "Conflict cannot resolve without explicit known candidate"],
"required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"],
"terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"]
}
+60
View File
@@ -0,0 +1,60 @@
package domain
import (
"errors"
"sort"
"git.cin.su/m/Assistant/contracts/work"
)
var (
ErrGraphBudgetExceeded = errors.New("work graph budget exceeded")
ErrConflictUnresolved = errors.New("conflict resolution is not explicit")
)
func ScheduleReadyNodes(graph work.WorkGraph, active int) ([]work.GraphNodeID, error) {
if err := ValidateWorkGraph(graph); err != nil {
return nil, err
}
if active < 0 || active > graph.Budget.MaxParallel {
return nil, ErrGraphBudgetExceeded
}
ready, err := ReadyGraphNodes(graph)
if err != nil {
return nil, err
}
available := graph.Budget.MaxParallel - active
if len(ready) > available {
ready = ready[:available]
}
return ready, nil
}
func AdmitGraph(graph work.WorkGraph) (work.WorkGraph, error) {
if err := ValidateWorkGraph(graph); err != nil {
return work.WorkGraph{}, err
}
for _, node := range graph.Nodes {
if node.Attempts > graph.Budget.MaxAttempts {
return work.WorkGraph{}, ErrGraphBudgetExceeded
}
}
graph.State = work.GraphAdmitted
return graph, nil
}
func ResolveGraphConflict(conflict work.ConflictSet, selectedCandidate string) (work.ConflictSet, error) {
if conflict.ID == "" || conflict.GraphID == "" || conflict.GraphRevision < 1 || conflict.NodeID == "" || conflict.Resolved || selectedCandidate == "" {
return work.ConflictSet{}, ErrConflictUnresolved
}
candidates := append([]string(nil), conflict.CandidateIDs...)
sort.Strings(candidates)
for _, candidate := range candidates {
if candidate == selectedCandidate {
conflict.CandidateIDs = candidates
conflict.Resolved = true
return conflict, nil
}
}
return work.ConflictSet{}, ErrConflictUnresolved
}
+23
View File
@@ -33,3 +33,26 @@ func TestWorkGraphKeepsDependentNodeBlocked(t *testing.T) {
t.Fatalf("dependent node became ready early: %#v %v", ready, err)
}
}
func TestGraphSchedulerEnforcesParallelBudgetAndExplicitConflict(t *testing.T) {
graph := work.WorkGraph{ID: "graph-1", Revision: 1, State: work.GraphDraft, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 2}, Nodes: []work.GraphNode{
{ID: "b", IdempotencyKey: "b-key", State: work.GraphNodeReady},
{ID: "a", IdempotencyKey: "a-key", State: work.GraphNodeReady},
}}
admitted, err := workdomain.AdmitGraph(graph)
if err != nil || admitted.State != work.GraphAdmitted {
t.Fatalf("graph was not admitted: %#v %v", admitted, err)
}
ready, err := workdomain.ScheduleReadyNodes(admitted, 1)
if err != nil || len(ready) != 1 || ready[0] != "a" {
t.Fatalf("scheduler ignored parallel budget/order: %#v %v", ready, err)
}
conflict := work.ConflictSet{ID: "conflict-1", GraphID: "graph-1", GraphRevision: 1, NodeID: "a", CandidateIDs: []string{"candidate-b", "candidate-a"}}
resolved, err := workdomain.ResolveGraphConflict(conflict, "candidate-a")
if err != nil || !resolved.Resolved || resolved.CandidateIDs[0] != "candidate-a" {
t.Fatalf("conflict was not explicitly resolved: %#v %v", resolved, err)
}
if _, err := workdomain.ResolveGraphConflict(conflict, "unknown"); err != workdomain.ErrConflictUnresolved {
t.Fatalf("unknown candidate was accepted: %v", err)
}
}