package callmemory

import (
	"context"
	"data-terminal-cloud/internal/billing"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/identity"
	"encoding/json"
	"errors"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"io"
	"log/slog"
	"mime"
	"net/http"
	"strings"
	"time"
)

type Service struct {
	pool          *pgxpool.Pool
	now           func() time.Time
	logger        *slog.Logger
	projection    ProjectionProvider
	extractor     IntentExtractor
	direct        DirectOptions
	managedSync   func(context.Context) error
	recordings    RecordingStore
	mixedRecorder MixedRecorder
	vikingConsole VikingConsoleProvider
}

func NewService(pool *pgxpool.Pool, logger *slog.Logger) *Service {
	if logger == nil {
		logger = slog.Default()
	}
	return &Service{pool: pool, now: time.Now, logger: logger}
}

type DeviceAuth func(*http.Request) (deviceauth.Identity, []byte, []byte, error)

func (s *Service) RegisterDeviceRoutes(mux *http.ServeMux, auth DeviceAuth) {
	for _, op := range []string{"warmup", "prepare", "events", "heartbeat", "end", "status", "command-acks"} {
		mux.HandleFunc("POST /api/v1/device/calls/"+op, func(w http.ResponseWriter, r *http.Request) {
			id, key, raw, e := auth(r)
			if e != nil {
				var typed interface {
					error
					HTTPStatus() int
				}
				if errors.As(e, &typed) && typed.HTTPStatus() >= 400 && typed.HTTPStatus() < 600 {
					s.write(w, r, nil, fault(typed.HTTPStatus(), typed.Error()))
				} else {
					s.write(w, r, nil, fault(503, "AUTHENTICATION_UNAVAILABLE"))
				}
				return
			}
			if len(raw) > MaxBody {
				s.write(w, r, nil, fault(413, "REQUEST_TOO_LARGE"))
				return
			}
			result, e := s.DeviceRequest(r.Context(), id, key, r.Header.Get("X-Request-ID"), op, raw)
			s.write(w, r, result, e)
		})
	}
}
func (s *Service) RegisterMemoryRoutes(mux *http.ServeMux, wrap func(http.Handler) http.Handler) {
	for _, op := range []string{"viking/query", "viking/delete", "query", "mutations", "mutations/status", "sync/status", "mutations/resolve", "mutations/confirm", "access/query", "access/grant", "calls/query", "calls/turns", "calls/recordings", "calls/recording-access"} {
		mux.Handle("POST /api/v1/memories/"+op, wrap(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			p, ok := identity.PrincipalFromContext(r.Context())
			if !ok {
				s.write(w, r, nil, fault(401, "AUTHENTICATION_REQUIRED"))
				return
			}
			mediaType, _, mediaErr := mime.ParseMediaType(r.Header.Get("Content-Type"))
			if mediaErr != nil || mediaType != "application/json" || r.URL.RawQuery != "" {
				s.write(w, r, nil, fault(400, "INVALID_REQUEST_FORMAT"))
				return
			}
			raw, e := io.ReadAll(io.LimitReader(r.Body, MaxMemoryBody+1))
			if e != nil || len(raw) > MaxMemoryBody {
				s.write(w, r, nil, fault(413, "REQUEST_TOO_LARGE"))
				return
			}
			out, e := s.MemoryRequest(r.Context(), p, op, raw)
			s.write(w, r, out, e)
		})))
	}
}
func (s *Service) write(w http.ResponseWriter, r *http.Request, data any, err error) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.Header().Set("Cache-Control", "no-store")
	id := r.Header.Get("X-Request-ID")
	if id == "" {
		id = uuid.NewString()
	}
	out := map[string]any{"ok": err == nil, "code": "OK", "request_id": id, "server_time_ms": s.now().UnixMilli()}
	status := 200
	if err == nil {
		out["data"] = data
	} else {
		e := &Error{}
		if !errors.As(err, &e) {
			e = fault(503, "SERVICE_UNAVAILABLE")
			s.logger.Error("call memory request failed", "request_id", id, "path", r.URL.Path)
		}
		status = e.Status
		out["code"] = e.Code
		out["message"] = e.Message
		out["recovery_action"] = e.Recovery
		if status == 503 {
			out["retry_after_ms"] = 2000
		}
	}
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(out)
}
func validUUID(s string) bool { u, e := uuid.Parse(s); return e == nil && u.String() == s }
func short(s string) bool {
	return len(s) > 0 && len(s) <= 128 && strings.IndexFunc(s, func(r rune) bool { return r < 33 || r > 126 }) < 0
}

