diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index e2c1e35..8c5b6f4 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -2,11 +2,14 @@ package postgres import ( "context" + "encoding/json" "errors" "fmt" + "reflect" "slices" "time" + "git.cin.su/m/Assistant/contracts/capability" "git.cin.su/m/Assistant/contracts/work" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -107,15 +110,32 @@ func (r *WorkRepository) SaveDispatchIntent(ctx context.Context, intent work.Age return err } defer tx.Rollback(ctx) - _, err = tx.Exec(ctx, `INSERT INTO work.agent_dispatch_intents(dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT(idempotency_key) DO NOTHING`, intent.DispatchID, intent.IdempotencyKey, intent.DelegationID, intent.RunID, intent.ImplementationID, intent.ImplementationRevision, intent.SourceRevision, intent.WorktreeID, intent.VerificationPlanRef, intent.CreatedAtUnixMillis, intent.SpecialistPurpose, intent.TaskEvaluationProfileID, intent.TaskEvaluationBindingID, intent.TaskEvaluationCaseIDs) + profileJSON, casesJSON, err := marshalTaskPayload(intent.TaskEvaluationProfile, intent.TaskEvaluationCases) + if err != nil { + return err + } + executionJSON, err := json.Marshal(intent.TaskEvaluationExecution) + if err != nil { + return err + } + _, err = tx.Exec(ctx, `INSERT INTO work.agent_dispatch_intents(dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids,task_evaluation_profile,task_evaluation_cases,task_evaluation_execution) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) ON CONFLICT(idempotency_key) DO NOTHING`, intent.DispatchID, intent.IdempotencyKey, intent.DelegationID, intent.RunID, intent.ImplementationID, intent.ImplementationRevision, intent.SourceRevision, intent.WorktreeID, intent.VerificationPlanRef, intent.CreatedAtUnixMillis, intent.SpecialistPurpose, intent.TaskEvaluationProfileID, intent.TaskEvaluationBindingID, intent.TaskEvaluationCaseIDs, profileJSON, casesJSON, executionJSON) if err != nil { return fmt.Errorf("save dispatch intent: %w", err) } var existing work.AgentDispatchIntent - err = tx.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids FROM work.agent_dispatch_intents WHERE idempotency_key=$1`, intent.IdempotencyKey).Scan(&existing.DispatchID, &existing.IdempotencyKey, &existing.DelegationID, &existing.RunID, &existing.ImplementationID, &existing.ImplementationRevision, &existing.SourceRevision, &existing.WorktreeID, &existing.VerificationPlanRef, &existing.CreatedAtUnixMillis, &existing.SpecialistPurpose, &existing.TaskEvaluationProfileID, &existing.TaskEvaluationBindingID, &existing.TaskEvaluationCaseIDs) + var profileJSONExisting, casesJSONExisting, executionJSONExisting []byte + err = tx.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids,task_evaluation_profile,task_evaluation_cases,task_evaluation_execution FROM work.agent_dispatch_intents WHERE idempotency_key=$1`, intent.IdempotencyKey).Scan(&existing.DispatchID, &existing.IdempotencyKey, &existing.DelegationID, &existing.RunID, &existing.ImplementationID, &existing.ImplementationRevision, &existing.SourceRevision, &existing.WorktreeID, &existing.VerificationPlanRef, &existing.CreatedAtUnixMillis, &existing.SpecialistPurpose, &existing.TaskEvaluationProfileID, &existing.TaskEvaluationBindingID, &existing.TaskEvaluationCaseIDs, &profileJSONExisting, &casesJSONExisting, &executionJSONExisting) if err != nil { return fmt.Errorf("read dispatch intent: %w", err) } + if err := unmarshalTaskPayload(profileJSONExisting, casesJSONExisting, &existing.TaskEvaluationProfile, &existing.TaskEvaluationCases); err != nil { + return err + } + if len(executionJSONExisting) > 0 && string(executionJSONExisting) != "null" { + if err := json.Unmarshal(executionJSONExisting, &existing.TaskEvaluationExecution); err != nil { + return err + } + } if !sameDispatchIntent(existing, intent) { return ErrDispatchIntentConflict } @@ -123,15 +143,24 @@ func (r *WorkRepository) SaveDispatchIntent(ctx context.Context, intent work.Age } func sameDispatchIntent(left, right work.AgentDispatchIntent) bool { - return left.DispatchID == right.DispatchID && left.IdempotencyKey == right.IdempotencyKey && left.DelegationID == right.DelegationID && left.RunID == right.RunID && left.ImplementationID == right.ImplementationID && left.ImplementationRevision == right.ImplementationRevision && left.SourceRevision == right.SourceRevision && left.WorktreeID == right.WorktreeID && left.VerificationPlanRef == right.VerificationPlanRef && left.CreatedAtUnixMillis == right.CreatedAtUnixMillis && left.SpecialistPurpose == right.SpecialistPurpose && left.TaskEvaluationProfileID == right.TaskEvaluationProfileID && left.TaskEvaluationBindingID == right.TaskEvaluationBindingID && slices.Equal(left.TaskEvaluationCaseIDs, right.TaskEvaluationCaseIDs) + return left.DispatchID == right.DispatchID && left.IdempotencyKey == right.IdempotencyKey && left.DelegationID == right.DelegationID && left.RunID == right.RunID && left.ImplementationID == right.ImplementationID && left.ImplementationRevision == right.ImplementationRevision && left.SourceRevision == right.SourceRevision && left.WorktreeID == right.WorktreeID && left.VerificationPlanRef == right.VerificationPlanRef && left.CreatedAtUnixMillis == right.CreatedAtUnixMillis && left.SpecialistPurpose == right.SpecialistPurpose && left.TaskEvaluationProfileID == right.TaskEvaluationProfileID && left.TaskEvaluationBindingID == right.TaskEvaluationBindingID && slices.Equal(left.TaskEvaluationCaseIDs, right.TaskEvaluationCaseIDs) && reflect.DeepEqual(left.TaskEvaluationProfile, right.TaskEvaluationProfile) && slices.Equal(left.TaskEvaluationCases, right.TaskEvaluationCases) && reflect.DeepEqual(left.TaskEvaluationExecution, right.TaskEvaluationExecution) } func (r *WorkRepository) LoadDispatchIntent(ctx context.Context, dispatchID string) (work.AgentDispatchIntent, error) { var intent work.AgentDispatchIntent - err := r.pool.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids FROM work.agent_dispatch_intents WHERE dispatch_id=$1`, dispatchID).Scan(&intent.DispatchID, &intent.IdempotencyKey, &intent.DelegationID, &intent.RunID, &intent.ImplementationID, &intent.ImplementationRevision, &intent.SourceRevision, &intent.WorktreeID, &intent.VerificationPlanRef, &intent.CreatedAtUnixMillis, &intent.SpecialistPurpose, &intent.TaskEvaluationProfileID, &intent.TaskEvaluationBindingID, &intent.TaskEvaluationCaseIDs) + var profileJSON, casesJSON, executionJSON []byte + err := r.pool.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids,task_evaluation_profile,task_evaluation_cases,task_evaluation_execution FROM work.agent_dispatch_intents WHERE dispatch_id=$1`, dispatchID).Scan(&intent.DispatchID, &intent.IdempotencyKey, &intent.DelegationID, &intent.RunID, &intent.ImplementationID, &intent.ImplementationRevision, &intent.SourceRevision, &intent.WorktreeID, &intent.VerificationPlanRef, &intent.CreatedAtUnixMillis, &intent.SpecialistPurpose, &intent.TaskEvaluationProfileID, &intent.TaskEvaluationBindingID, &intent.TaskEvaluationCaseIDs, &profileJSON, &casesJSON, &executionJSON) if err != nil { return work.AgentDispatchIntent{}, fmt.Errorf("load dispatch intent: %w", err) } + if err := unmarshalTaskPayload(profileJSON, casesJSON, &intent.TaskEvaluationProfile, &intent.TaskEvaluationCases); err != nil { + return work.AgentDispatchIntent{}, err + } + if len(executionJSON) > 0 && string(executionJSON) != "null" { + if err := json.Unmarshal(executionJSON, &intent.TaskEvaluationExecution); err != nil { + return work.AgentDispatchIntent{}, err + } + } return intent, nil } @@ -200,7 +229,11 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph return err } for _, node := range graph.Nodes { - if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.DispatchID, node.IdempotencyKey, node.State, node.Attempts, node.Purpose, node.TaskProfileID, node.BindingID, node.CaseIDs); err != nil { + profileJSON, casesJSON, err := marshalTaskPayload(node.TaskProfile, node.TaskCases) + if err != nil { + return err + } + if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids,task_evaluation_profile,task_evaluation_cases) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.DispatchID, node.IdempotencyKey, node.State, node.Attempts, node.Purpose, node.TaskProfileID, node.BindingID, node.CaseIDs, profileJSON, casesJSON); err != nil { return err } } @@ -218,13 +251,18 @@ func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revi if err != nil { return work.WorkGraph{}, fmt.Errorf("load graph: %w", err) } - rows, err := r.pool.Query(ctx, `SELECT node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision) + rows, err := r.pool.Query(ctx, `SELECT node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids,task_evaluation_profile,task_evaluation_cases FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision) if err != nil { return work.WorkGraph{}, err } for rows.Next() { var node work.GraphNode - if err := rows.Scan(&node.ID, &node.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts, &node.Purpose, &node.TaskProfileID, &node.BindingID, &node.CaseIDs); err != nil { + var profileJSON, casesJSON []byte + if err := rows.Scan(&node.ID, &node.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts, &node.Purpose, &node.TaskProfileID, &node.BindingID, &node.CaseIDs, &profileJSON, &casesJSON); err != nil { + rows.Close() + return work.WorkGraph{}, err + } + if err := unmarshalTaskPayload(profileJSON, casesJSON, &node.TaskProfile, &node.TaskCases); err != nil { rows.Close() return work.WorkGraph{}, err } @@ -570,3 +608,32 @@ func latestRowErr(err error) error { } return err } + +func marshalTaskPayload(profile *capability.TaskEvaluationProfile, cases []capability.TaskEvaluationCase) ([]byte, []byte, error) { + var profileJSON []byte + var err error + if profile != nil { + profileJSON, err = json.Marshal(profile) + if err != nil { + return nil, nil, err + } + } + casesJSON, err := json.Marshal(cases) + return profileJSON, casesJSON, err +} + +func unmarshalTaskPayload(profileJSON, casesJSON []byte, profile **capability.TaskEvaluationProfile, cases *[]capability.TaskEvaluationCase) error { + if len(profileJSON) > 0 && string(profileJSON) != "null" { + var value capability.TaskEvaluationProfile + if err := json.Unmarshal(profileJSON, &value); err != nil { + return err + } + *profile = &value + } + if len(casesJSON) > 0 && string(casesJSON) != "null" { + if err := json.Unmarshal(casesJSON, cases); err != nil { + return err + } + } + return nil +} diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go index 33e8717..59c3892 100644 --- a/cmd/agent-worker/main.go +++ b/cmd/agent-worker/main.go @@ -49,9 +49,8 @@ func main() { os.Exit(1) } } else { - executor = newLocalFallbackExecutor() - localFallback = true - fmt.Fprintln(os.Stderr, "protected agent execution bridge URL is not configured; local fallback preserves durable work without external mutation") + executor = newTaskEvaluationExecutor(envOr("COGNITIVE_WORKER_URL", "http://cognitive-worker:8083"), envOr("WORK_GRAPH_RUNTIME_URL", "http://work-graph-runtime:8088"), &http.Client{Timeout: *leaseTTL}) + fmt.Fprintln(os.Stderr, "protected agent execution bridge URL is not configured; task evaluations use the local durable evaluator, other work remains in fallback") } go func() { if serveErr := http.ListenAndServe(*statusAddr, workerStatusHandler(executor != nil && !localFallback, localFallback)); serveErr != nil { @@ -107,6 +106,13 @@ func loadProtectedBridgeURLFrom(root, path string) (string, bool) { return value, true } +func envOr(name, fallback string) string { + if value := strings.TrimSpace(os.Getenv(name)); value != "" { + return value + } + return fallback +} + func resolveDatabasePassword(name, refName string, consumer secretcontracts.RuntimeConsumer) (string, error) { if _, ok := os.LookupEnv(name); ok { return "", fmt.Errorf("%s direct environment secret is forbidden", name) diff --git a/cmd/agent-worker/task_evaluation_executor.go b/cmd/agent-worker/task_evaluation_executor.go new file mode 100644 index 0000000..2dcd877 --- /dev/null +++ b/cmd/agent-worker/task_evaluation_executor.go @@ -0,0 +1,92 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "git.cin.su/m/Assistant/contracts/artifact" + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" + capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain" +) + +type taskEvaluationExecutor struct { + cognitiveURL string + callbackURL string + client *http.Client + fallback localFallbackExecutor +} + +func newTaskEvaluationExecutor(cognitiveURL, callbackURL string, client *http.Client) taskEvaluationExecutor { + if client == nil { + client = http.DefaultClient + } + return taskEvaluationExecutor{cognitiveURL: strings.TrimRight(cognitiveURL, "/"), callbackURL: strings.TrimRight(callbackURL, "/"), client: client, fallback: newLocalFallbackExecutor()} +} + +func (e taskEvaluationExecutor) Execute(ctx context.Context, request work.AgentExecutionRequest) (work.AgentExecutionResult, error) { + intent := request.Intent + if intent.TaskEvaluationExecution == nil || intent.TaskEvaluationProfile == nil || len(intent.TaskEvaluationCases) == 0 { + return e.fallback.Execute(ctx, request) + } + execution := intent.TaskEvaluationExecution + if execution.GraphID == "" || execution.NodeID == "" || execution.BindingRevision < 1 || execution.ActorID == "" || execution.PrincipalID == "" || execution.ScopeID == "" { + return e.fallback.Execute(ctx, request) + } + if e.cognitiveURL == "" || e.callbackURL == "" { + return e.fallback.Execute(ctx, request) + } + payload := capability.TaskEvaluationExecutionRequest{Profile: *intent.TaskEvaluationProfile, CapabilityID: intent.TaskEvaluationProfile.CapabilityID, BindingID: intent.TaskEvaluationBindingID, BindingRevision: execution.BindingRevision, ActorID: execution.ActorID, PrincipalID: execution.PrincipalID, ScopeID: execution.ScopeID, Cases: intent.TaskEvaluationCases, MaxOutputBytes: 4096} + response, err := postJSON[capability.TaskEvaluationExecutionResponse](ctx, e.client, e.cognitiveURL+"/v1/task-evaluations/execute", payload) + if err != nil { + return work.AgentExecutionResult{}, err + } + if len(response.Observations) == 0 || response.ObservedAtUnixMillis < 1 { + return work.AgentExecutionResult{}, fmt.Errorf("task evaluation returned no observations") + } + if len(intent.TaskEvaluationCases) == 0 { + return work.AgentExecutionResult{}, fmt.Errorf("task evaluation cases are missing") + } + evaluation, err := capabilitydomain.BuildTaskEvaluation(*intent.TaskEvaluationProfile, capability.TaskEvaluationID("evaluation:"+intent.DispatchID), response.BindingID, response.BindingRevision, intent.TaskEvaluationProfile.CapabilityID, response.Observations, artifact.ArtifactID(intent.TaskEvaluationCases[0].EvidenceArtifactRef), response.ObservedAtUnixMillis) + if err != nil { + return work.AgentExecutionResult{}, err + } + callback := work.TaskEvaluationEvidenceCallback{GraphID: execution.GraphID, Revision: execution.GraphRevision, NodeID: execution.NodeID, Evaluation: evaluation, Succeeded: true} + if _, err := postJSON[json.RawMessage](ctx, e.client, e.callbackURL+"/v1/work-graphs/task-evaluations/callback", callback); err != nil { + return work.AgentExecutionResult{}, err + } + return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "task evaluation executed, aggregated and accepted by Work Graph callback", OutcomeRef: string(evaluation.ID), ArtifactRefs: []string{string(evaluation.ArtifactID)}, ObservedAtUnixMillis: time.Now().UnixMilli()}, nil +} + +func postJSON[T any](ctx context.Context, client *http.Client, url string, payload any) (T, error) { + var result T + body, err := json.Marshal(payload) + if err != nil { + return result, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return result, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + return result, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return result, fmt.Errorf("POST %s: status %s", url, resp.Status) + } + if resp.StatusCode == http.StatusNoContent { + return result, nil + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return result, err + } + return result, nil +} diff --git a/cmd/agent-worker/task_evaluation_executor_test.go b/cmd/agent-worker/task_evaluation_executor_test.go new file mode 100644 index 0000000..b4c0d7b --- /dev/null +++ b/cmd/agent-worker/task_evaluation_executor_test.go @@ -0,0 +1,38 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" +) + +func TestTaskEvaluationExecutorInvokesAggregatesAndCallbacks(t *testing.T) { + profile := capability.TaskEvaluationProfile{ID: "profile-1", Revision: 1, CapabilityID: "capability-1", Stage: capability.TaskEvaluationGeneral, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-1"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/task-evaluations/execute" { + _ = json.NewEncoder(w).Encode(capability.TaskEvaluationExecutionResponse{ProfileID: profile.ID, BindingID: "binding-1", BindingRevision: 3, Observations: []capability.TaskCaseObservation{{CaseID: "case-1", Result: capability.EvaluationSupported, EvidenceArtifactRef: "artifact-1"}}, ObservedAtUnixMillis: 10}) + return + } + if r.URL.Path == "/v1/work-graphs/task-evaluations/callback" { + var callback work.TaskEvaluationEvidenceCallback + if err := json.NewDecoder(r.Body).Decode(&callback); err != nil || callback.GraphID != "graph-1" || callback.NodeID != "node-1" || callback.Evaluation.Result != capability.EvaluationSupported { + http.Error(w, "invalid callback", http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusNoContent) + return + } + http.NotFound(w, r) + })) + defer server.Close() + executor := newTaskEvaluationExecutor(server.URL, server.URL, server.Client()) + result, err := executor.Execute(context.Background(), work.AgentExecutionRequest{Intent: work.AgentDispatchIntent{DispatchID: "dispatch-1", TaskEvaluationBindingID: "binding-1", TaskEvaluationProfile: &profile, TaskEvaluationCases: []capability.TaskEvaluationCase{{CaseID: "case-1", Prompt: "answer", EvidenceArtifactRef: "artifact-1"}}, TaskEvaluationExecution: &work.TaskEvaluationDispatchContext{GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", BindingRevision: 3, ActorID: "actor-1", PrincipalID: "principal-1", ScopeID: "scope-1"}}}) + if err != nil || result.State != work.AgentExecutionCompleted || result.OutcomeRef == "" { + t.Fatalf("result=%+v err=%v", result, err) + } +} diff --git a/contracts/work/graph.go b/contracts/work/graph.go index b4b9623..0ee1dd9 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -46,6 +46,8 @@ type GraphNode struct { TaskProfileID capability.TaskEvaluationProfileID `json:"task_profile_id,omitempty"` BindingID capability.CapabilityBindingID `json:"binding_id,omitempty"` CaseIDs []string `json:"case_ids,omitempty"` + TaskProfile *capability.TaskEvaluationProfile `json:"task_profile,omitempty"` + TaskCases []capability.TaskEvaluationCase `json:"task_cases,omitempty"` } type GraphEdge struct { @@ -79,6 +81,11 @@ type TaskEvaluationGraphRequest struct { ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"` ParentGraphState GraphState `json:"parent_graph_state,omitempty"` TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"` + Cases []capability.TaskEvaluationCase `json:"cases,omitempty"` + BindingRevision int64 `json:"binding_revision,omitempty"` + ActorID string `json:"actor_id,omitempty"` + PrincipalID string `json:"principal_id,omitempty"` + ScopeID string `json:"scope_id,omitempty"` } // TaskEvaluationEvidenceCallback is emitted by a specialist after it has diff --git a/contracts/work/lease.go b/contracts/work/lease.go index 25351cd..825280e 100644 --- a/contracts/work/lease.go +++ b/contracts/work/lease.go @@ -3,7 +3,9 @@ package work import ( "time" + "git.cin.su/m/Assistant/contracts/authority" "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/identity" ) type AgentLeaseID string @@ -68,4 +70,17 @@ type AgentDispatchIntent struct { TaskEvaluationProfileID capability.TaskEvaluationProfileID `json:"task_evaluation_profile_id,omitempty"` TaskEvaluationBindingID capability.CapabilityBindingID `json:"task_evaluation_binding_id,omitempty"` TaskEvaluationCaseIDs []string `json:"task_evaluation_case_ids,omitempty"` + TaskEvaluationProfile *capability.TaskEvaluationProfile `json:"task_evaluation_profile,omitempty"` + TaskEvaluationCases []capability.TaskEvaluationCase `json:"task_evaluation_cases,omitempty"` + TaskEvaluationExecution *TaskEvaluationDispatchContext `json:"task_evaluation_execution,omitempty"` +} + +type TaskEvaluationDispatchContext struct { + GraphID string `json:"graph_id"` + GraphRevision int64 `json:"graph_revision"` + NodeID GraphNodeID `json:"node_id"` + BindingRevision int64 `json:"binding_revision"` + ActorID identity.ActorID `json:"actor_id"` + PrincipalID identity.PrincipalID `json:"principal_id"` + ScopeID authority.ScopeID `json:"scope_id"` } diff --git a/database/migrations/041_task_evaluation_payload.sql b/database/migrations/041_task_evaluation_payload.sql new file mode 100644 index 0000000..83228eb --- /dev/null +++ b/database/migrations/041_task_evaluation_payload.sql @@ -0,0 +1,16 @@ +BEGIN; + +SET ROLE work_role; + +ALTER TABLE work.agent_dispatch_intents + ADD COLUMN IF NOT EXISTS task_evaluation_profile JSONB, + ADD COLUMN IF NOT EXISTS task_evaluation_cases JSONB NOT NULL DEFAULT '[]', + ADD COLUMN IF NOT EXISTS task_evaluation_execution JSONB; + +ALTER TABLE work.graph_nodes + ADD COLUMN IF NOT EXISTS task_evaluation_profile JSONB, + ADD COLUMN IF NOT EXISTS task_evaluation_cases JSONB NOT NULL DEFAULT '[]', + ADD COLUMN IF NOT EXISTS task_evaluation_execution JSONB; + +RESET ROLE; +COMMIT; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 1737f99..eb32e6f 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000212-task-evaluator-invocation.json", - "task_id": "DEV-000212", + "task_file": "dev/tasks/DEV-000213-worker-evaluator-link.json", + "task_id": "DEV-000213", "task_revision": 1 } diff --git a/dev/tasks/DEV-000213-worker-evaluator-link.json b/dev/tasks/DEV-000213-worker-evaluator-link.json new file mode 100644 index 0000000..24d3ead --- /dev/null +++ b/dev/tasks/DEV-000213-worker-evaluator-link.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000213", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "a86d1994c3f48db86592eb87fb54533297e0671c", + "risk_tier": "E", + "intent": "Carry the complete bounded task-evaluation profile and case payload through durable Work Graph dispatch so agent-worker can invoke Cognitive Worker, aggregate observations, and return the existing Work Graph callback after restart-safe recovery.", + "requirement_ids": ["dispatch-to-evaluator link", "payload durability", "worker recovery", "observation aggregation", "callback integration"], + "allowed_change_scope": {"modules": ["work", "agent-worker"], "paths": ["contracts/work/**", "modules/work/**", "adapters/postgres/work_repository.go", "cmd/agent-worker/**", "cmd/work-graph-runtime/**", "database/migrations/041_task_evaluation_payload.sql", "tests/agent_execution/**", "tests/model_qualification/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000213-worker-evaluator-link.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "modules/capability/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["TaskEvaluationExecutionRequest/Response", "Work Graph callback", "existing AgentExecutor and recovery contract"], "may_change": ["typed dispatch payload and worker evaluator executor"]}, + "semantic_owner": "work owns dispatch lifecycle; capability owns evaluation evidence; cognitive-worker owns provider invocation", + "canonical_state": ["work dispatch payload is durable input; Capability remains evidence owner"], + "state_transitions": ["task graph node -> durable payload dispatch", "worker claim -> evaluator invocation", "response -> aggregate TaskEvaluation", "aggregate -> Work Graph evidence callback"], + "side_effects_and_privileges": ["HTTP calls to internal cognitive/core/work boundaries; no direct foreign DB and no new secret"], + "persistence_and_migrations": ["041 adds JSONB payload columns to work graph nodes and dispatch intents"], + "failure_retry_restart": ["payload is loaded after stale claim recovery; evaluator failure leaves dispatch retryable", "callback only follows a persisted aggregate; unknown remains unknown"], + "assumptions": ["Cognitive Worker endpoint is available as optional local execution path", "external bridge remains preferred for non-evaluation agent tasks"], + "acceptance_scenarios": ["profile/cases survive save-load", "task dispatch invokes evaluator with original prompt payload", "all observations aggregate deterministically", "callback receives matching binding/profile/artifact", "evaluator or callback failure does not complete dispatch", "legacy dispatch remains unchanged"], + "required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "migration/schema checks", "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 7e3289b..fdfa795 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -5,6 +5,9 @@ import ( "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" @@ -157,6 +160,13 @@ func AdmitAndScheduleTaskEvaluation(ctx context.Context, store ports.GraphStore, 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 } } diff --git a/modules/work/application/task_evaluation_dispatch_test.go b/modules/work/application/task_evaluation_dispatch_test.go index a9bd9fb..1d51727 100644 --- a/modules/work/application/task_evaluation_dispatch_test.go +++ b/modules/work/application/task_evaluation_dispatch_test.go @@ -33,6 +33,7 @@ func TestAdmitAndScheduleTaskEvaluationCarriesSpecialistContext(t *testing.T) { request := TaskEvaluationScheduleRequest{ Plan: work.TaskEvaluationGraphRequest{ GraphID: "graph-task", Profile: capability.TaskEvaluationProfile{ID: "profile-general", Revision: 1, CapabilityID: "capability-chat", Stage: capability.TaskEvaluationGeneral, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-a", "case-b"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100}, + Cases: []capability.TaskEvaluationCase{{CaseID: "case-a", Prompt: "one", EvidenceArtifactRef: "artifact-a"}, {CaseID: "case-b", Prompt: "two", EvidenceArtifactRef: "artifact-b"}}, BindingIDs: []capability.CapabilityBindingID{"binding-a"}, DispatchIDs: []string{"dispatch-a"}, IdempotencyPrefix: "eval", Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 2}, }, Intents: []work.AgentDispatchIntent{{DispatchID: "dispatch-a", IdempotencyKey: "idem-a", DelegationID: "delegation-a", RunID: "run-a", ImplementationID: "impl", ImplementationRevision: 1, SourceRevision: "source", WorktreeID: "tree", VerificationPlanRef: "verify", CreatedAtUnixMillis: 1}}, diff --git a/modules/work/domain/task_evaluation_graph.go b/modules/work/domain/task_evaluation_graph.go index a034174..712fbcf 100644 --- a/modules/work/domain/task_evaluation_graph.go +++ b/modules/work/domain/task_evaluation_graph.go @@ -20,6 +20,24 @@ func BuildTaskEvaluationGraph(request work.TaskEvaluationGraphRequest) (work.Wor if err := request.Profile.Validate(); err != nil { return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph } + if len(request.Cases) != len(request.Profile.RequiredCaseIDs) { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + seenCases := make(map[string]struct{}, len(request.Cases)) + for _, item := range request.Cases { + if item.CaseID == "" || item.Prompt == "" || item.EvidenceArtifactRef == "" { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + if _, exists := seenCases[item.CaseID]; exists { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + seenCases[item.CaseID] = struct{}{} + } + for _, requiredID := range request.Profile.RequiredCaseIDs { + if _, exists := seenCases[requiredID]; !exists { + return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph + } + } if request.Profile.Stage == capability.TaskEvaluationGeneral { if request.ParentGraphID != "" || request.ParentGraphRevision != 0 || request.ParentGraphState != "" || request.TriggerEvidenceRef != "" { return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph @@ -56,7 +74,15 @@ func BuildTaskEvaluationGraph(request work.TaskEvaluationGraphRequest) (work.Wor TaskProfileID: request.Profile.ID, BindingID: bindingID, CaseIDs: append([]string(nil), request.Profile.RequiredCaseIDs...), + TaskProfile: cloneTaskEvaluationProfile(request.Profile), + TaskCases: append([]capability.TaskEvaluationCase(nil), request.Cases...), }) } 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 } + +func cloneTaskEvaluationProfile(value capability.TaskEvaluationProfile) *capability.TaskEvaluationProfile { + copyValue := value + copyValue.RequiredCaseIDs = append([]string(nil), value.RequiredCaseIDs...) + return ©Value +} diff --git a/modules/work/domain/task_evaluation_graph_test.go b/modules/work/domain/task_evaluation_graph_test.go index becd2da..c943bec 100644 --- a/modules/work/domain/task_evaluation_graph_test.go +++ b/modules/work/domain/task_evaluation_graph_test.go @@ -9,7 +9,7 @@ import ( 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}} + 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}, Cases: []capability.TaskEvaluationCase{{CaseID: "case-a", Prompt: "short answer", EvidenceArtifactRef: "artifact-a"}, {CaseID: "case-b", Prompt: "long answer", EvidenceArtifactRef: "artifact-b"}}} if stage == capability.TaskEvaluationAdditional { request.Profile.ParentProfileID = "profile-general" request.Profile.TriggerEvidenceRef = "artifact:unresolved-1"