Assistant/cmd/cognitive-worker/gateway.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"git.cin.su/m/Assistant/adapters/models"
	"git.cin.su/m/Assistant/contracts/capability"
	"git.cin.su/m/Assistant/contracts/cognitive"
	"git.cin.su/m/Assistant/contracts/interaction"
	"git.cin.su/m/Assistant/platform/ids"
)

const creatorAdvisoryCapabilityID capability.CapabilityID = "creator-advisory"

type bindingResolver interface {
	Resolve(context.Context, capability.CognitiveRequirement) (capability.ResolvedCognitiveBinding, error)
	Observe(context.Context, capability.CognitiveUsageObservation) error
	RegisterDiscovery(context.Context, capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error)
}

type admissionClient interface {
	Admit(context.Context, string, capability.AdmissionRequest) (capability.AdmissionDecision, capability.AdmissionLease, error)
	Release(context.Context, capability.AdmissionLease) error
}

type providerRegistration struct {
	ProviderInstanceID capability.ProviderInstanceID
	Kind               capability.ProviderKind
	EndpointRef        capability.EndpointBindingRef
	CredentialRef      capability.CredentialRef
	AuthMode           capability.AuthMode
	BillingMode        capability.BillingMode
	DataPolicy         capability.DataPolicy
	Provider           advisoryProvider
	Discovery          models.LocalProvider
}

type cognitiveGateway struct {
	resolver             bindingResolver
	providers            map[capability.ProviderKind][]providerRegistration
	builtinLocal         *providerRegistration
	externalRecovery     []providerRegistration
	now                  func() time.Time
	allowCanary          bool
	canaryMaxOutputBytes int
	admission            admissionClient
}

