Assistant/cmd/creator-web/server.go

package main

import (
	"embed"
	"encoding/json"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"net/url"
	"strings"
	"unicode/utf8"

	"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/contracts/interaction"
	"git.cin.su/m/Assistant/contracts/situation"
	"git.cin.su/m/Assistant/platform/ids"
)

const (
	maxRequestBytes         = 8 << 10
	maxAdvisoryRequestBytes = 20 << 10
	maxResponseBytes        = 1 << 20
)

//go:embed static
var staticFiles embed.FS

type gateway struct {
	core       *url.URL
	continuity *url.URL
	commit     *url.URL
	cognitive  *url.URL
	client     *http.Client
	static     http.Handler
}

type routeTarget uint8

const (
	targetCore routeTarget = iota
	targetContinuity
	targetCommit
)

func newHandler(coreURL, continuityURL, commitURL string, client *http.Client, cognitiveURLs ...string) (http.Handler, error) {
	core, err := parseRuntimeURL(coreURL, "Core")
	if err != nil {
		return nil, err
	}
	continuity, err := parseRuntimeURL(continuityURL, "Continuity")
	if err != nil {
		return nil, err
	}
	commit, err := parseRuntimeURL(commitURL, "Commit")
	if err != nil {
		return nil, err
	}
	cognitiveURL := "http://cognitive-worker:8083"
	if len(cognitiveURLs) > 0 {
		cognitiveURL = cognitiveURLs[0]
	}
	cognitiveRuntime, err := parseRuntimeURL(cognitiveURL, "Cognitive")
	if err != nil {
		return nil, err
	}
	assets, err := fs.Sub(staticFiles, "static")
	if err != nil {
		return nil, fmt.Errorf("load embedded console: %w", err)
	}
	if client == nil {
		client = http.DefaultClient
	}
	return securityHeaders(&gateway{core: core, continuity: continuity, commit: commit, cognitive: cognitiveRuntime, client: client, static: http.FileServer(http.FS(assets))}), nil
}

func parseRuntimeURL(rawURL, name string) (*url.URL, error) {
	parsed, err := url.Parse(rawURL)
	if err != nil || parsed.Scheme == "" || parsed.Host == "" {
		return nil, fmt.Errorf("invalid %s Runtime URL", name)
	}
	return parsed, nil
}

func (g *gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path == "/healthz" {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]string{"status": "ok", "runtime": "creator-web"})
		return
	}
	if strings.HasPrefix(r.URL.Path, "/api/") {
		if r.Method == http.MethodGet && r.URL.Path == "/api/advisory-policy" {
			g.handleAdvisoryPolicy(w, r)
			return
		}
		if r.Method == http.MethodPost && r.URL.Path == "/api/advisory" {
			g.handleAdvisory(w, r)
			return
		}
		if r.Method == http.MethodPost && r.URL.Path == "/api/chat/messages" {
			g.handleChatMessage(w, r)
			return
		}
		g.forwardAPI(w, r)
		return
	}
	if r.Method != http.MethodGet && r.Method != http.MethodHead {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if r.URL.Path == "/" {
		w.Header().Set("Cache-Control", "no-store")
	}
	if r.URL.Path == "/chat" {
		r = r.Clone(r.Context())
		r.URL.Path = "/chat.html"
		w.Header().Set("Cache-Control", "no-store")
	}
	g.static.ServeHTTP(w, r)
}

type browserAdvisoryRequest struct {
	Message            string `json:"message"`
	SessionID          string `json:"session_id"`
	WorkspaceID        string `json:"workspace_id"`
	EgressAcknowledged bool   `json:"egress_acknowledged"`
}

type advisoryPolicy struct {
	ExternalEgress          bool
	AcknowledgementRequired bool
	Disclosure              string
}

type browserChatRequest struct {
	ConversationID      string `json:"conversation_id"`
	Message             string `json:"message"`
	CorrelationID       string `json:"correlation_id"`
	CorrectsUtteranceID string `json:"corrects_utterance_id"`
}

type browserChatResponse struct {
	ConversationID    string `json:"conversation_id"`
	UtteranceID       string `json:"utterance_id"`
	SpeechAct         string `json:"speech_act"`
	ReplyText         string `json:"reply_text"`
	Cognition         string `json:"cognition"`
	ActiveFocus       string `json:"active_focus"`
	ProjectProvenance string `json:"project_provenance"`
	Clarification     string `json:"clarification"`
	WorkPlanned       bool   `json:"work_planned"`
}