type CallKey struct {
	SessionID  *string `json:"business_session_id,omitempty"`
	BootID     string  `json:"boot_id"`
	HookID     string  `json:"hook_event_id,omitempty"`
	Generation int64   `json:"call_generation"`
	Binding    int64   `json:"binding_generation"`
}
type StatusInput struct {
	SessionID *string `json:"business_session_id"`
	BootID    string  `json:"boot_id"`
	HookID    string  `json:"hook_event_id"`
}
type WarmupInput struct {
	BootID       string `json:"boot_id"`
	Binding      int64  `json:"binding_generation"`
	Firmware     string `json:"firmware_version"`
	SDK          string `json:"sdk_version"`
	AudioID      string `json:"audio_profile_id"`
	KnownVersion int64  `json:"known_config_version"`
	Contract     string `json:"contract_version,omitempty"`
}
type PrepareInput struct {
	CallKey
	AudioID string `json:"audio_profile_id"`
	Purpose string `json:"purpose"`
}
type HeartbeatInput struct {
	CallKey
	ManifestID string `json:"manifest_id"`
	AttemptID  string `json:"media_attempt_id"`
	Epoch      int64  `json:"observed_privacy_epoch"`
	Runtime    string `json:"runtime_state"`
	LastSeq    int64  `json:"last_event_seq"`
	Uptime     int64  `json:"uptime_ms"`
}
type EndInput struct {
	CallKey
	Reason       string `json:"reason"`
	Elapsed      int64  `json:"elapsed_ms"`
	LastSeq      int64  `json:"last_event_seq"`
	AudioStopped bool   `json:"local_audio_stopped"`
	SDKStatus    string `json:"sdk_stop_status"`
}
type AckInput struct {
	SessionID        string  `json:"business_session_id"`
	CommandID        string  `json:"command_id"`
	Generation       int64   `json:"call_generation"`
	Phase            string  `json:"phase"`
	ErrorCode        *string `json:"error_code"`
	MutationID       *string `json:"mutation_id,omitempty"`
	MutationRevision *int64  `json:"mutation_revision,omitempty"`
}
type session struct {
	ID, Subject, State, Remote, Transcript, ManifestID, ManifestStatus, B64, MAC, Boot, Hook string
	Generation, Binding, Epoch, Revision, LastSeq                                            int64
	Expires, Lease                                                                           *time.Time
	AudioOff, SDKStopped                                                                     bool
}