func (g cognitiveGateway) policyHandler(w http.ResponseWriter, r *http.Request) {
	var request cognitive.GatewayAdvisoryPolicyRequest
	if !decodeJSON(w, r, 8<<10, &request) || request.PrincipalID == "" {
		return
	}
	requirement := capability.CognitiveRequirement{CapabilityID: creatorAdvisoryCapabilityID, OwnerPrincipalID: request.PrincipalID, DataPolicy: request.DataPolicy, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, AllowCanary: g.allowCanary, MaxOutputBytes: 1}
	selected, err := g.resolver.Resolve(r.Context(), requirement)
	policy := cognitive.GatewayAdvisoryPolicy{}
	if err == nil && selected.Binding.DataPolicy != capability.DataPolicyLocalOnly {
		policy.ExternalEgress = true
		policy.AcknowledgementRequired = true
		policy.Disclosure = "Текст вопроса будет передан внешнему когнитивному провайдеру; секреты и внутренний контекст не передаются."
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(policy)
}

func (g cognitiveGateway) advisoryHandler(w http.ResponseWriter, r *http.Request) {
	var request cognitive.GatewayAdvisoryRequest
	if !decodeJSON(w, r, 80<<10, &request) {
		return
	}
	if request.Version != cognitive.CreatorAdvisoryV1 || request.RequestID == "" || request.ActorID == "" || request.PrincipalID == "" || request.ScopeID == "" || request.Message == "" || request.MaxOutputBytes < 1 {
		http.Error(w, "invalid advisory request", http.StatusBadRequest)
		return
	}
	maxOutputBytes := request.MaxOutputBytes
	if g.allowCanary && g.canaryMaxOutputBytes > 0 && maxOutputBytes > g.canaryMaxOutputBytes {
		maxOutputBytes = g.canaryMaxOutputBytes
	}
	requirement := capability.CognitiveRequirement{CapabilityID: creatorAdvisoryCapabilityID, OwnerPrincipalID: request.PrincipalID, DataPolicy: request.DataPolicy, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, AllowCanary: g.allowCanary, InputBytes: int64(len(request.Message)), MaxOutputBytes: int64(maxOutputBytes), Objective: selectionObjectiveForMeaning(request.Meaning, request.Message)}
	selected, err := g.resolver.Resolve(r.Context(), requirement)
	if err != nil {
		writeGatewayResponse(w, g.invokeBuiltinAdvisory(r.Context(), request, maxOutputBytes))
		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 {
		writeGatewayResponse(w, g.invokeBuiltinAdvisory(r.Context(), request, maxOutputBytes))
		return
	}
	invocation := cognitive.AdvisoryRequest{Version: request.Version, RequestID: request.RequestID, ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: request.Message, Context: request.Context, Locale: request.Locale, MaxOutputBytes: maxOutputBytes, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ProviderModelID: selected.Model.ProviderModelID}
	started := g.now()
	response, invokeErr := g.invokeAdmitted(r.Context(), string(request.RequestID), provider, invocation, maxOutputBytes)
	if invokeErr != nil {
		response = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, FinishReason: cognitive.FinishProviderUnavailable}
	}
	g.observe(r.Context(), selected, response, started)
	if fallbackEligible(response.FinishReason) {
		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.DataPolicy == capability.DataPolicyLocalOnly || request.EgressAcknowledged) {
			fallbackRequest := invocation
			fallbackRequest.ProviderBindingID, fallbackRequest.ProviderBindingRevision = fallback.Binding.ID, fallback.Binding.Revision
			fallbackRequest.ProviderInstanceID, fallbackRequest.ModelDescriptorID, fallbackRequest.ProviderModelID = fallback.Provider.ID, fallback.Model.ID, fallback.Model.ProviderModelID
			fallbackStarted := g.now()
			fallbackResponse, fallbackErr := g.invokeAdmitted(r.Context(), string(request.RequestID)+":fallback", fallbackProvider, fallbackRequest, maxOutputBytes)
			if fallbackErr != nil {
				fallbackResponse = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: fallback.Binding.ID, ProviderBindingRevision: fallback.Binding.Revision, FinishReason: cognitive.FinishProviderUnavailable}
			}
			g.observe(r.Context(), fallback, fallbackResponse, fallbackStarted)
			response = fallbackResponse
		}
	}
	if response.FinishReason != cognitive.FinishCompleted && fallbackEligible(response.FinishReason) {
		writeGatewayResponse(w, g.invokeBuiltinAdvisory(r.Context(), request, maxOutputBytes))
		return
	}
	writeGatewayResponse(w, cognitive.GatewayAdvisoryResponse{Version: response.Version, RequestID: response.RequestID, AdvisoryText: response.AdvisoryText, ReferencedExcerptIDs: response.ReferencedExcerptIDs, FinishReason: response.FinishReason, Usage: response.Usage})
}

func (g cognitiveGateway) invokeAdmitted(ctx context.Context, leaseID string, provider advisoryProvider, invocation cognitive.AdvisoryRequest, maxOutputBytes int) (cognitive.AdvisoryResponse, error) {
	var lease capability.AdmissionLease
	if g.admission != nil {
		decision, admittedLease, err := g.admission.Admit(ctx, leaseID, capability.AdmissionRequest{InputBytes: len(invocation.Message), MaxOutputBytes: maxOutputBytes, DeadlineMillis: 0})
		if err != nil {
			return cognitive.AdvisoryResponse{}, err
		}
		if decision.State != capability.AdmissionAdmitted {
			return cognitive.AdvisoryResponse{Version: invocation.Version, RequestID: invocation.RequestID, ProviderBindingID: invocation.ProviderBindingID, ProviderBindingRevision: invocation.ProviderBindingRevision, FinishReason: cognitive.FinishRateLimited}, nil
		}
		lease = admittedLease
		defer func() { _ = g.admission.Release(ctx, lease) }()
	}
	return provider.Advise(ctx, invocation)
}

func selectionObjectiveForMessage(message string) capability.SelectionObjective {
	lower := strings.ToLower(message)
	if strings.Contains(lower, "дешев") || strings.Contains(lower, "бесплат") || strings.Contains(lower, "эконом") || strings.Contains(lower, "cost") || strings.Contains(lower, "cheap") {
		return capability.SelectionEconomy
	}
	if strings.Contains(lower, "быстр") || strings.Contains(lower, "срочно") || strings.Contains(lower, "статус") || strings.Contains(lower, "quick") || strings.Contains(lower, "fast") {
		return capability.SelectionSpeed
	}
	if strings.Contains(lower, "сложн") || strings.Contains(lower, "анализ") || strings.Contains(lower, "код") || strings.Contains(lower, "план") || strings.Contains(lower, "глубок") || strings.Contains(lower, "reason") || strings.Contains(lower, "code") {
		return capability.SelectionQuality
	}
	return capability.SelectionBalanced
}