func (g *gateway) handleAdvisoryPolicy(w http.ResponseWriter, r *http.Request) {
	authorization := r.Header.Get("Authorization")
	if authorization == "" {
		http.Error(w, "creator session denied", http.StatusUnauthorized)
		return
	}
	var session identity.CreatorSession
	status, err := g.getTrustedJSON(r, g.core, "/v1/creator/sessions/current", "", authorization, &session)
	if err != nil {
		http.Error(w, "creator session unavailable", http.StatusBadGateway)
		return
	}
	if status != http.StatusOK {
		http.Error(w, "creator session denied", status)
		return
	}
	if session.ActorID == "" || session.PrincipalID == "" {
		http.Error(w, "creator session denied", http.StatusUnauthorized)
		return
	}
	policyRequest := cognitive.GatewayAdvisoryPolicyRequest{PrincipalID: session.PrincipalID, DataPolicy: capability.DataPolicyExternalMessageOnly}
	payload, err := json.Marshal(policyRequest)
	if err != nil {
		http.Error(w, "advisory policy unavailable", http.StatusServiceUnavailable)
		return
	}
	target := *g.cognitive
	target.Path = "/v1/advisory-policy"
	upstreamRequest, err := http.NewRequestWithContext(r.Context(), http.MethodPost, target.String(), strings.NewReader(string(payload)))
	if err != nil {
		http.Error(w, "advisory policy unavailable", http.StatusBadGateway)
		return
	}
	upstreamRequest.Header.Set("Content-Type", "application/json")
	response, err := g.client.Do(upstreamRequest)
	if err != nil {
		http.Error(w, "advisory policy unavailable", http.StatusBadGateway)
		return
	}
	defer response.Body.Close()
	var policy advisoryPolicy
	if response.StatusCode != http.StatusOK || decodeStrictJSON(response.Body, maxResponseBytes, &policy) != nil {
		http.Error(w, "advisory policy unavailable", http.StatusBadGateway)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(policy)
}

func (g *gateway) handleAdvisory(w http.ResponseWriter, r *http.Request) {
	var browserRequest browserAdvisoryRequest
	if err := decodeStrictJSON(r.Body, maxAdvisoryRequestBytes, &browserRequest); err != nil || browserRequest.Message == "" || len(browserRequest.Message) > 16*1024 || !validID(browserRequest.SessionID) || !validID(browserRequest.WorkspaceID) {
		http.Error(w, "invalid advisory request", http.StatusBadRequest)
		return
	}
	authorization := r.Header.Get("Authorization")
	if authorization == "" {
		http.Error(w, "creator session denied", http.StatusUnauthorized)
		return
	}
	var session identity.CreatorSession
	status, err := g.getTrustedJSON(r, g.core, "/v1/creator/sessions/current", "", authorization, &session)
	if err != nil {
		http.Error(w, "creator session unavailable", http.StatusBadGateway)
		return
	}
	if status != http.StatusOK {
		http.Error(w, "creator session denied", status)
		return
	}
	query := url.Values{"session_id": {browserRequest.SessionID}, "workspace_id": {browserRequest.WorkspaceID}}.Encode()
	var focus situation.FocusBinding
	status, err = g.getTrustedJSON(r, g.core, "/v1/creator/focus", query, authorization, &focus)
	if err != nil {
		http.Error(w, "creator focus unavailable", http.StatusBadGateway)
		return
	}
	if status != http.StatusOK {
		http.Error(w, "creator focus unavailable", status)
		return
	}
	if session.ActorID == "" || session.PrincipalID == "" || focus.ActorID != session.ActorID || focus.ScopeID == "" {
		http.Error(w, "creator focus denied", http.StatusForbidden)
		return
	}
	requestID, err := ids.New()
	if err != nil {
		http.Error(w, "advisory unavailable", http.StatusServiceUnavailable)
		return
	}
	trusted := cognitive.GatewayAdvisoryRequest{Version: cognitive.CreatorAdvisoryV1, RequestID: cognitive.RequestID(requestID), ActorID: session.ActorID, PrincipalID: session.PrincipalID, ScopeID: focus.ScopeID, Message: browserRequest.Message, Context: []cognitive.ContextExcerpt{}, Locale: "ru-RU", MaxOutputBytes: 16 * 1024, DataPolicy: capability.DataPolicyExternalMessageOnly, EgressAcknowledged: browserRequest.EgressAcknowledged}
	payload, err := json.Marshal(trusted)
	if err != nil {
		http.Error(w, "advisory unavailable", http.StatusServiceUnavailable)
		return
	}
	target := *g.cognitive
	target.Path = "/v1/advisories"
	upstreamRequest, err := http.NewRequestWithContext(r.Context(), http.MethodPost, target.String(), strings.NewReader(string(payload)))
	if err != nil {
		http.Error(w, "advisory unavailable", http.StatusBadGateway)
		return
	}
	upstreamRequest.Header.Set("Content-Type", "application/json")
	response, err := g.client.Do(upstreamRequest)
	if err != nil {
		http.Error(w, "advisory unavailable", http.StatusBadGateway)
		return
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		http.Error(w, "advisory unavailable", http.StatusBadGateway)
		return
	}
	var advisory cognitive.GatewayAdvisoryResponse
	if err := decodeStrictJSON(response.Body, maxResponseBytes, &advisory); err != nil || !validAdvisoryResponse(advisory, trusted) {
		http.Error(w, "invalid advisory response", http.StatusBadGateway)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(advisory)
}

func (g *gateway) handleChatMessage(w http.ResponseWriter, r *http.Request) {
	var browserRequest browserChatRequest
	if err := decodeStrictJSON(r.Body, maxAdvisoryRequestBytes, &browserRequest); err != nil || strings.TrimSpace(browserRequest.Message) == "" || len(browserRequest.Message) > 16*1024 {
		http.Error(w, "invalid chat message", http.StatusBadRequest)
		return
	}
	authorization := r.Header.Get("Authorization")
	if authorization == "" {
		http.Error(w, "creator session denied", http.StatusUnauthorized)
		return
	}
	var session identity.CreatorSession
	status, err := g.getTrustedJSON(r, g.core, "/v1/creator/sessions/current", "", authorization, &session)
	if err != nil {
		http.Error(w, "creator session unavailable", http.StatusBadGateway)
		return
	}
	if status != http.StatusOK || session.ActorID == "" || session.PrincipalID == "" {
		http.Error(w, "creator session denied", status)
		return
	}
	coreRequest := interaction.ChatMessageRequest{ConversationID: interaction.ConversationID(browserRequest.ConversationID), Message: browserRequest.Message, CorrelationID: browserRequest.CorrelationID, CorrectsUtteranceID: interaction.HumanUtteranceID(browserRequest.CorrectsUtteranceID)}
	payload, err := json.Marshal(coreRequest)
	if err != nil {
		http.Error(w, "chat unavailable", http.StatusServiceUnavailable)
		return
	}
	var saved interaction.ChatMessageResponse
	status, err = g.postTrustedJSON(r, g.core, "/v1/creator/chat/messages", authorization, payload, &saved)
	if err != nil {
		http.Error(w, "chat unavailable", http.StatusBadGateway)
		return
	}
	if status != http.StatusCreated {
		http.Error(w, "chat unavailable", status)
		return
	}
	reply := saved.FallbackText
	cognition := "not_requested"
	if saved.PlanningReply != "" {
		reply = saved.PlanningReply
	}
	if saved.NeedsCognition && !saved.WorkPlanned && saved.Clarification != interaction.ClarificationRequired {
		cognition = "unavailable"
		requestID, idErr := ids.New()
		if idErr == nil {
			trusted := cognitive.GatewayAdvisoryRequest{Version: cognitive.CreatorAdvisoryV1, RequestID: cognitive.RequestID(requestID), ActorID: session.ActorID, PrincipalID: session.PrincipalID, ScopeID: "domovoy-global-scope", Message: browserRequest.Message, Context: []cognitive.ContextExcerpt{}, Locale: "ru-RU", MaxOutputBytes: 4096, DataPolicy: capability.DataPolicyLocalOnly, EgressAcknowledged: false}
			payload, err = json.Marshal(trusted)
			if err == nil {
				var advisory cognitive.GatewayAdvisoryResponse
				status, err = g.postTrustedJSON(r, g.cognitive, "/v1/advisories", "", payload, &advisory)
				if err == nil && status == http.StatusOK && validAdvisoryResponse(advisory, trusted) {
					cognition = string(advisory.FinishReason)
					if advisory.FinishReason == cognitive.FinishCompleted {
						reply = advisory.AdvisoryText
					}
				}
			}
		}
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(browserChatResponse{ConversationID: string(saved.ConversationID), UtteranceID: string(saved.UtteranceID), SpeechAct: string(saved.SpeechAct), ReplyText: reply, Cognition: cognition, ActiveFocus: saved.ActiveProjectName, ProjectProvenance: string(saved.ProjectProvenance), Clarification: string(saved.Clarification), WorkPlanned: saved.WorkPlanned})
}

func validAdvisoryResponse(response cognitive.GatewayAdvisoryResponse, request cognitive.GatewayAdvisoryRequest) bool {
	if response.Version != cognitive.CreatorAdvisoryV1 || response.RequestID != request.RequestID || len(response.ReferencedExcerptIDs) != 0 {
		return false
	}
	if response.Usage != nil && (response.Usage.InputUnits < 0 || response.Usage.OutputUnits < 0) {
		return false
	}
	switch response.FinishReason {
	case cognitive.FinishCompleted:
		return response.AdvisoryText != "" && len(response.AdvisoryText) <= request.MaxOutputBytes && utf8.ValidString(response.AdvisoryText)
	case cognitive.FinishLengthLimit, cognitive.FinishContentRejected, cognitive.FinishProviderUnavailable, cognitive.FinishModelUnavailable, cognitive.FinishRateLimited, cognitive.FinishAuthenticationFailed, cognitive.FinishBillingOrBudgetBlocked, cognitive.FinishContextTooLarge, cognitive.FinishUnsupportedCapability, cognitive.FinishTimeout, cognitive.FinishInvalidProviderResponse, cognitive.FinishInvalidStructuredOutput, cognitive.FinishDegraded:
		return response.AdvisoryText == ""
	default:
		return false
	}
}

func (g *gateway) getTrustedJSON(r *http.Request, upstream *url.URL, path, rawQuery, authorization string, destination any) (int, error) {
	target := *upstream
	target.Path = path
	target.RawQuery = rawQuery
	request, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target.String(), nil)
	if err != nil {
		return 0, err
	}
	request.Header.Set("Authorization", authorization)
	response, err := g.client.Do(request)
	if err != nil {
		return 0, err
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, maxResponseBytes))
		return response.StatusCode, nil
	}
	if err := decodeStrictJSON(response.Body, maxResponseBytes, destination); err != nil {
		return response.StatusCode, err
	}
	return response.StatusCode, nil
}