func readSession(ctx context.Context, tx pgx.Tx, device string, k CallKey) (session, error) {
	var c session
	sid := ""
	if k.SessionID != nil {
		sid = *k.SessionID
		if !validUUID(sid) {
			return c, fault(400, "INVALID_SESSION")
		}
	}
	e := tx.QueryRow(ctx, `SELECT id::text,COALESCE(subject_id::text,''),call_state,remote_state,transcript_state,COALESCE(manifest_id::text,''),manifest_status,COALESCE(manifest_b64,''),COALESCE(manifest_mac,''),boot_id::text,hook_event_id,call_generation,binding_generation,privacy_epoch,memory_revision,highest_contiguous_seq,expires_at,lease_expires_at,local_audio_off,sdk_stopped FROM call_session WHERE device_id=$1::uuid AND (($2<>'' AND id=NULLIF($2,'')::uuid) OR ($2='' AND boot_id=NULLIF($3,'')::uuid AND hook_event_id=$4)) FOR UPDATE`, device, sid, k.BootID, k.HookID).Scan(&c.ID, &c.Subject, &c.State, &c.Remote, &c.Transcript, &c.ManifestID, &c.ManifestStatus, &c.B64, &c.MAC, &c.Boot, &c.Hook, &c.Generation, &c.Binding, &c.Epoch, &c.Revision, &c.LastSeq, &c.Expires, &c.Lease, &c.AudioOff, &c.SDKStopped)
	if errors.Is(e, pgx.ErrNoRows) {
		return c, fault(404, "CALL_NOT_FOUND")
	}
	return c, e
}
func (s *Service) DeviceRequest(ctx context.Context, d deviceauth.Identity, key []byte, requestID, op string, raw []byte) (any, error) {
	if !validUUID(requestID) || len(raw) > MaxBody || d.EnrollmentMode != "physical_confirm" {
		return nil, fault(400, "CALL_PROTOCOL_REQUIRED")
	}
	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	tx, e := s.pool.Begin(ctx)
	if e != nil {
		return nil, e
	}
	defer tx.Rollback(context.Background())
	// Serializes end-before-prepare and duplicate request IDs for this device.
	if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,11))`, d.ID); e != nil {
		return nil, e
	}
	var cached []byte
	var oldPath, oldHash string
	e = tx.QueryRow(ctx, `SELECT path,body_digest,response FROM call_request_result WHERE actor=$1 AND request_id=$2::uuid`, "device:"+d.ID, requestID).Scan(&oldPath, &oldHash, &cached)
	if e != nil && !errors.Is(e, pgx.ErrNoRows) {
		return nil, e
	}
	if len(cached) > 0 && (oldPath != op || oldHash != hash(raw)) {
		return nil, fault(409, "IDEMPOTENCY_CONFLICT")
	}
	var result any
	switch op {
	case "warmup":
		var in WarmupInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.warmup(ctx, tx, d, in)
		}
	case "prepare":
		var in PrepareInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.prepare(ctx, tx, d, key, in, hash(raw))
		}
	case "heartbeat":
		var in HeartbeatInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.heartbeat(ctx, tx, d, in, cached)
		}
	case "events":
		var in EventsInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.events(ctx, tx, d, in)
		}
	case "end":
		var in EndInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.end(ctx, tx, d, in)
		}
	case "status":
		var in StatusInput
		if e = StrictJSON(raw, &in); e == nil {
			var c session
			c, e = readSession(ctx, tx, d.ID, CallKey{SessionID: in.SessionID, BootID: in.BootID, HookID: in.HookID})
			if e == nil && (c.Boot != in.BootID || c.Hook != in.HookID) {
				e = fault(409, "CALL_IDENTITY_MISMATCH")
			}
			if e == nil {
				result, e = s.status(ctx, tx, c)
			}
		}
	case "command-acks":
		var in AckInput
		if e = StrictJSON(raw, &in); e == nil {
			result, e = s.ack(ctx, tx, d, in)
		}
	default:
		e = fault(404, "NOT_FOUND")
	}
	if e != nil {
		return nil, e
	}
	if len(cached) == 0 {
		_, e = tx.Exec(ctx, `INSERT INTO call_request_result(actor,request_id,path,body_digest,response) VALUES($1,$2::uuid,$3,$4,$5::jsonb)`, "device:"+d.ID, requestID, op, hash(raw), jsonBytes(result))
		if e != nil {
			return nil, e
		}
	}
	if e = tx.Commit(ctx); e != nil {
		return nil, e
	}
	return result, nil
}
func (s *Service) warmup(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, in WarmupInput) (any, error) {
	if !validUUID(in.BootID) || in.Binding != d.BindingGeneration || !short(in.SDK) || !short(in.Firmware) || !short(in.AudioID) || in.KnownVersion < 0 || (in.Contract != "" && in.Contract != ContractVersion) {
		return nil, fault(409, "CALL_PROTOCOL_REQUIRED")
	}
	var profile string
	e := tx.QueryRow(ctx, `SELECT id FROM call_provider_profile WHERE published AND sdk_version=$1 AND audio_profile->>'id'=$2 ORDER BY created_at DESC LIMIT 1`, in.SDK, in.AudioID).Scan(&profile)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil, fault(409, "PROVIDER_PROFILE_UNVERIFIED")
	}
	if e != nil {
		return nil, e
	}
	var allowed bool
	e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device d JOIN device_binding b ON b.device_id=d.id AND b.unbound_at IS NULL WHERE d.id=$1::uuid AND d.onboarding_boot_id=$2::uuid AND d.binding_generation=$3 AND b.generation=$3 AND d.status='registered')`, d.ID, in.BootID, in.Binding).Scan(&allowed)
	if e != nil {
		return nil, e
	}
	if !allowed {
		return nil, fault(409, "BINDING_GENERATION_CONFLICT")
	}
	var configVersion int64
	var lifecycle string
	e = tx.QueryRow(ctx, `SELECT config_version,status FROM device_activation_job WHERE device_id=$1::uuid AND binding_generation=$2 AND config_expires_at>now() AND status NOT IN ('REVOKED','FAILED') ORDER BY created_at DESC LIMIT 1`, d.ID, in.Binding).Scan(&configVersion, &lifecycle)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil, fault(409, "DEVICE_NOT_READY")
	}
	if e != nil {
		return nil, e
	}
	_, e = tx.Exec(ctx, `INSERT INTO call_device_ready(device_id,boot_id,binding_generation,provider_profile_id,firmware_version,expires_at) VALUES($1::uuid,$2::uuid,$3,$4,$5,now()+interval '5 minutes') ON CONFLICT(device_id) DO UPDATE SET boot_id=excluded.boot_id,binding_generation=excluded.binding_generation,provider_profile_id=excluded.provider_profile_id,firmware_version=excluded.firmware_version,expires_at=excluded.expires_at`, d.ID, in.BootID, in.Binding, profile, in.Firmware)
	if e != nil {
		return nil, e
	}
	var subject string
	e = tx.QueryRow(ctx, `SELECT subject_id::text FROM device_binding WHERE device_id=$1::uuid AND generation=$2 AND unbound_at IS NULL`, d.ID, in.Binding).Scan(&subject)
	if e != nil {
		return nil, e
	}
	transcriptAllowed, e := transcriptPermission(ctx, tx, subject)
	return map[string]any{"device_state": map[string]any{"activation_state": d.ActivationState, "binding_generation": in.Binding, "lifecycle_state": lifecycle}, "binding_generation": in.Binding, "config_version": configVersion, "provider_profile_id": profile, "engine_prepare_allowed": true, "transcript_upload_allowed": transcriptAllowed, "poll_after_ms": 30000, "contract_version": ContractVersion}, e
}
func (s *Service) prepare(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, key []byte, in PrepareInput, bodyHash string) (any, error) {
	if !validUUID(in.BootID) || !short(in.HookID) || in.Generation < 0 || in.Binding != d.BindingGeneration || !short(in.AudioID) || (in.Purpose != "NORMAL" && in.Purpose != "FIRST_TEST") {
		return nil, fault(400, "INVALID_PREPARE")
	}
	var subject, tenant, boot, lifecycle string
	var binding int64
	e := tx.QueryRow(ctx, `SELECT b.subject_id::text,b.tenant_id::text,d.onboarding_boot_id::text,d.binding_generation,j.status FROM device d JOIN device_binding b ON b.device_id=d.id AND b.unbound_at IS NULL AND b.generation=d.binding_generation JOIN subject_profile sp ON sp.id=b.subject_id AND sp.status='active' JOIN household h ON h.id=b.household_id AND h.status='active' JOIN device_activation_job j ON j.device_id=d.id AND j.binding_generation=d.binding_generation WHERE d.id=$1::uuid AND d.status='registered' AND j.config_ack_at IS NOT NULL AND j.engine_ready_at IS NOT NULL AND j.sdk_registered_at IS NOT NULL AND j.config_expires_at>now() AND d.activation_state NOT IN ('REVOKED','LOST') AND (j.status<>'READY' OR j.ready_binding_generation=d.binding_generation)`, d.ID).Scan(&subject, &tenant, &boot, &binding, &lifecycle)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil, fault(409, "DEVICE_NOT_READY")
	}
	if e != nil {
		return nil, e
	}
	if boot != in.BootID || binding != in.Binding {
		return nil, fault(409, "BINDING_GENERATION_CONFLICT")
	}
	var voiceAllowed bool
	e = tx.QueryRow(ctx, `SELECT COALESCE((SELECT effective_consents @> '{"voice_chat":true}'::jsonb FROM consent_snapshot WHERE subject_id=$1::uuid ORDER BY snapshot_version DESC LIMIT 1),false) AND NOT EXISTS(SELECT 1 FROM privacy_request WHERE subject_id=$1::uuid AND request_type='DELETE')`, subject).Scan(&voiceAllowed)
	if e != nil {
		return nil, e
	}
	if !voiceAllowed {
		return nil, fault(403, "VOICE_CHAT_CONSENT_REQUIRED")
	}
	if in.Purpose == "NORMAL" && lifecycle != "READY" {
		return nil, fault(409, "DEVICE_NOT_READY")
	}
	if in.Purpose == "FIRST_TEST" {
		var permit bool
		e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_activation_attempt a JOIN device_activation_job j ON j.id=a.activation_id WHERE j.device_id=$1::uuid AND a.binding_generation=$2 AND a.stage='CONVERSATION_TEST' AND a.executor='DEVICE' AND a.status='CONTACT_POSSIBLE' AND a.permit_expires_at>now())`, d.ID, in.Binding).Scan(&permit)
		if e != nil {
			return nil, e
		}
		if !permit {
			return nil, fault(403, "FIRST_TEST_NOT_AUTHORIZED")
		}
	}
	_, e = tx.Exec(ctx, `INSERT INTO memory_user_state(subject_id,tenant_id) VALUES($1::uuid,$2::uuid) ON CONFLICT DO NOTHING`, subject, tenant)
	if e != nil {
		return nil, e
	}
	var revision, visible, epoch int64
	var legacy bool
	e = tx.QueryRow(ctx, `SELECT memory_revision,viking_visible_revision,privacy_epoch,legacy_projection_pending FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, subject).Scan(&revision, &visible, &epoch, &legacy)
	if e != nil {
		return nil, e
	}
	old, findErr := readSession(ctx, tx, d.ID, in.CallKey)
	if findErr == nil {
		if old.State == "CANCELLED" {
			return nil, fault(409, "CALL_CANCELLED")
		}
		var digest string
		if e = tx.QueryRow(ctx, `SELECT prepare_digest FROM call_session WHERE id=$1::uuid`, old.ID).Scan(&digest); e != nil {
			return nil, e
		}
		if digest != bodyHash {
			return nil, fault(409, "IDEMPOTENCY_CONFLICT")
		}
		if old.ManifestStatus != "ISSUED" || old.Epoch != epoch || old.Binding != binding {
			return nil, fault(409, "MANIFEST_REVOKED")
		}
		if old.Expires == nil || !old.Expires.After(s.now()) {
			return nil, fault(410, "PREPARE_EXPIRED")
		}
		return map[string]any{"manifest_b64": old.B64, "manifest_mac": old.MAC}, nil
	}
	var fe *Error
	if !errors.As(findErr, &fe) || fe.Status != 404 {
		return nil, findErr
	}
	var active bool
	e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE device_id=$1::uuid AND manifest_status IN ('ISSUED','REVOKED') AND NOT(local_audio_off AND sdk_stopped AND remote_state IN ('NOT_STARTED','STOPPED_OBSERVED'))) OR EXISTS(SELECT 1 FROM conversation WHERE device_id=$1::uuid AND status IN ('ROOM_ISSUED','STARTING_AGENT','ACTIVE','STOPPING'))`, d.ID).Scan(&active)
	if e != nil {
		return nil, e
	}
	if active {
		return nil, fault(409, "MEDIA_CLEANUP_PENDING")
	}
	budget, e := billing.CheckQuotaTx(ctx, tx, d, s.now())
	if errors.Is(e, billing.ErrQuotaExceeded) {
		return nil, fault(429, "CALL_QUOTA_EXCEEDED")
	}
	if e != nil {
		return nil, e
	}
	p, e := loadProfile(ctx, tx, d.ID, in.BootID, in.Binding)
	if e != nil {
		return nil, e
	}
	if p.Audio["id"] != in.AudioID {
		return nil, fault(409, "AUDIO_PROFILE_MISMATCH")
	}
	var critical, consent bool
	e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM memory_projection_job WHERE subject_id=$1::uuid AND critical AND state<>'VISIBLE'),EXISTS(SELECT 1 FROM consent_record WHERE subject_id=$1::uuid AND consent_type='personalized_memory' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now()))`, subject).Scan(&critical, &consent)
	if e != nil {
		return nil, e
	}
	mode, e := SelectMemoryMode(revision, visible, critical || legacy, consent, p.Capabilities["H05"] && p.Evidence["H05"] != "")
	if e != nil {
		return nil, e
	}
	contextText := ""
	if mode != "OFF" {
		contextText, e = loadContext(ctx, tx, subject)
		if e != nil {
			return nil, e
		}
	}
	sid := uuid.NewString()
	params, e := p.Compile(mode, contextText, "你好，今天想聊些什么？", "u_"+hash([]byte(tenant+":"+subject)), "platform_companion_v2", sid)
	if e != nil {
		return nil, e
	}
	now := s.now()
	maxCall := int64(3600000)
	if v, ok := p.Audio["max_call_ms"].(float64); ok && v > 0 && int64(v) < maxCall {
		maxCall = int64(v)
	}
	if budget != nil && budget.MaxCallDurationSec > 0 && int64(budget.MaxCallDurationSec)*1000 < maxCall {
		maxCall = int64(budget.MaxCallDurationSec) * 1000
	}
	if in.Purpose == "FIRST_TEST" && maxCall > 60000 {
		maxCall = 60000
	}
	m := Manifest{ContractVersion: ContractVersion, ManifestID: uuid.NewString(), SessionID: sid, DeviceID: d.PublicID, BootID: in.BootID, HookID: in.HookID, CallGeneration: in.Generation, BindingGeneration: in.Binding, IssuedAt: now.UnixMilli(), ExpiresAt: now.UnixMilli() + ManifestMS, Mode: "RTC", BotID: p.BotID, ProfileID: p.ID, Purpose: in.Purpose, FirstTurnMode: "AI_WELCOME", MemoryMode: mode, MemoryRevision: revision, VisibleRevision: visible, PrivacyEpoch: epoch, ContextVersion: revision, AudioProfileID: in.AudioID, VendorParams: params, LeaseMS: LeaseMS, LeaseExpiresAt: now.UnixMilli() + LeaseMS, MaxCallMS: maxCall}
	// Physical factory identity is carried by the authentication adapter.
	if strings.HasPrefix(d.CredentialRef, "onboarding://") {
		m.DeviceID = strings.TrimPrefix(d.CredentialRef, "onboarding://")
	}
	b64, mac, e := EncodeManifest(key, m)
	if e != nil {
		return nil, e
	}
	_, e = tx.Exec(ctx, `INSERT INTO call_session(id,device_id,tenant_id,subject_id,boot_id,hook_event_id,call_generation,binding_generation,prepare_digest,purpose,manifest_id,manifest_b64,manifest_mac,manifest_status,provider_profile_id,privacy_epoch,memory_revision,expires_at,lease_expires_at,memory_mode) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,$6,$7,$8,$9,$10,$11::uuid,$12,$13,'ISSUED',$14,$15,$16,$17,$18,$19)`, sid, d.ID, tenant, subject, in.BootID, in.HookID, in.Generation, in.Binding, bodyHash, in.Purpose, m.ManifestID, b64, mac, p.ID, epoch, revision, time.UnixMilli(m.ExpiresAt), time.UnixMilli(m.LeaseExpiresAt), mode)
	if e == nil && budget != nil {
		_, e = tx.Exec(ctx, `UPDATE call_session SET budget_snapshot=$2::jsonb WHERE id=$1::uuid`, sid, jsonBytes(budget))
	}
	if e == nil && in.Purpose == "FIRST_TEST" {
		tag, err := tx.Exec(ctx, `UPDATE device_activation_attempt a SET call_session_id=$1::uuid FROM device_activation_job j WHERE a.activation_id=j.id AND j.device_id=$2::uuid AND a.id=j.current_attempt_id AND a.stage='CONVERSATION_TEST' AND a.status='CONTACT_POSSIBLE' AND a.call_session_id IS NULL AND a.conversation_id IS NULL AND a.permit_expires_at>now()`, sid, d.ID)
		if err != nil {
			return nil, err
		}
		if tag.RowsAffected() != 1 {
			return nil, fault(409, "FIRST_TEST_ALREADY_CONSUMED")
		}
	}
	return map[string]any{"manifest_b64": b64, "manifest_mac": mac}, e
}
func loadProfile(ctx context.Context, tx pgx.Tx, device, boot string, binding int64) (Profile, error) {
	var p Profile
	var audio, template, mapping, caps, evidence []byte
	e := tx.QueryRow(ctx, `SELECT p.id,p.sdk_version,p.bot_id,p.model_version,p.transport,p.audio_profile,p.vendor_template,p.field_mapping,p.capabilities,p.evidence FROM call_provider_profile p JOIN call_device_ready r ON r.provider_profile_id=p.id WHERE r.device_id=$1::uuid AND r.boot_id=$2::uuid AND r.binding_generation=$3 AND r.expires_at>now() AND p.published`, device, boot, binding).Scan(&p.ID, &p.SDKVersion, &p.BotID, &p.ModelVersion, &p.Transport, &audio, &template, &mapping, &caps, &evidence)
	if errors.Is(e, pgx.ErrNoRows) {
		return p, fault(409, "WARMUP_REQUIRED")
	}
	if e != nil {
		return p, e
	}
	for _, pair := range []struct {
		b []byte
		v any
	}{{audio, &p.Audio}, {template, &p.Template}, {mapping, &p.Mapping}, {caps, &p.Capabilities}, {evidence, &p.Evidence}} {
		if e = json.Unmarshal(pair.b, pair.v); e != nil {
			return p, e
		}
	}
	return p, nil
}
func loadContext(ctx context.Context, tx pgx.Tx, subject string) (string, error) {
	rows, e := tx.Query(ctx, `SELECT cm.public_id::text,cm.entity_key,mv.content,cm.memory_type,COALESCE(cm.item_status,''),cm.proactive_policy,cm.version,COALESCE(cm.source_time,cm.updated_at) FROM curated_memory cm JOIN memory_version mv ON mv.id=cm.current_version_id WHERE cm.subject_id=$1::uuid AND cm.status='ACTIVE' AND cm.sensitivity='normal' AND (cm.valid_from IS NULL OR cm.valid_from<=now()) AND (cm.valid_to IS NULL OR cm.valid_to>now()) ORDER BY cm.updated_at DESC LIMIT 200`, subject)
	if e != nil {
		return "", e
	}
	defer rows.Close()
	facts := []ContextFact{}
	for rows.Next() {
		var f ContextFact
		if e = rows.Scan(&f.ID, &f.EntityKey, &f.Text, &f.Type, &f.ItemStatus, &f.Policy, &f.Version, &f.SourceTime); e != nil {
			return "", e
		}
		facts = append(facts, f)
	}
	return BuildContext(facts, 1000), rows.Err()
}