func selectionObjectiveForMeaning(meaning cognitive.TaskMeaning, message string) capability.SelectionObjective {
	if meaning.CapabilityKind != "" || len(meaning.CapabilityHints) > 0 || meaning.SpeechAct != "" || meaning.RequestedOutcome != "" {
		kinds := append([]capability.CapabilityKind{meaning.CapabilityKind}, meaning.CapabilityHints...)
		for _, kind := range kinds {
			switch kind {
			case capability.CapabilityKindModifyCode, capability.CapabilityKindRunTests, capability.CapabilityKindVerifyFact, capability.CapabilityKindInspectProject, capability.CapabilityKindReadDatabase, capability.CapabilityKindQueryLogs:
				return capability.SelectionQuality
			case capability.CapabilityKindScheduleReminder:
				return capability.SelectionSpeed
			}
		}
		if meaning.RequestedOutcome != "" || meaning.SpeechAct == "REQUEST_TO_INVESTIGATE" || meaning.SpeechAct == "DIRECTIVE" {
			return capability.SelectionQuality
		}
		if meaning.ConfidenceMillis > 0 {
			return capability.SelectionBalanced
		}
	}
	return selectionObjectiveForMessage(message)
}

func builtinAdvisory(request cognitive.GatewayAdvisoryRequest) cognitive.GatewayAdvisoryResponse {
	text := "Я рядом и продолжаю работу во встроенном режиме Домового. Внешний когнитивный маршрут сейчас недоступен, поэтому опираюсь на уже установленный смысл сообщения и локальные возможности."
	switch request.Meaning.InformationRequestKind {
	case interaction.InformationRequestCognitiveModelInventory:
		text = "Запрос понят как проверка моделей. Встроенный режим Домового доступен; я могу показать подключённые модели по назначению, качеству, скорости и стоимости. Ключи и пароли в разговор не нужны."
	case interaction.InformationRequestCapabilityInventory:
		text = "Запрос понят как проверка возможностей. Я могу объяснить доступные capabilities и отличить то, что уже подключено, от того, что пока является только планом или capability gap."
	case interaction.InformationRequestCognitiveChannelStatus:
		text = "Запрос понят как проверка когнитивного канала. Локальный режим остаётся доступен; я объясню наблюдаемое состояние и что можно сделать без ожидания внешней модели."
	case interaction.InformationRequestCognitiveRecovery:
		text = "Запрос понят как восстановление ИИ. Начинаю с локального маршрута: фиксирую, что недоступно, проверяю безопасные варианты восстановления и сформулирую следующий обратимый шаг."
	default:
		switch request.Meaning.SpeechAct {
		case "REQUEST_TO_INVESTIGATE":
			text = "Я принял это как исследование. Даже без внешней модели могу сформулировать гипотезы, отделить известное от неизвестного и предложить следующий проверяемый шаг."
		case "QUESTION":
			text = "Я отвечаю во встроенном режиме Домового. Если для полного ответа не хватает внешней модели, объясню это прямо и продолжу с тем, что можно проверить локально."
		}
	}
	return cognitive.GatewayAdvisoryResponse{Version: request.Version, RequestID: request.RequestID, AdvisoryText: text, FinishReason: cognitive.FinishCompleted}
}

