279 lines
11 KiB
Go
279 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/adapters/models"
|
|
"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"
|
|
"git.cin.su/m/Assistant/platform/ids"
|
|
)
|
|
|
|
type qualificationObserver interface {
|
|
ObserveQualification(context.Context, capability.ModelQualificationObservation) error
|
|
}
|
|
|
|
type httpQualificationObserver struct{ resolver *httpBindingResolver }
|
|
|
|
func (o httpQualificationObserver) ObserveQualification(ctx context.Context, value capability.ModelQualificationObservation) error {
|
|
if o.resolver == nil {
|
|
return fmt.Errorf("qualification observer resolver is nil")
|
|
}
|
|
return o.resolver.post(ctx, "/v1/cognitive-registry/model-qualification-observations", value, nil)
|
|
}
|
|
|
|
type modelQualificationRequest struct {
|
|
Prompt string `json:"prompt"`
|
|
MaxOutputBytes int `json:"max_output_bytes"`
|
|
ModelIDs []capability.ProviderModelID `json:"model_ids,omitempty"`
|
|
}
|
|
|
|
type modelQualificationResult struct {
|
|
ProviderModelID capability.ProviderModelID `json:"provider_model_id"`
|
|
Status string `json:"status"`
|
|
FinishReason cognitive.FinishReason `json:"finish_reason,omitempty"`
|
|
LatencyMillis int64 `json:"latency_millis"`
|
|
OutputObserved bool `json:"output_observed"`
|
|
QualityAssessed bool `json:"quality_assessed"`
|
|
Category capability.ModelRatingCategory `json:"category"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type modelQualificationResponse struct {
|
|
Provider capability.ProviderInstanceID `json:"provider"`
|
|
ObservedAtUnixMillis int64 `json:"observed_at_unix_millis"`
|
|
Results []modelQualificationResult `json:"results"`
|
|
}
|
|
|
|
const qualificationControlPlaneTimeout = 45 * time.Second
|
|
|
|
// qualificationCoordinator bounds duplicate control-plane probes within one
|
|
// cognitive-worker process. It is deliberately not canonical state: the
|
|
// Capability owner remains the sole owner of qualification evidence and
|
|
// activation. A restart may repeat an incomplete probe, but cannot create a
|
|
// second semantic writer or bypass independent evaluation.
|
|
type qualificationCoordinator struct {
|
|
mu sync.Mutex
|
|
flights map[string]*qualificationFlight
|
|
}
|
|
|
|
type qualificationFlight struct {
|
|
done chan struct{}
|
|
response modelQualificationResponse
|
|
err error
|
|
waiters int
|
|
}
|
|
|
|
func (c *qualificationCoordinator) do(ctx context.Context, key string, operation func() (modelQualificationResponse, error)) (modelQualificationResponse, bool, error) {
|
|
c.mu.Lock()
|
|
if c.flights == nil {
|
|
c.flights = make(map[string]*qualificationFlight)
|
|
}
|
|
if current, exists := c.flights[key]; exists {
|
|
current.waiters++
|
|
c.mu.Unlock()
|
|
select {
|
|
case <-current.done:
|
|
return current.response, false, current.err
|
|
case <-ctx.Done():
|
|
return modelQualificationResponse{}, false, ctx.Err()
|
|
}
|
|
}
|
|
current := &qualificationFlight{done: make(chan struct{})}
|
|
c.flights[key] = current
|
|
c.mu.Unlock()
|
|
|
|
current.response, current.err = operation()
|
|
close(current.done)
|
|
c.mu.Lock()
|
|
delete(c.flights, key)
|
|
c.mu.Unlock()
|
|
return current.response, true, current.err
|
|
}
|
|
|
|
type qualificationPhaseTrace struct {
|
|
correlationID string
|
|
started time.Time
|
|
}
|
|
|
|
func newQualificationPhaseTrace() qualificationPhaseTrace {
|
|
correlationID, err := ids.New()
|
|
if err != nil {
|
|
correlationID = ids.ID(fmt.Sprintf("qualification-%d", time.Now().UnixNano()))
|
|
}
|
|
return qualificationPhaseTrace{correlationID: string(correlationID), started: time.Now()}
|
|
}
|
|
|
|
func (t qualificationPhaseTrace) phase(name string, fields ...string) {
|
|
log.Printf("qualification_phase correlation_id=%s phase=%s elapsed_ms=%d %s", t.correlationID, name, time.Since(t.started).Milliseconds(), strings.Join(fields, " "))
|
|
}
|
|
|
|
func qualificationFlightKey(provider string, modelIDs []capability.ProviderModelID) string {
|
|
ids := make([]string, 0, len(modelIDs))
|
|
for _, modelID := range modelIDs {
|
|
ids = append(ids, string(modelID))
|
|
}
|
|
sort.Strings(ids)
|
|
return provider + "|" + strings.Join(ids, ",")
|
|
}
|
|
|
|
func modelQualificationHandler(providers map[string]providerRegistration, owner string, observers ...qualificationObserver) http.Handler {
|
|
coordinator := &qualificationCoordinator{}
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
trace := newQualificationPhaseTrace()
|
|
trace.phase("Q0", "event=request_accepted")
|
|
name := r.PathValue("provider")
|
|
registration, ok := providers[name]
|
|
if !ok || registration.Discovery == nil || registration.Provider == nil {
|
|
http.Error(w, "provider qualification not found", http.StatusNotFound)
|
|
return
|
|
}
|
|
var request modelQualificationRequest
|
|
if !decodeJSON(w, r, 16<<10, &request) || request.Prompt == "" {
|
|
return
|
|
}
|
|
if request.MaxOutputBytes < 1 {
|
|
request.MaxOutputBytes = 512
|
|
}
|
|
if request.MaxOutputBytes > 4096 {
|
|
http.Error(w, "max_output_bytes exceeds qualification bound", http.StatusBadRequest)
|
|
return
|
|
}
|
|
controlCtx, cancel := context.WithTimeout(r.Context(), qualificationControlPlaneTimeout)
|
|
defer cancel()
|
|
response, leader, err := coordinator.do(controlCtx, qualificationFlightKey(name, request.ModelIDs), func() (modelQualificationResponse, error) {
|
|
trace.phase("Q1", "event=binding_candidate_loaded")
|
|
modelsFound, probe := registration.Discovery.Discover(controlCtx)
|
|
if probe.Health != capability.HealthAvailable || len(modelsFound) == 0 {
|
|
return modelQualificationResponse{}, fmt.Errorf("provider unavailable for qualification")
|
|
}
|
|
if len(request.ModelIDs) > 0 {
|
|
modelsFound = slices.DeleteFunc(modelsFound, func(model models.DiscoveredModel) bool {
|
|
return !slices.Contains(request.ModelIDs, model.ProviderModelID)
|
|
})
|
|
}
|
|
if len(modelsFound) == 0 {
|
|
return modelQualificationResponse{}, fmt.Errorf("requested model not discovered")
|
|
}
|
|
if len(modelsFound) > 16 {
|
|
modelsFound = modelsFound[:16]
|
|
}
|
|
trace.phase("Q2", fmt.Sprintf("event=provider_model_resolved model_count=%d", len(modelsFound)))
|
|
trace.phase("Q3", "event=bounded_invocation_started")
|
|
results := qualifyModels(controlCtx, registration, owner, request.Prompt, request.MaxOutputBytes, modelsFound)
|
|
trace.phase("Q4", "event=bounded_invocation_completed")
|
|
trace.phase("Q5", fmt.Sprintf("event=invocation_evidence_built result_count=%d", len(results)))
|
|
if len(observers) > 0 && observers[0] != nil {
|
|
trace.phase("Q6", "event=observation_persistence_started")
|
|
if err := persistQualificationResults(controlCtx, observers[0], registration, results); err != nil {
|
|
return modelQualificationResponse{}, fmt.Errorf("persist model qualification observation: %w", err)
|
|
}
|
|
trace.phase("Q7", "event=observation_persistence_completed")
|
|
}
|
|
// Q8-Q17 are intentionally outside this worker's authority. Capability
|
|
// owns independent evaluation and governed promotion; this trace makes
|
|
// that hand-off explicit instead of disguising it as activation.
|
|
trace.phase("Q8", "event=independent_evaluation_not_started owner=capability")
|
|
return modelQualificationResponse{Provider: registration.ProviderInstanceID, ObservedAtUnixMillis: time.Now().UnixMilli(), Results: results}, nil
|
|
})
|
|
if !leader {
|
|
trace.phase("Q0", "event=joined_in_progress_qualification")
|
|
}
|
|
if err != nil {
|
|
if controlCtx.Err() != nil {
|
|
http.Error(w, "qualification timeout", http.StatusGatewayTimeout)
|
|
return
|
|
}
|
|
if strings.Contains(err.Error(), "requested model not discovered") {
|
|
http.Error(w, "requested model not discovered", http.StatusNotFound)
|
|
return
|
|
}
|
|
if strings.Contains(err.Error(), "provider unavailable") {
|
|
http.Error(w, "provider unavailable for qualification", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
http.Error(w, "persist model qualification observation", http.StatusBadGateway)
|
|
return
|
|
}
|
|
trace.phase("Q17", "event=qualification_response_returned activation=not_attempted")
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = jsonEncode(w, response)
|
|
})
|
|
}
|
|
|
|
func qualifyModels(ctx context.Context, registration providerRegistration, owner, prompt string, maxOutputBytes int, discovered []models.DiscoveredModel) []modelQualificationResult {
|
|
results := make([]modelQualificationResult, len(discovered))
|
|
var wg sync.WaitGroup
|
|
for index, model := range discovered {
|
|
wg.Add(1)
|
|
go func(index int, model models.DiscoveredModel) {
|
|
defer wg.Done()
|
|
started := time.Now()
|
|
request := cognitive.AdvisoryRequest{Version: cognitive.CreatorAdvisoryV1, RequestID: cognitive.RequestID(fmt.Sprintf("qualification:%s:%d", model.ProviderModelID, started.UnixNano())), ActorID: identity.ActorID("qualification-runtime"), PrincipalID: identity.PrincipalID(owner), ScopeID: authority.ScopeID("qualification-scope"), Message: prompt, Locale: "en-US", MaxOutputBytes: maxOutputBytes, ProviderBindingID: "qualification-binding-v1", ProviderBindingRevision: 1, ProviderInstanceID: registration.ProviderInstanceID, ModelDescriptorID: capability.ModelDescriptorID("qualification-" + string(model.ProviderModelID)), ProviderModelID: model.ProviderModelID}
|
|
response, err := registration.Provider.Advise(ctx, request)
|
|
result := modelQualificationResult{ProviderModelID: model.ProviderModelID, LatencyMillis: time.Since(started).Milliseconds(), QualityAssessed: false, Category: capability.ModelRatingUnrated}
|
|
if err != nil {
|
|
result.Status = "UNAVAILABLE"
|
|
result.FinishReason = cognitive.FinishProviderUnavailable
|
|
result.Error = err.Error()
|
|
results[index] = result
|
|
return
|
|
}
|
|
result.FinishReason = response.FinishReason
|
|
result.OutputObserved = response.AdvisoryText != ""
|
|
if response.FinishReason == cognitive.FinishCompleted && result.OutputObserved {
|
|
result.Status = "REACHABLE"
|
|
} else {
|
|
result.Status = "RESPONDED_WITHOUT_QUALIFICATION"
|
|
}
|
|
results[index] = result
|
|
}(index, model)
|
|
}
|
|
wg.Wait()
|
|
return results
|
|
}
|
|
|
|
func persistQualificationResults(ctx context.Context, observer qualificationObserver, registration providerRegistration, results []modelQualificationResult) error {
|
|
for _, result := range results {
|
|
observationID, err := ids.New()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
status := capability.ModelQualificationRespondedWithoutRating
|
|
switch result.Status {
|
|
case "REACHABLE":
|
|
status = capability.ModelQualificationReachable
|
|
case "UNAVAILABLE":
|
|
status = capability.ModelQualificationUnavailable
|
|
}
|
|
observation := capability.ModelQualificationObservation{
|
|
ID: capability.ModelQualificationObservationID(observationID), ProviderInstanceID: registration.ProviderInstanceID, ProviderModelID: result.ProviderModelID,
|
|
Status: status, FinishReason: string(result.FinishReason), LatencyMillis: result.LatencyMillis,
|
|
OutputObserved: result.OutputObserved, QualityAssessed: false, Category: capability.ModelRatingUnrated,
|
|
ObservedAtUnixMillis: time.Now().UnixMilli(),
|
|
}
|
|
if err := observation.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := observer.ObserveQualification(ctx, observation); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func jsonEncode(w http.ResponseWriter, value any) error {
|
|
return json.NewEncoder(w).Encode(value)
|
|
}
|