Assistant/cmd/cognitive-worker/task_evaluation.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"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
	}

	caseSet := make(map[string]struct{}, len(request.Profile.RequiredCaseIDs))
	seenCases := make(map[string]struct{}, len(request.Cases))
	for _, id := range request.Profile.RequiredCaseIDs {
		caseSet[id] = struct{}{}
	}
	for _, 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{}{}
	}
	if len(seenCases) != len(caseSet) {
		http.Error(w, "task evaluation case set is incomplete", http.StatusBadRequest)
		return
	}

	// One bounded fan-out wave uses one binding snapshot. If that candidate is
	// unavailable, only this binding is excluded and the existing resolver
	// chooses the next eligible candidate. Provider output remains untrusted
	// until the independent verifier accepts it downstream.
	requirement := capability.CognitiveRequirement{CapabilityID: request.CapabilityID, OwnerPrincipalID: request.PrincipalID, DataPolicy: selected.Binding.DataPolicy, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, AllowCanary: g.allowCanary, MaxOutputBytes: request.MaxOutputBytes}
	var observations []capability.TaskCaseObservation
	var responses []capability.TaskCaseResponse
	for attempt := 0; attempt < 3; attempt++ {
		provider := g.provider(selected)
		if provider == nil {
			observations, responses = taskEvaluationUnavailable(request.Cases)
		} else {
			observations, responses = g.executeTaskEvaluationCases(r.Context(), request, selected, provider)
		}
		if taskEvaluationResponsesUsable(responses) {
			break
		}
		requirement.ExcludedBindingIDs = append(requirement.ExcludedBindingIDs, selected.Binding.ID)
		fallback, resolveErr := g.resolver.Resolve(r.Context(), requirement)
		fallbackProvider := g.provider(fallback)
		if resolveErr != nil || fallbackProvider == nil || fallback.Binding.ID == selected.Binding.ID || (fallback.Binding.DataPolicy != capability.DataPolicyLocalOnly && !request.EgressAcknowledged) {
			break
		}
		selected = fallback
	}

	writeTaskEvaluationJSON(w, capability.TaskEvaluationExecutionResponse{ProfileID: request.Profile.ID, BindingID: selected.Binding.ID, BindingRevision: selected.Binding.Revision, Observations: observations, Responses: responses, ObservedAtUnixMillis: time.Now().UnixMilli()})
}

func (g cognitiveGateway) executeTaskEvaluationCases(ctx context.Context, request capability.TaskEvaluationExecutionRequest, selected capability.ResolvedCognitiveBinding, provider advisoryProvider) ([]capability.TaskCaseObservation, []capability.TaskCaseResponse) {
	observations := make([]capability.TaskCaseObservation, len(request.Cases))
	responses := make([]capability.TaskCaseResponse, len(request.Cases))
	var wg sync.WaitGroup
	for i, item := range request.Cases {
		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, Locale: "en-US", 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(ctx, string(invocation.RequestID), provider, invocation, int(request.MaxOutputBytes))
			latency := time.Since(started).Microseconds()
			observations[index] = capability.TaskCaseObservation{CaseID: item.CaseID, Result: capability.EvaluationUnknown, CriticalFailure: invokeErr != nil || response.FinishReason != cognitive.FinishCompleted || strings.TrimSpace(response.AdvisoryText) == "", EvidenceArtifactRef: item.EvidenceArtifactRef, LatencyMicros: latency}
			responses[index] = capability.TaskCaseResponse{CaseID: item.CaseID, ResponseText: response.AdvisoryText, LatencyMicros: latency, EvidenceArtifactRef: item.EvidenceArtifactRef}
		}(i, item)
	}
	wg.Wait()
	return observations, responses
}

func taskEvaluationUnavailable(cases []capability.TaskEvaluationCase) ([]capability.TaskCaseObservation, []capability.TaskCaseResponse) {
	observations := make([]capability.TaskCaseObservation, len(cases))
	responses := make([]capability.TaskCaseResponse, len(cases))
	for i, item := range cases {
		observations[i] = capability.TaskCaseObservation{CaseID: item.CaseID, Result: capability.EvaluationUnknown, CriticalFailure: true, EvidenceArtifactRef: item.EvidenceArtifactRef}
		responses[i] = capability.TaskCaseResponse{CaseID: item.CaseID, EvidenceArtifactRef: item.EvidenceArtifactRef}
	}
	return observations, responses
}

func taskEvaluationResponsesUsable(responses []capability.TaskCaseResponse) bool {
	if len(responses) == 0 {
		return false
	}
	for _, response := range responses {
		if strings.TrimSpace(response.ResponseText) == "" {
			return false
		}
	}
	return true
}

func writeTaskEvaluationJSON(w http.ResponseWriter, value any) {
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(value)
}