func (g cognitiveGateway) invokeBuiltinAdvisory(ctx context.Context, request cognitive.GatewayAdvisoryRequest, maxOutputBytes int) cognitive.GatewayAdvisoryResponse {
	if g.builtinLocal != nil && g.builtinLocal.Provider != nil {
		invocation := cognitive.AdvisoryRequest{Version: request.Version, RequestID: request.RequestID, ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: request.Message, Context: request.Context, Locale: request.Locale, MaxOutputBytes: maxOutputBytes, ProviderBindingID: "builtin-local-binding-v1", ProviderBindingRevision: 1, ProviderInstanceID: "builtin-local-lifeboat-v1", ModelDescriptorID: "builtin-local-model-v1", ProviderModelID: "domovoy-lifeboat"}
		started := g.now()
		response, err := g.builtinLocal.Provider.Advise(ctx, invocation)
		if response.RequestID == "" {
			response.RequestID = request.RequestID
		}
		if response.Version == "" {
			response.Version = request.Version
		}
		if err != nil {
			response = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: "builtin-local-binding-v1", ProviderBindingRevision: 1, FinishReason: cognitive.FinishInvalidProviderResponse}
		}
		g.observeBuiltin(ctx, g.builtinLocal, response, started)
		if err == nil && response.FinishReason == cognitive.FinishCompleted && strings.TrimSpace(response.AdvisoryText) != "" {
			return cognitive.GatewayAdvisoryResponse{Version: request.Version, RequestID: request.RequestID, AdvisoryText: response.AdvisoryText, ReferencedExcerptIDs: response.ReferencedExcerptIDs, FinishReason: response.FinishReason, Usage: response.Usage}
		}
	}
	if request.EgressAcknowledged {
		if response, ok := g.recoverExternalAdvisory(ctx, request, maxOutputBytes); ok {
			return response
		}
	}
	return builtinAdvisory(request)
}

func (g cognitiveGateway) recoverExternalAdvisory(ctx context.Context, request cognitive.GatewayAdvisoryRequest, maxOutputBytes int) (cognitive.GatewayAdvisoryResponse, bool) {
	for _, candidate := range g.externalRecovery {
		if candidate.Provider == nil || candidate.Discovery == nil {
			continue
		}
		probeCtx, cancel := context.WithTimeout(ctx, 8*time.Second)
		modelsFound, probe := candidate.Discovery.Discover(probeCtx)
		cancel()
		if probe.Health != capability.HealthAvailable || len(modelsFound) == 0 {
			continue
		}
		model := modelsFound[0]
		invocation := cognitive.AdvisoryRequest{Version: request.Version, RequestID: request.RequestID, ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: request.Message, Context: request.Context, Locale: request.Locale, MaxOutputBytes: maxOutputBytes, ProviderBindingID: "recovery-external-binding-v1", ProviderBindingRevision: 1, ProviderInstanceID: capability.ProviderInstanceID("recovery-" + string(candidate.ProviderInstanceID)), ModelDescriptorID: capability.ModelDescriptorID("recovery-model-" + string(model.ProviderModelID)), ProviderModelID: model.ProviderModelID}
		response, err := candidate.Provider.Advise(ctx, invocation)
		if err == nil && response.FinishReason == cognitive.FinishCompleted && strings.TrimSpace(response.AdvisoryText) != "" {
			return cognitive.GatewayAdvisoryResponse{Version: request.Version, RequestID: request.RequestID, AdvisoryText: response.AdvisoryText, ReferencedExcerptIDs: response.ReferencedExcerptIDs, FinishReason: response.FinishReason, Usage: response.Usage}, true
		}
	}
	return cognitive.GatewayAdvisoryResponse{}, false
}

func (g cognitiveGateway) observe(ctx context.Context, selected capability.ResolvedCognitiveBinding, response cognitive.AdvisoryResponse, started time.Time) {
	usage := response.Usage
	observationID, idErr := ids.New()
	if idErr == nil {
		observation := capability.CognitiveUsageObservation{ObservationID: observationID, RequestID: ids.ID(response.RequestID), BindingID: selected.Binding.ID, BindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ObservedAtUnixMillis: g.now().UnixMilli(), LatencyMillis: g.now().Sub(started).Milliseconds(), ResultClass: capability.CognitiveResultClass(response.FinishReason)}
		if usage != nil {
			observation.InputUnits, observation.OutputUnits = usage.InputUnits, usage.OutputUnits
		}
		if response.FinishReason != cognitive.FinishCompleted {
			observation.FailureType = capability.CognitiveFailureType(response.FinishReason)
		}
		if selected.Provider.Kind == "openai-compatible-chat-v1" || selected.Provider.Kind == "ollama-v1" {
			zero := int64(0)
			observation.ActualCostMicros = &zero
		}
		_ = g.resolver.Observe(ctx, observation)
	}
}

