diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 9c5fc15..5d4e90a 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -42,6 +42,19 @@ func main() { } writeJSON(w, map[string]any{"graph": graph, "scheduled_nodes": ready}) }) + mux.HandleFunc("POST /v1/work-graphs/task-evaluations/schedule", func(w http.ResponseWriter, r *http.Request) { + var request application.TaskEvaluationScheduleRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&request); err != nil { + http.Error(w, "invalid task evaluation graph request", http.StatusBadRequest) + return + } + graph, ready, err := application.AdmitAndScheduleTaskEvaluation(r.Context(), repository, request) + if err != nil { + http.Error(w, err.Error(), http.StatusUnprocessableEntity) + return + } + writeJSON(w, map[string]any{"graph": graph, "scheduled_nodes": ready}) + }) mux.HandleFunc("POST /v1/work-graphs/complete", func(w http.ResponseWriter, r *http.Request) { var request application.GraphCompleteRequest if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&request); err != nil { diff --git a/contracts/work/graph.go b/contracts/work/graph.go index dccaf61..137eb30 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -1,11 +1,17 @@ package work +import "git.cin.su/m/Assistant/contracts/capability" + type GraphNodeID string type GraphState string type GraphNodeState string +type GraphNodePurpose string const ( + GraphNodePurposeTaskEvaluation GraphNodePurpose = "TASK_EVALUATION_SPECIALIST" + GraphNodePurposeGeneric GraphNodePurpose = "GENERIC" + GraphDraft GraphState = "DRAFT" GraphAdmitted GraphState = "ADMITTED" GraphRunning GraphState = "RUNNING" @@ -31,11 +37,15 @@ type GraphBudget struct { } type GraphNode struct { - ID GraphNodeID `json:"id"` - DispatchID string `json:"dispatch_id"` - IdempotencyKey string `json:"idempotency_key"` - State GraphNodeState `json:"state"` - Attempts int `json:"attempts"` + ID GraphNodeID `json:"id"` + DispatchID string `json:"dispatch_id"` + IdempotencyKey string `json:"idempotency_key"` + State GraphNodeState `json:"state"` + Attempts int `json:"attempts"` + Purpose GraphNodePurpose `json:"purpose,omitempty"` + TaskProfileID capability.TaskEvaluationProfileID `json:"task_profile_id,omitempty"` + BindingID capability.CapabilityBindingID `json:"binding_id,omitempty"` + CaseIDs []string `json:"case_ids,omitempty"` } type GraphEdge struct { @@ -44,12 +54,31 @@ type GraphEdge struct { } type WorkGraph struct { - ID string `json:"id"` - Revision int64 `json:"revision"` - State GraphState `json:"state"` - Nodes []GraphNode `json:"nodes"` - Edges []GraphEdge `json:"edges"` - Budget GraphBudget `json:"budget"` + ID string `json:"id"` + Revision int64 `json:"revision"` + State GraphState `json:"state"` + Nodes []GraphNode `json:"nodes"` + Edges []GraphEdge `json:"edges"` + Budget GraphBudget `json:"budget"` + ParentGraphID string `json:"parent_graph_id,omitempty"` + ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"` + TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"` +} + +// TaskEvaluationGraphRequest describes a durable fan-out. The graph contains +// one specialist node per binding; each node receives the profile's bounded +// cases and reports evidence back to Capability for fan-in aggregation. +type TaskEvaluationGraphRequest struct { + GraphID string `json:"graph_id"` + Profile capability.TaskEvaluationProfile `json:"profile"` + BindingIDs []capability.CapabilityBindingID `json:"binding_ids"` + DispatchIDs []string `json:"dispatch_ids"` + IdempotencyPrefix string `json:"idempotency_prefix"` + Budget GraphBudget `json:"budget"` + ParentGraphID string `json:"parent_graph_id,omitempty"` + ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"` + ParentGraphState GraphState `json:"parent_graph_state,omitempty"` + TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"` } type ResultCandidate struct { diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 4f9ebb0..9e32acd 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000208-task-aware-model-evaluation.json", - "task_id": "DEV-000208", + "task_file": "dev/tasks/DEV-000209-task-evaluator-work-graph.json", + "task_id": "DEV-000209", "task_revision": 1 } diff --git a/dev/tasks/DEV-000209-task-evaluator-work-graph.json b/dev/tasks/DEV-000209-task-evaluator-work-graph.json new file mode 100644 index 0000000..4b6ef13 --- /dev/null +++ b/dev/tasks/DEV-000209-task-evaluator-work-graph.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000209", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "a82592a1300c06ed7a568c058faca6560e085ab3", + "risk_tier": "E", + "intent": "Create durable Work Graph plans for task-aware model evaluation: run the general profile first, fan out independent model specialists in parallel, and permit bounded additional profiles linked to unresolved evidence without blocking the chat path.", + "requirement_ids": ["general-first model evaluation", "parallel specialist evaluation", "bounded additional task tests", "fan-out/fan-in", "no synchronous chat blocking", "provider and hardware neutral"], + "allowed_change_scope": {"modules": ["work"], "paths": ["contracts/work/**", "modules/work/**", "cmd/work-graph-runtime/**", "tests/model_qualification/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000209-task-evaluator-work-graph.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "database/**"]}, + "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": ["capability.TaskEvaluationProfile", "capability.TaskEvaluationProfileStage", "capability.TaskEvaluationProfileID", "existing Work Graph schedule contract"], "may_change": ["typed Work Graph node purpose and task-evaluation planning contract"]}, + "semantic_owner": "work", + "canonical_state": ["work-owned durable graph and node lifecycle; capability remains owner of model evaluation evidence"], + "state_transitions": ["GENERAL profile -> admitted evaluation graph", "eligible model bindings -> parallel specialist nodes", "all specialist nodes -> fan-in completion state", "ADDITIONAL profile -> a separate bounded graph linked by parent/trigger evidence"], + "side_effects_and_privileges": ["none; graph planning only, no provider calls and no secret access"], + "persistence_and_migrations": ["uses existing work graph persistence; no new schema"], + "failure_retry_restart": ["each specialist remains independently retryable through existing node attempts/leases; partial or unknown graph state is preserved", "invalid stage/parent linkage is rejected before scheduling"], + "assumptions": ["a later worker adapter executes each dispatched specialist and posts evidence to capability", "chat/request handlers enqueue the graph and return without waiting for all nodes"], + "acceptance_scenarios": ["additional profile cannot be scheduled as a replacement for general profile", "one node is created per binding and profile with independent dispatch identity", "parallel budget is retained for fan-out", "graph metadata is sufficient for a worker to identify profile, binding and specialist role", "provider kind, endpoint and hardware are absent from planning inputs"], + "required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production public health check"], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY"] +} diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index 3b8281d..5a00c38 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -18,6 +18,12 @@ type GraphScheduleRequest struct { 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"` @@ -105,6 +111,18 @@ func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request Graph 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 + } + return AdmitAndSchedule(ctx, store, GraphScheduleRequest{Graph: graph, Intents: request.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 { diff --git a/modules/work/domain/task_evaluation_graph.go b/modules/work/domain/task_evaluation_graph.go new file mode 100644 index 0000000..a034174 --- /dev/null +++ b/modules/work/domain/task_evaluation_graph.go @@ -0,0 +1,62 @@ +package domain + +import ( + "errors" + "fmt" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" +) + +var ErrInvalidTaskEvaluationGraph = errors.New("invalid task evaluation graph request") + +// BuildTaskEvaluationGraph turns one bounded profile into durable specialist +// nodes. It deliberately has no provider, endpoint or hardware inputs: those +// are resolved later by the cognitive worker for each binding. +func BuildTaskEvaluationGraph(request work.TaskEvaluationGraphRequest) (work.WorkGraph, error) { + if request.GraphID == "" || request.IdempotencyPrefix == "" || len(request.BindingIDs) == 0 || len(request.BindingIDs) != len(request.DispatchIDs) || request.Budget.MaxParallel < 1 || request.Budget.MaxAttempts < 1 { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if err := request.Profile.Validate(); err != nil { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if request.Profile.Stage == capability.TaskEvaluationGeneral { + if request.ParentGraphID != "" || request.ParentGraphRevision != 0 || request.ParentGraphState != "" || request.TriggerEvidenceRef != "" { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + } else { + if request.ParentGraphID == "" || request.ParentGraphRevision < 1 || request.TriggerEvidenceRef == "" || (request.ParentGraphState != work.GraphCompleted && request.ParentGraphState != work.GraphPartial) { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if request.Profile.ParentProfileID == "" { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + } + seenBindings := make(map[capability.CapabilityBindingID]struct{}, len(request.BindingIDs)) + seenDispatches := make(map[string]struct{}, len(request.DispatchIDs)) + nodes := make([]work.GraphNode, 0, len(request.BindingIDs)) + for i, bindingID := range request.BindingIDs { + if bindingID == "" || request.DispatchIDs[i] == "" { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if _, ok := seenBindings[bindingID]; ok { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if _, ok := seenDispatches[request.DispatchIDs[i]]; ok { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + seenBindings[bindingID] = struct{}{} + seenDispatches[request.DispatchIDs[i]] = struct{}{} + nodes = append(nodes, work.GraphNode{ + ID: work.GraphNodeID(fmt.Sprintf("%s-node-%d", request.GraphID, i+1)), + DispatchID: request.DispatchIDs[i], + IdempotencyKey: fmt.Sprintf("%s:%s:%d", request.IdempotencyPrefix, bindingID, request.Profile.Revision), + State: work.GraphNodeBlocked, + Purpose: work.GraphNodePurposeTaskEvaluation, + TaskProfileID: request.Profile.ID, + BindingID: bindingID, + CaseIDs: append([]string(nil), request.Profile.RequiredCaseIDs...), + }) + } + return work.WorkGraph{ID: request.GraphID, Revision: 1, State: work.GraphDraft, Nodes: nodes, Budget: request.Budget, ParentGraphID: request.ParentGraphID, ParentGraphRevision: request.ParentGraphRevision, TriggerEvidenceRef: request.TriggerEvidenceRef}, nil +} diff --git a/modules/work/domain/task_evaluation_graph_test.go b/modules/work/domain/task_evaluation_graph_test.go new file mode 100644 index 0000000..becd2da --- /dev/null +++ b/modules/work/domain/task_evaluation_graph_test.go @@ -0,0 +1,52 @@ +package domain + +import ( + "testing" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" +) + +func evaluationGraphRequest(stage capability.TaskEvaluationProfileStage) work.TaskEvaluationGraphRequest { + profile := capability.TaskEvaluationProfile{ID: "profile-1", Revision: 1, CapabilityID: "capability-1", Stage: stage, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-a", "case-b"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100} + request := work.TaskEvaluationGraphRequest{GraphID: "graph-1", Profile: profile, BindingIDs: []capability.CapabilityBindingID{"binding-a", "binding-b"}, DispatchIDs: []string{"dispatch-a", "dispatch-b"}, IdempotencyPrefix: "task-eval", Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 2}} + if stage == capability.TaskEvaluationAdditional { + request.Profile.ParentProfileID = "profile-general" + request.Profile.TriggerEvidenceRef = "artifact:unresolved-1" + request.Profile.Stage = capability.TaskEvaluationAdditional + request.ParentGraphID, request.ParentGraphRevision, request.ParentGraphState, request.TriggerEvidenceRef = "graph-general", 2, work.GraphPartial, "artifact:unresolved-1" + } + return request +} + +func TestBuildTaskEvaluationGraphFansOutOneNodePerBinding(t *testing.T) { + graph, err := BuildTaskEvaluationGraph(evaluationGraphRequest(capability.TaskEvaluationGeneral)) + if err != nil { + t.Fatal(err) + } + if len(graph.Nodes) != 2 || graph.Nodes[0].Purpose != work.GraphNodePurposeTaskEvaluation || graph.Nodes[0].TaskProfileID != "profile-1" || len(graph.Nodes[0].CaseIDs) != 2 { + t.Fatalf("unexpected task graph nodes: %+v", graph.Nodes) + } + if graph.ParentGraphID != "" || graph.TriggerEvidenceRef != "" { + t.Fatalf("general graph unexpectedly has expansion lineage: %+v", graph) + } +} + +func TestBuildTaskEvaluationGraphRequiresCompletedOrPartialGeneralParent(t *testing.T) { + request := evaluationGraphRequest(capability.TaskEvaluationAdditional) + request.ParentGraphState = work.GraphRunning + if _, err := BuildTaskEvaluationGraph(request); err == nil { + t.Fatal("expected additional graph to require a finished parent graph") + } +} + +func TestBuildTaskEvaluationGraphRejectsGeneralReplacement(t *testing.T) { + request := evaluationGraphRequest(capability.TaskEvaluationGeneral) + request.ParentGraphID = "old-graph" + request.ParentGraphRevision = 2 + request.ParentGraphState = work.GraphPartial + request.TriggerEvidenceRef = "artifact:unresolved" + if _, err := BuildTaskEvaluationGraph(request); err == nil { + t.Fatal("expected general graph replacement to be rejected") + } +}