func (g *gateway) postTrustedJSON(r *http.Request, upstream *url.URL, path, authorization string, payload []byte, destination any) (int, error) {
	target := *upstream
	target.Path = path
	request, err := http.NewRequestWithContext(r.Context(), http.MethodPost, target.String(), strings.NewReader(string(payload)))
	if err != nil {
		return 0, err
	}
	request.Header.Set("Content-Type", "application/json")
	if authorization != "" {
		request.Header.Set("Authorization", authorization)
	}
	response, err := g.client.Do(request)
	if err != nil {
		return 0, err
	}
	defer response.Body.Close()
	if response.StatusCode < 200 || response.StatusCode > 299 {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, maxResponseBytes))
		return response.StatusCode, nil
	}
	if err := decodeStrictJSON(response.Body, maxResponseBytes, destination); err != nil {
		return response.StatusCode, err
	}
	return response.StatusCode, nil
}

func decodeStrictJSON(reader io.Reader, limit int64, destination any) error {
	decoder := json.NewDecoder(io.LimitReader(reader, limit+1))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		return err
	}
	if err := decoder.Decode(&struct{}{}); err != io.EOF {
		return fmt.Errorf("trailing JSON")
	}
	return nil
}

func (g *gateway) forwardAPI(w http.ResponseWriter, r *http.Request) {
	routeTarget, upstreamPath, allowed := creatorRoute(r.Method, r.URL.Path)
	if !allowed {
		http.Error(w, "creator route denied", http.StatusNotFound)
		return
	}
	body, err := io.ReadAll(io.LimitReader(r.Body, maxRequestBytes+1))
	if err != nil {
		http.Error(w, "invalid request", http.StatusBadRequest)
		return
	}
	if len(body) > maxRequestBytes {
		http.Error(w, "request too large", http.StatusRequestEntityTooLarge)
		return
	}
	upstream := g.core
	if routeTarget == targetContinuity {
		upstream = g.continuity
	} else if routeTarget == targetCommit {
		upstream = g.commit
	}
	target := *upstream
	target.Path = upstreamPath
	target.RawQuery = r.URL.RawQuery
	request, err := http.NewRequestWithContext(r.Context(), r.Method, target.String(), strings.NewReader(string(body)))
	if err != nil {
		http.Error(w, "creator gateway unavailable", http.StatusBadGateway)
		return
	}
	copyRequestHeader(request.Header, r.Header, "Authorization")
	copyRequestHeader(request.Header, r.Header, "Accept")
	copyRequestHeader(request.Header, r.Header, "Content-Type")
	response, err := g.client.Do(request)
	if err != nil {
		http.Error(w, "creator gateway unavailable", http.StatusBadGateway)
		return
	}
	defer response.Body.Close()
	copyResponseHeader(w.Header(), response.Header, "Content-Type")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(response.StatusCode)
	_, _ = io.Copy(w, io.LimitReader(response.Body, maxResponseBytes))
}