// observeBuiltin gives the always-available local provider the same evidence
// path as a resolved binding. The identity is stable and contains references
// only; no endpoint, credential, prompt, or response text is persisted here.
func (g cognitiveGateway) observeBuiltin(ctx context.Context, registration *providerRegistration, response cognitive.AdvisoryResponse, started time.Time) {
	if registration == nil {
		return
	}
	selected := capability.ResolvedCognitiveBinding{
		Binding:  capability.CognitiveBinding{ID: "builtin-local-binding-v1", Revision: 1},
		Provider: capability.ProviderInstance{ID: "builtin-local-lifeboat-v1", Revision: 1, Kind: registration.Kind},
		Model:    capability.ModelDescriptor{ID: "builtin-local-model-v1", Revision: 1, ProviderInstanceID: "builtin-local-lifeboat-v1", ProviderModelID: "domovoy-lifeboat"},
	}
	g.observe(ctx, selected, response, started)
}

func fallbackEligible(reason cognitive.FinishReason) bool {
	switch reason {
	case cognitive.FinishProviderUnavailable, cognitive.FinishModelUnavailable, cognitive.FinishTimeout, cognitive.FinishDegraded:
		return true
	}
	return false
}

func (g cognitiveGateway) provider(selected capability.ResolvedCognitiveBinding) advisoryProvider {
	for _, registration := range g.providers[selected.Provider.Kind] {
		if registration.EndpointRef == selected.Provider.EndpointRef && registration.CredentialRef == selected.Provider.CredentialRef {
			return registration.Provider
		}
	}
	return nil
}

func writeGatewayResponse(w http.ResponseWriter, response cognitive.GatewayAdvisoryResponse) {
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(response)
}

func decodeJSON(w http.ResponseWriter, r *http.Request, limit int64, destination any) bool {
	r.Body = http.MaxBytesReader(w, r.Body, limit)
	decoder := json.NewDecoder(r.Body)
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		http.Error(w, "invalid request", http.StatusBadRequest)
		return false
	}
	if err := decoder.Decode(&struct{}{}); err != io.EOF {
		http.Error(w, "invalid request", http.StatusBadRequest)
		return false
	}
	return true
}

type httpBindingResolver struct {
	base   *url.URL
	client *http.Client
}

func newHTTPBindingResolver(raw string, client *http.Client) (*httpBindingResolver, error) {
	base, err := url.Parse(raw)
	if err != nil || base.Scheme == "" || base.Host == "" {
		return nil, fmt.Errorf("invalid Core Runtime URL")
	}
	return &httpBindingResolver{base: base, client: client}, nil
}

func (r *httpBindingResolver) Resolve(ctx context.Context, value capability.CognitiveRequirement) (capability.ResolvedCognitiveBinding, error) {
	var result capability.ResolvedCognitiveBinding
	err := r.post(ctx, "/v1/cognitive-bindings/resolve", value, &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)
}

func (r *httpBindingResolver) RegisterDiscovery(ctx context.Context, value capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error) {
	var result capability.CognitiveDiscoveryRegistration
	err := r.post(ctx, "/v1/cognitive-registry/discoveries", value, &result)
	return result, err
}

func (r *httpBindingResolver) post(ctx context.Context, path string, value, destination any) error {
	payload, err := json.Marshal(value)
	if err != nil {
		return err
	}
	target := *r.base
	target.Path = path
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, target.String(), bytes.NewReader(payload))
	if err != nil {
		return err
	}
	request.Header.Set("Content-Type", "application/json")
	response, err := r.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode < 200 || response.StatusCode >= 300 {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))
		return fmt.Errorf("registry status %d", response.StatusCode)
	}
	if destination != nil {
		return json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(destination)
	}
	return nil
}