diff --git a/adapters/postgres/capability_repository.go b/adapters/postgres/capability_repository.go index 4799746..8a7fb04 100644 --- a/adapters/postgres/capability_repository.go +++ b/adapters/postgres/capability_repository.go @@ -437,6 +437,24 @@ func (r *CapabilityRepository) LoadCognitiveCandidatesForProfile(ctx context.Con return r.loadCognitiveCandidates(ctx, capabilityID, owner, profileID) } +func (r *CapabilityRepository) LoadCognitiveCandidate(ctx context.Context, bindingID capability.CapabilityBindingID, revision int64) (capability.ResolvedCognitiveBinding, error) { + var capabilityID capability.CapabilityID + var owner identity.PrincipalID + if err := r.pool.QueryRow(ctx, `SELECT capability_id,owner_principal_id FROM capability.cognitive_bindings WHERE binding_id=$1 AND revision=$2`, bindingID, revision).Scan(&capabilityID, &owner); err != nil { + return capability.ResolvedCognitiveBinding{}, err + } + candidates, err := r.loadCognitiveCandidates(ctx, capabilityID, owner, "") + if err != nil { + return capability.ResolvedCognitiveBinding{}, err + } + for _, candidate := range candidates { + if candidate.Binding.ID == bindingID && candidate.Binding.Revision == revision { + return candidate, nil + } + } + return capability.ResolvedCognitiveBinding{}, pgx.ErrNoRows +} + func (r *CapabilityRepository) loadCognitiveCandidates(ctx context.Context, capabilityID capability.CapabilityID, owner identity.PrincipalID, profileID capability.TaskEvaluationProfileID) ([]capability.ResolvedCognitiveBinding, error) { rows, err := r.pool.Query(ctx, `SELECT b.binding_id,b.revision,b.capability_id,b.owner_principal_id,b.provider_instance_id,b.provider_revision,b.model_descriptor_id,b.model_revision,b.data_policy,b.qualification,b.activation,b.health,b.limit_policy_ref, diff --git a/cmd/cognitive-worker/gateway.go b/cmd/cognitive-worker/gateway.go index aa9d05f..26731c3 100644 --- a/cmd/cognitive-worker/gateway.go +++ b/cmd/cognitive-worker/gateway.go @@ -26,6 +26,10 @@ type bindingResolver interface { RegisterDiscovery(context.Context, capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error) } +type candidateBindingResolver interface { + ResolveCandidate(context.Context, capability.CapabilityBindingID, int64) (capability.ResolvedCognitiveBinding, error) +} + type admissionClient interface { Admit(context.Context, string, capability.AdmissionRequest) (capability.AdmissionDecision, capability.AdmissionLease, error) Release(context.Context, capability.AdmissionLease) error @@ -341,6 +345,26 @@ func (r *httpBindingResolver) Resolve(ctx context.Context, value capability.Cogn return result, err } +func (r *httpBindingResolver) ResolveCandidate(ctx context.Context, bindingID capability.CapabilityBindingID, revision int64) (capability.ResolvedCognitiveBinding, error) { + var result capability.ResolvedCognitiveBinding + target := *r.base + target.Path = "/v1/cognitive-bindings/" + url.PathEscape(string(bindingID)) + "/revisions/" + fmt.Sprint(revision) + "/snapshot" + request, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil) + if err != nil { + return result, err + } + response, err := r.client.Do(request) + if err != nil { + return result, err + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return result, fmt.Errorf("candidate status %d", response.StatusCode) + } + err = json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(&result) + return result, err +} + func (r *httpBindingResolver) Observe(ctx context.Context, value capability.CognitiveUsageObservation) error { return r.post(ctx, "/v1/cognitive-usage-observations", value, nil) } diff --git a/cmd/cognitive-worker/main.go b/cmd/cognitive-worker/main.go index 17c5b3a..b8fcf34 100644 --- a/cmd/cognitive-worker/main.go +++ b/cmd/cognitive-worker/main.go @@ -166,6 +166,7 @@ func main() { mux.HandleFunc("POST /v1/advisory-policy", gateway.policyHandler) mux.HandleFunc("POST /v1/advisories", gateway.advisoryHandler) mux.HandleFunc("POST /v1/interpretations", gateway.interpretationHandler) + mux.HandleFunc("POST /v1/task-evaluations/execute", gateway.taskEvaluationHandler) discovery := discoveryService{resolver: resolver, providers: discoveries, owner: identity.PrincipalID(envOr("COGNITIVE_OWNER_PRINCIPAL_ID", "domovoy-creator-principal")), now: time.Now} mux.HandleFunc("POST /v1/provider-discovery/{provider}", discovery.handler) refreshProviderDiscoveries(resolver, discovery) diff --git a/cmd/cognitive-worker/task_evaluation.go b/cmd/cognitive-worker/task_evaluation.go new file mode 100644 index 0000000..e528e28 --- /dev/null +++ b/cmd/cognitive-worker/task_evaluation.go @@ -0,0 +1,93 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "sync" + "time" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/cognitive" +) + +func (g cognitiveGateway) taskEvaluationHandler(w http.ResponseWriter, r *http.Request) { + var request capability.TaskEvaluationExecutionRequest + if !decodeJSON(w, r, 256<<10, &request) { + return + } + if err := request.Profile.Validate(); err != nil || request.CapabilityID == "" || request.CapabilityID != request.Profile.CapabilityID || request.BindingID == "" || request.BindingRevision < 1 || request.ActorID == "" || request.PrincipalID == "" || request.ScopeID == "" || request.MaxOutputBytes < 1 || len(request.Cases) != len(request.Profile.RequiredCaseIDs) { + http.Error(w, "invalid task evaluation execution request", http.StatusBadRequest) + return + } + resolver, ok := g.resolver.(candidateBindingResolver) + if !ok { + http.Error(w, "exact candidate resolver unavailable", http.StatusServiceUnavailable) + return + } + selected, err := resolver.ResolveCandidate(r.Context(), request.BindingID, request.BindingRevision) + if err != nil || selected.Binding.ID != request.BindingID || selected.Binding.Revision != request.BindingRevision || selected.Binding.CapabilityID != request.CapabilityID || selected.Binding.OwnerPrincipalID != request.PrincipalID { + http.Error(w, "cognitive binding snapshot unavailable", http.StatusConflict) + return + } + if selected.Binding.DataPolicy != capability.DataPolicyLocalOnly && !request.EgressAcknowledged { + http.Error(w, "external egress acknowledgement required", http.StatusBadRequest) + return + } + provider := g.provider(selected) + if provider == nil { + http.Error(w, "cognitive provider unavailable", http.StatusServiceUnavailable) + return + } + caseSet := make(map[string]struct{}, len(request.Profile.RequiredCaseIDs)) + for _, id := range request.Profile.RequiredCaseIDs { + caseSet[id] = struct{}{} + } + observations := make([]capability.TaskCaseObservation, len(request.Cases)) + var wg sync.WaitGroup + seenCases := make(map[string]struct{}, len(request.Cases)) + for i, item := range request.Cases { + if item.CaseID == "" || item.Prompt == "" || item.EvidenceArtifactRef == "" { + http.Error(w, "invalid task evaluation case", http.StatusBadRequest) + return + } + if _, exists := caseSet[item.CaseID]; !exists { + http.Error(w, "task evaluation case is not in profile", http.StatusBadRequest) + return + } + if _, exists := seenCases[item.CaseID]; exists { + http.Error(w, "duplicate task evaluation case", http.StatusBadRequest) + return + } + seenCases[item.CaseID] = struct{}{} + wg.Add(1) + go func(index int, item capability.TaskEvaluationCase) { + defer wg.Done() + started := time.Now() + invocation := cognitive.AdvisoryRequest{Version: cognitive.CreatorAdvisoryV1, RequestID: cognitive.RequestID(fmt.Sprintf("task-evaluation:%s:%s", request.Profile.ID, item.CaseID)), ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: item.Prompt, MaxOutputBytes: int(request.MaxOutputBytes), ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ProviderModelID: selected.Model.ProviderModelID} + response, invokeErr := g.invokeAdmitted(r.Context(), string(invocation.RequestID), provider, invocation, int(request.MaxOutputBytes)) + observation := capability.TaskCaseObservation{CaseID: item.CaseID, LatencyMicros: time.Since(started).Microseconds(), EvidenceArtifactRef: item.EvidenceArtifactRef} + if invokeErr == nil && response.FinishReason == cognitive.FinishCompleted && response.AdvisoryText != "" { + observation.Result = capability.EvaluationSupported + // Invocation success proves reachability/support for this case, + // not semantic answer quality. A verifier may score quality later. + observation.QualityScoreMillis = 0 + } else { + observation.Result = capability.EvaluationUnknown + observation.CriticalFailure = invokeErr != nil + } + observations[index] = observation + }(i, item) + } + if len(seenCases) != len(caseSet) { + http.Error(w, "task evaluation case set is incomplete", http.StatusBadRequest) + return + } + wg.Wait() + writeTaskEvaluationJSON(w, capability.TaskEvaluationExecutionResponse{ProfileID: request.Profile.ID, BindingID: selected.Binding.ID, BindingRevision: selected.Binding.Revision, Observations: observations, ObservedAtUnixMillis: time.Now().UnixMilli()}) +} + +func writeTaskEvaluationJSON(w http.ResponseWriter, value any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(value) +} diff --git a/cmd/cognitive-worker/task_evaluation_test.go b/cmd/cognitive-worker/task_evaluation_test.go new file mode 100644 index 0000000..a1ffa5b --- /dev/null +++ b/cmd/cognitive-worker/task_evaluation_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.cin.su/m/Assistant/contracts/authority" + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/cognitive" + "git.cin.su/m/Assistant/contracts/identity" +) + +type candidateRegistryStub struct { + registryStub + selected capability.ResolvedCognitiveBinding +} + +func (s *candidateRegistryStub) ResolveCandidate(context.Context, capability.CapabilityBindingID, int64) (capability.ResolvedCognitiveBinding, error) { + return s.selected, nil +} + +func TestTaskEvaluationHandlerInvokesEveryBoundedCase(t *testing.T) { + selected := gatewaySelection() + registry := &candidateRegistryStub{selected: selected} + provider := &stubAdvisoryProvider{response: cognitive.AdvisoryResponse{Version: cognitive.CreatorAdvisoryV1, FinishReason: cognitive.FinishCompleted, AdvisoryText: "observed response"}} + gateway := cognitiveGateway{ + resolver: registry, + providers: map[capability.ProviderKind][]providerRegistration{ + selected.Provider.Kind: {{ProviderInstanceID: selected.Provider.ID, Kind: selected.Provider.Kind, EndpointRef: selected.Provider.EndpointRef, CredentialRef: selected.Provider.CredentialRef, Provider: provider}}, + }, + now: func() time.Time { return time.UnixMilli(1000) }, + } + profile := capability.TaskEvaluationProfile{ID: "profile-general", Revision: 1, CapabilityID: selected.Binding.CapabilityID, Stage: capability.TaskEvaluationGeneral, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-a", "case-b"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100} + request := capability.TaskEvaluationExecutionRequest{Profile: profile, CapabilityID: selected.Binding.CapabilityID, BindingID: selected.Binding.ID, BindingRevision: selected.Binding.Revision, ActorID: identity.ActorID("actor-1"), PrincipalID: selected.Binding.OwnerPrincipalID, ScopeID: authority.ScopeID("scope-1"), MaxOutputBytes: 512, EgressAcknowledged: true, Cases: []capability.TaskEvaluationCase{{CaseID: "case-a", Prompt: "one", EvidenceArtifactRef: "artifact:a"}, {CaseID: "case-b", Prompt: "two", EvidenceArtifactRef: "artifact:b"}}} + payload, _ := json.Marshal(request) + recorder := httptest.NewRecorder() + gateway.taskEvaluationHandler(recorder, httptest.NewRequest(http.MethodPost, "/v1/task-evaluations/execute", bytes.NewReader(payload))) + if recorder.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + var response capability.TaskEvaluationExecutionResponse + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if len(response.Observations) != 2 || response.Observations[0].Result != capability.EvaluationSupported || response.Observations[1].Result != capability.EvaluationSupported { + t.Fatalf("unexpected observations: %+v", response.Observations) + } +} diff --git a/cmd/core-runtime/main.go b/cmd/core-runtime/main.go index abfbcb6..0f96a40 100644 --- a/cmd/core-runtime/main.go +++ b/cmd/core-runtime/main.go @@ -834,6 +834,20 @@ func main() { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(binding) }) + mux.HandleFunc("GET /v1/cognitive-bindings/{id}/revisions/{revision}/snapshot", func(w http.ResponseWriter, r *http.Request) { + revision, err := strconv.ParseInt(r.PathValue("revision"), 10, 64) + if err != nil || revision < 1 { + http.Error(w, "invalid cognitive binding revision", http.StatusBadRequest) + return + } + snapshot, err := capabilityRepository.LoadCognitiveCandidate(r.Context(), capability.CapabilityBindingID(r.PathValue("id")), revision) + if err != nil { + http.Error(w, "cognitive binding snapshot not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(snapshot) + }) mux.HandleFunc("POST /v1/cognitive-bindings/resolve", func(w http.ResponseWriter, r *http.Request) { var requirement capability.CognitiveRequirement decoder := json.NewDecoder(io.LimitReader(r.Body, 16*1024+1)) diff --git a/contracts/capability/task_evaluation_execution.go b/contracts/capability/task_evaluation_execution.go new file mode 100644 index 0000000..bf00cae --- /dev/null +++ b/contracts/capability/task_evaluation_execution.go @@ -0,0 +1,35 @@ +package capability + +import ( + "git.cin.su/m/Assistant/contracts/authority" + "git.cin.su/m/Assistant/contracts/identity" +) + +// TaskEvaluationCase carries one bounded prompt selected by the evaluator. +// The prompt is an input artifact, not a model-produced fact. +type TaskEvaluationCase struct { + CaseID string + Prompt string + EvidenceArtifactRef string +} + +type TaskEvaluationExecutionRequest struct { + Profile TaskEvaluationProfile + CapabilityID CapabilityID + BindingID CapabilityBindingID + BindingRevision int64 + ActorID identity.ActorID + PrincipalID identity.PrincipalID + ScopeID authority.ScopeID + Cases []TaskEvaluationCase + MaxOutputBytes int64 + EgressAcknowledged bool +} + +type TaskEvaluationExecutionResponse struct { + ProfileID TaskEvaluationProfileID + BindingID CapabilityBindingID + BindingRevision int64 + Observations []TaskCaseObservation + ObservedAtUnixMillis int64 +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 3a6fe8e..1737f99 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000211-task-evaluator-callback.json", - "task_id": "DEV-000211", + "task_file": "dev/tasks/DEV-000212-task-evaluator-invocation.json", + "task_id": "DEV-000212", "task_revision": 1 } diff --git a/dev/tasks/DEV-000212-task-evaluator-invocation.json b/dev/tasks/DEV-000212-task-evaluator-invocation.json new file mode 100644 index 0000000..16167a6 --- /dev/null +++ b/dev/tasks/DEV-000212-task-evaluator-invocation.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000212", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "dff6c91c8e08611d73231a04a1af67efeebed22b", + "risk_tier": "E", + "intent": "Execute bounded task-evaluation cases against an exact Capability binding through Cognitive Gateway and return typed observations for the Work Graph callback.", + "requirement_ids": ["actual model invocation", "exact binding evaluation", "bounded cases", "provider-neutral evidence", "unknown on provider failure"], + "allowed_change_scope": {"modules": ["capability", "core-runtime", "cognitive-worker"], "paths": ["contracts/capability/**", "adapters/postgres/capability_repository.go", "cmd/core-runtime/**", "cmd/cognitive-worker/**", "tests/model_qualification/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000212-task-evaluator-invocation.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "database/**", "modules/identity/**"]}, + "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": ["TaskEvaluationProfile", "TaskCaseObservation", "Cognitive Gateway provider boundary", "Capability cognitive binding snapshot boundary"], "may_change": ["task evaluation execution request/response"]}, + "semantic_owner": "capability owns evaluation meaning; cognitive-worker owns provider invocation adapter boundary", + "canonical_state": ["none; this task returns observations only and does not promote or persist activation"], + "state_transitions": ["execution request -> exact binding snapshot", "case -> provider invocation -> SUPPORTED or UNKNOWN observation", "all observations -> Work Graph callback input"], + "side_effects_and_privileges": ["provider calls only through existing cognitive adapter; no direct DB and no new secret access"], + "persistence_and_migrations": ["none; durable persistence remains callback/Core Capability owner"], + "failure_retry_restart": ["provider timeout/auth/unavailable maps to UNKNOWN observation with evidence reference; caller may retry as a new attempt", "partial case set is never returned as supported aggregate"], + "assumptions": ["specialist supplies case prompts and artifact references from the approved bounded profile", "quality scoring beyond bounded invocation is performed by a verifier or later evaluator and is not fabricated here"], + "acceptance_scenarios": ["exact binding snapshot is required", "all supplied cases are invoked", "successful non-empty response yields supported observation with latency", "provider failure yields unknown observation", "unknown is never silently converted to supported", "no provider or hardware field participates in scoring"], + "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"] +}