func creatorRoute(method, path string) (routeTarget, string, bool) {
	switch {
	case method == http.MethodPost && path == "/api/session":
		return targetCore, "/v1/creator/sessions", true
	case method == http.MethodGet && path == "/api/session/current":
		return targetCore, "/v1/creator/sessions/current", true
	case method == http.MethodGet && path == "/api/focus":
		return targetCore, "/v1/creator/focus", true
	case method == http.MethodGet && strings.HasPrefix(path, "/api/tasks/"):
		id := strings.TrimPrefix(path, "/api/tasks/")
		if validID(id) {
			return targetCore, "/v1/creator/tasks/" + id, true
		}
	case method == http.MethodGet && strings.HasPrefix(path, "/api/capabilities/") && strings.HasSuffix(path, "/health"):
		id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/capabilities/"), "/health")
		if validID(id) {
			return targetCore, "/v1/creator/capabilities/" + id + "/health", true
		}
	case method == http.MethodGet && strings.HasPrefix(path, "/api/prospective-items/"):
		id := strings.TrimPrefix(path, "/api/prospective-items/")
		if validID(id) {
			return targetContinuity, "/v1/creator/prospective-items/" + id, true
		}
	case method == http.MethodGet && path == "/api/prospective-items":
		return targetContinuity, "/v1/creator/prospective-items", true
	case method == http.MethodGet && path == "/api/actions":
		return targetCommit, "/v1/creator/authorized-actions", true
	case method == http.MethodPost && path == "/api/actions/satisfy-delivered":
		return targetCommit, "/v1/creator/continuity/satisfy-delivered", true
	case method == http.MethodPost && path == "/api/actions/notify-delivered":
		return targetCommit, "/v1/creator/continuity/notify-delivered", true
	case method == http.MethodPost && strings.HasPrefix(path, "/api/actions/") && strings.HasSuffix(path, "/reconcile"):
		id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/actions/"), "/reconcile")
		if validID(id) {
			return targetCommit, "/v1/creator/authorized-actions/" + id + "/reconcile", true
		}
	}
	return targetCore, "", false
}

func validID(value string) bool {
	if value == "" || len(value) > 128 {
		return false
	}
	for _, char := range value {
		if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || strings.ContainsRune("._:-", char) {
			continue
		}
		return false
	}
	return true
}

func copyRequestHeader(destination, source http.Header, name string) {
	if value := source.Get(name); value != "" {
		destination.Set(name, value)
	}
}

func copyResponseHeader(destination, source http.Header, name string) {
	if value := source.Get(name); value != "" {
		destination.Set(name, value)
	}
}

func securityHeaders(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
		w.Header().Set("Referrer-Policy", "no-referrer")
		w.Header().Set("X-Content-Type-Options", "nosniff")
		w.Header().Set("X-Frame-Options", "DENY")
		next.ServeHTTP(w, r)
	})
}