diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 14d4146..30b185c 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -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 } diff --git a/dev/tasks/DEV-000132-work-graph-admission-scheduler.json b/dev/tasks/DEV-000132-work-graph-admission-scheduler.json new file mode 100644 index 0000000..fff53a5 --- /dev/null +++ b/dev/tasks/DEV-000132-work-graph-admission-scheduler.json @@ -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"] +} diff --git a/modules/work/domain/graph_scheduler.go b/modules/work/domain/graph_scheduler.go new file mode 100644 index 0000000..ea5b4ca --- /dev/null +++ b/modules/work/domain/graph_scheduler.go @@ -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 +} diff --git a/tests/agent_execution/graph_test.go b/tests/agent_execution/graph_test.go index 034c252..cc959fd 100644 --- a/tests/agent_execution/graph_test.go +++ b/tests/agent_execution/graph_test.go @@ -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) + } +}