880 lines
36 KiB
Go
880 lines
36 KiB
Go
package orchestration
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"data-terminal-cloud/internal/billing"
|
||
"data-terminal-cloud/internal/conversation"
|
||
"data-terminal-cloud/internal/deviceauth"
|
||
"data-terminal-cloud/internal/firmwareversion"
|
||
"data-terminal-cloud/internal/retrieval"
|
||
|
||
"github.com/google/uuid"
|
||
"github.com/jackc/pgx/v5"
|
||
"github.com/jackc/pgx/v5/pgconn"
|
||
)
|
||
|
||
var (
|
||
ErrDeviceNotBound = errors.New("device is not bound to an active subject")
|
||
ErrConsentRequired = errors.New("voice chat consent is required")
|
||
ErrFirmwareUpdateRequired = errors.New("device firmware does not satisfy the minimum version")
|
||
ErrConversationAlreadyActive = errors.New("device already has an unfinished conversation")
|
||
ErrConversationNotFound = errors.New("conversation was not found for device")
|
||
ErrConversationStateConflict = errors.New("conversation state does not allow operation")
|
||
ErrAllocationExpired = errors.New("room allocation expired")
|
||
ErrRTCIdentityMismatch = errors.New("RTC room or user does not match allocation")
|
||
ErrConfigVersionConflict = errors.New("device configuration revision is stale")
|
||
ErrRuntimeConfigUnavailable = errors.New("published runtime routing configuration is unavailable")
|
||
ErrContextPreparationFailed = errors.New("authorized subject context preparation failed")
|
||
ErrUsageQuotaExceeded = billing.ErrQuotaExceeded
|
||
)
|
||
|
||
const contextPreparationFailureCode = "context_preparation_failed"
|
||
|
||
const (
|
||
contextSummaryLimit = 1000
|
||
recentSummaryLimit = 3
|
||
recentSummaryContentLimit = 240
|
||
)
|
||
|
||
type AllocationOptions struct {
|
||
Now time.Time
|
||
AllocationExpires time.Time
|
||
TokenExpires time.Time
|
||
RTCAppID string
|
||
VoiceProvider string
|
||
EngineMode string
|
||
HeartbeatInterval int
|
||
MemoryQueryVector string
|
||
MemoryPolicy string
|
||
RequirePublishedRouting bool
|
||
}
|
||
|
||
type Repository struct{}
|
||
|
||
type ActivationResult struct {
|
||
Response ActivateResponse
|
||
InternalID string
|
||
ProviderTaskRef string
|
||
StartRequested bool
|
||
}
|
||
|
||
type StopResult struct {
|
||
Response StopResponse
|
||
InternalID string
|
||
ProviderTaskRef string
|
||
StopRequested bool
|
||
}
|
||
|
||
func NewRepository() *Repository { return &Repository{} }
|
||
|
||
func (r *Repository) Stop(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
identity deviceauth.Identity,
|
||
conversationPublicID, requestID string,
|
||
input StopInput,
|
||
now time.Time,
|
||
) (StopResult, error) {
|
||
var internalID, providerTaskRef string
|
||
var status conversation.Status
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT c.id::text, c.status, c.provider_task_ref
|
||
FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id
|
||
WHERE c.public_id=$1::uuid AND c.device_id=$2::uuid AND c.tenant_id=$3::uuid
|
||
FOR UPDATE OF c, r`, conversationPublicID, identity.ID, identity.TenantID,
|
||
).Scan(&internalID, &status, &providerTaskRef)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return StopResult{}, ErrConversationNotFound
|
||
}
|
||
if err != nil {
|
||
return StopResult{}, fmt.Errorf("read conversation stop state: %w", err)
|
||
}
|
||
result := StopResult{
|
||
Response: StopResponse{ConversationID: conversationPublicID, Status: status, AcceptedAt: now},
|
||
InternalID: internalID, ProviderTaskRef: providerTaskRef,
|
||
}
|
||
if conversation.IsTerminal(status) {
|
||
return result, nil
|
||
}
|
||
if status == conversation.StatusStopping {
|
||
_, err := tx.Exec(ctx, `
|
||
UPDATE conversation SET stop_reason=COALESCE(stop_reason,$2),
|
||
last_heartbeat_seq=GREATEST(last_heartbeat_seq,$3), updated_at=$4
|
||
WHERE id=$1::uuid`, internalID, input.Reason, input.LastHeartbeatSequence, now)
|
||
if err != nil {
|
||
return StopResult{}, fmt.Errorf("refresh repeated stop: %w", err)
|
||
}
|
||
return result, nil
|
||
}
|
||
if status == conversation.StatusRoomIssued {
|
||
if err := conversation.ValidateTransition(status, conversation.StatusExpired); err != nil {
|
||
return StopResult{}, err
|
||
}
|
||
_, err := tx.Exec(ctx, `
|
||
UPDATE conversation SET status='EXPIRED', stop_reason=$2, ended_at=$3,
|
||
lease_expires_at=NULL, last_heartbeat_seq=GREATEST(last_heartbeat_seq,$4),
|
||
version=version+1, updated_at=$3 WHERE id=$1::uuid AND status='ROOM_ISSUED'`,
|
||
internalID, input.Reason, now, input.LastHeartbeatSequence)
|
||
if err == nil {
|
||
_, err = tx.Exec(ctx, `UPDATE rtc_task SET status='COMPLETED', lease_expires_at=NULL,
|
||
last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, internalID, now)
|
||
}
|
||
if err == nil {
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (
|
||
conversation_id, from_status, to_status, reason, source, request_id
|
||
) VALUES ($1::uuid, 'ROOM_ISSUED', 'EXPIRED', 'device stopped before activation', 'device', NULLIF($2,''))`,
|
||
internalID, requestID)
|
||
}
|
||
if err != nil {
|
||
return StopResult{}, fmt.Errorf("expire unactivated conversation: %w", err)
|
||
}
|
||
result.Response.Status = conversation.StatusExpired
|
||
return result, nil
|
||
}
|
||
if status != conversation.StatusStartingAgent && status != conversation.StatusActive {
|
||
return StopResult{}, ErrConversationStateConflict
|
||
}
|
||
if err := conversation.ValidateTransition(status, conversation.StatusStopping); err != nil {
|
||
return StopResult{}, err
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
UPDATE conversation SET status='STOPPING', stop_reason=$2,
|
||
lease_expires_at=NULL, last_heartbeat_seq=GREATEST(last_heartbeat_seq,$3),
|
||
version=version+1, updated_at=$4 WHERE id=$1::uuid AND status=$5`,
|
||
internalID, input.Reason, input.LastHeartbeatSequence, now, status)
|
||
if err == nil {
|
||
_, err = tx.Exec(ctx, `
|
||
UPDATE rtc_task SET status='STOPPING', lease_expires_at=NULL, updated_at=$2
|
||
WHERE conversation_id=$1::uuid`, internalID, now)
|
||
}
|
||
if err == nil {
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (
|
||
conversation_id, from_status, to_status, reason, source, request_id
|
||
) VALUES ($1::uuid, $2, 'STOPPING', 'device requested conversation stop', 'device', NULLIF($3,''))`,
|
||
internalID, status, requestID)
|
||
}
|
||
if err != nil {
|
||
return StopResult{}, fmt.Errorf("stop active conversation: %w", err)
|
||
}
|
||
result.Response.Status = conversation.StatusStopping
|
||
result.StopRequested = true
|
||
return result, nil
|
||
}
|
||
|
||
func (r *Repository) Activate(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
identity deviceauth.Identity,
|
||
conversationPublicID, requestID string,
|
||
input ActivateInput,
|
||
now time.Time,
|
||
) (ActivationResult, error) {
|
||
var internalID, providerTaskRef, roomID, rtcUserID string
|
||
var status conversation.Status
|
||
var allocationExpires time.Time
|
||
var desiredRevision int64
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT c.id::text, c.status, c.allocation_expires_at, c.provider_task_ref,
|
||
r.room_id, r.device_rtc_user_id, dt.desired_revision
|
||
FROM conversation c
|
||
JOIN rtc_task r ON r.conversation_id = c.id
|
||
JOIN device_twin dt ON dt.device_id = c.device_id AND dt.tenant_id = c.tenant_id
|
||
JOIN device_binding b ON b.device_id = c.device_id AND b.tenant_id = c.tenant_id
|
||
AND b.subject_id = c.subject_id AND b.unbound_at IS NULL
|
||
WHERE c.public_id = $1::uuid AND c.device_id = $2::uuid AND c.tenant_id = $3::uuid
|
||
FOR UPDATE OF c, r`, conversationPublicID, identity.ID, identity.TenantID,
|
||
).Scan(&internalID, &status, &allocationExpires, &providerTaskRef, &roomID, &rtcUserID, &desiredRevision)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return ActivationResult{}, ErrConversationNotFound
|
||
}
|
||
if err != nil {
|
||
return ActivationResult{}, fmt.Errorf("read conversation activation state: %w", err)
|
||
}
|
||
if input.RoomID != roomID || input.RTCUserID != rtcUserID {
|
||
return ActivationResult{}, ErrRTCIdentityMismatch
|
||
}
|
||
if input.RTCJoinResult != "success" || input.JoinedAt.IsZero() || input.JoinedAt.After(now.Add(5*time.Minute)) {
|
||
return ActivationResult{}, ErrRTCIdentityMismatch
|
||
}
|
||
if input.ConfigRevision != desiredRevision {
|
||
return ActivationResult{}, ErrConfigVersionConflict
|
||
}
|
||
result := ActivationResult{
|
||
Response: ActivateResponse{ConversationID: conversationPublicID, Status: status, AcceptedAt: now},
|
||
InternalID: internalID, ProviderTaskRef: providerTaskRef,
|
||
}
|
||
if status == conversation.StatusStartingAgent || status == conversation.StatusActive {
|
||
return result, nil
|
||
}
|
||
if status != conversation.StatusRoomIssued {
|
||
return ActivationResult{}, ErrConversationStateConflict
|
||
}
|
||
if !allocationExpires.After(now) {
|
||
return ActivationResult{}, ErrAllocationExpired
|
||
}
|
||
if err := conversation.ValidateTransition(status, conversation.StatusStartingAgent); err != nil {
|
||
return ActivationResult{}, err
|
||
}
|
||
tag, err := tx.Exec(ctx, `
|
||
UPDATE conversation
|
||
SET status='STARTING_AGENT', version=version+1, updated_at=$2
|
||
WHERE id=$1::uuid AND status='ROOM_ISSUED'`, internalID, now)
|
||
if err != nil {
|
||
return ActivationResult{}, fmt.Errorf("start conversation activation: %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return ActivationResult{}, ErrConversationStateConflict
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
UPDATE rtc_task SET status='STARTING', updated_at=$2
|
||
WHERE conversation_id=$1::uuid AND status='ALLOCATED'`, internalID, now)
|
||
if err != nil {
|
||
return ActivationResult{}, fmt.Errorf("mark RTC task starting: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (
|
||
conversation_id, from_status, to_status, reason, source, request_id
|
||
) VALUES ($1::uuid, 'ROOM_ISSUED', 'STARTING_AGENT', 'device confirmed RTC join', 'device', NULLIF($2,''))`,
|
||
internalID, requestID)
|
||
if err != nil {
|
||
return ActivationResult{}, fmt.Errorf("record activation state: %w", err)
|
||
}
|
||
result.Response.Status = conversation.StatusStartingAgent
|
||
result.StartRequested = true
|
||
return result, nil
|
||
}
|
||
|
||
func (r *Repository) Create(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
identity deviceauth.Identity,
|
||
input CreateInput,
|
||
options AllocationOptions,
|
||
) (CreateResponse, string, error) {
|
||
if options.VoiceProvider == "" {
|
||
options.VoiceProvider = "mock"
|
||
}
|
||
var householdID, householdPublicID, subjectID, subjectPublicID, subjectName string
|
||
var profileDataJSON, conversationPreferencesJSON json.RawMessage
|
||
var minimumFirmwareVersion, currentFirmwareVersion *string
|
||
var profileVersion int64
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT b.household_id::text, h.public_id::text, b.subject_id::text,
|
||
s.public_id::text, s.preferred_name, s.profile_data, s.conversation_preferences, s.profile_version,
|
||
d.minimum_firmware_version, dt.firmware_version
|
||
FROM device d
|
||
JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id
|
||
JOIN device_binding b ON b.device_id = d.id AND b.tenant_id = d.tenant_id AND b.unbound_at IS NULL
|
||
JOIN household h ON h.id = b.household_id AND h.tenant_id = b.tenant_id AND h.status = 'active'
|
||
JOIN subject_profile s ON s.id = b.subject_id AND s.tenant_id = b.tenant_id AND s.status = 'active'
|
||
WHERE d.id = $1::uuid AND d.tenant_id = $2::uuid AND d.status = 'registered'
|
||
FOR UPDATE OF d, b, s, dt`, identity.ID, identity.TenantID,
|
||
).Scan(&householdID, &householdPublicID, &subjectID, &subjectPublicID, &subjectName,
|
||
&profileDataJSON, &conversationPreferencesJSON, &profileVersion, &minimumFirmwareVersion, ¤tFirmwareVersion)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return CreateResponse{}, "", ErrDeviceNotBound
|
||
}
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("read active device binding: %w", err)
|
||
}
|
||
if minimumFirmwareVersion != nil && (currentFirmwareVersion == nil || !firmwareversion.AtLeast(*currentFirmwareVersion, *minimumFirmwareVersion)) {
|
||
return CreateResponse{}, "", ErrFirmwareUpdateRequired
|
||
}
|
||
var activeConversationExists bool
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT EXISTS(
|
||
SELECT 1 FROM conversation
|
||
WHERE device_id=$1::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')
|
||
)`, identity.ID).Scan(&activeConversationExists); err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("check active device conversation: %w", err)
|
||
}
|
||
if activeConversationExists {
|
||
return CreateResponse{}, "", ErrConversationAlreadyActive
|
||
}
|
||
budgetSnapshot, err := billing.CheckQuotaTx(ctx, tx, identity, options.Now)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("check usage quota: %w", err)
|
||
}
|
||
engineMode := options.EngineMode
|
||
promptBundleVersion := "minimal-v2"
|
||
publishedConfigID, configuredEngineMode, err := selectRuntimeConfigurationTx(ctx, tx, identity.TenantID, subjectID, identity.ID, identity.PublicID)
|
||
if err == nil {
|
||
if configuredEngineMode != "s2s_pure" && configuredEngineMode != "s2s_hybrid" && configuredEngineMode != "cascade" {
|
||
return CreateResponse{}, "", fmt.Errorf("published routing configuration contains an invalid engine mode")
|
||
}
|
||
engineMode = configuredEngineMode
|
||
promptBundleVersion = "cfg:" + publishedConfigID
|
||
} else if !errors.Is(err, pgx.ErrNoRows) {
|
||
return CreateResponse{}, "", fmt.Errorf("read published runtime configuration: %w", err)
|
||
} else if options.RequirePublishedRouting {
|
||
return CreateResponse{}, "", ErrRuntimeConfigUnavailable
|
||
}
|
||
|
||
var sourceConsentPublicID string
|
||
var snapshotVersion int64
|
||
var effectiveConsents, sourceConsentIDs json.RawMessage
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT public_id::text, snapshot_version, effective_consents, source_consent_ids
|
||
FROM consent_snapshot
|
||
WHERE tenant_id = $1::uuid AND subject_id = $2::uuid
|
||
AND effective_consents @> '{"voice_chat":true}'::jsonb
|
||
ORDER BY created_at DESC, id DESC
|
||
LIMIT 1`, identity.TenantID, subjectID,
|
||
).Scan(&sourceConsentPublicID, &snapshotVersion, &effectiveConsents, &sourceConsentIDs)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return CreateResponse{}, "", ErrConsentRequired
|
||
}
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("read effective consent: %w", err)
|
||
}
|
||
|
||
consentSnapshotID := uuid.NewString()
|
||
consentSnapshotPublicID := uuid.NewString()
|
||
traceSources, _ := json.Marshal([]string{sourceConsentPublicID})
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO consent_snapshot (
|
||
id, public_id, tenant_id, subject_id, snapshot_version,
|
||
effective_consents, source_consent_ids
|
||
) VALUES ($1::uuid, $2::uuid, $3::uuid, $4::uuid, $5, $6::jsonb, $7::jsonb)`,
|
||
consentSnapshotID, consentSnapshotPublicID, identity.TenantID, subjectID,
|
||
snapshotVersion, effectiveConsents, traceSources,
|
||
)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("freeze consent snapshot: %w", err)
|
||
}
|
||
|
||
var memories []retrieval.Memory
|
||
if options.MemoryPolicy == "" {
|
||
memories, err = retrieval.SelectMemoryForContext(ctx, tx, identity.TenantID, subjectID, subjectName+" "+input.RequestedGoal, 5)
|
||
} else if options.MemoryQueryVector != "" {
|
||
memories, err = retrieval.SelectMemoryForContextVector(ctx, tx, identity.TenantID, subjectID, options.MemoryQueryVector, 5)
|
||
}
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("retrieve conversation memory: %w", err)
|
||
}
|
||
memoryVersionsJSON, _ := json.Marshal(memories)
|
||
memoryContents := make([]string, 0, len(memories))
|
||
for _, memory := range memories {
|
||
memoryContents = append(memoryContents, truncateContextText(memory.Content, 300))
|
||
}
|
||
profileContext, err := authorizedProfileContext(profileDataJSON, conversationPreferencesJSON)
|
||
if err != nil {
|
||
return r.createContextPreparationFailure(ctx, tx, identity, householdID, subjectID, profileVersion,
|
||
consentSnapshotID, memoryVersionsJSON, promptBundleVersion, engineMode,
|
||
privacyFromConsent(effectiveConsents), options)
|
||
}
|
||
var recentSummaries []string
|
||
if consentAllowsConversationContinuity(effectiveConsents) {
|
||
recentSummaries, err = selectRecentConversationSummaries(ctx, tx, identity.TenantID, subjectID, recentSummaryLimit)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("retrieve recent conversation summaries: %w", err)
|
||
}
|
||
}
|
||
contextSummary := buildContextSummary(subjectName, profileContext, recentSummaries, memoryContents)
|
||
contextValue := struct {
|
||
SubjectID string `json:"subject_id"`
|
||
PreferredName string `json:"preferred_name"`
|
||
ProfileVersion int64 `json:"profile_version"`
|
||
RequestedGoal string `json:"requested_goal"`
|
||
ProfileContext authorizedSubjectContext `json:"profile_context"`
|
||
RecentSummaries []string `json:"recent_summaries"`
|
||
MemoryContents []string `json:"memory_contents"`
|
||
RetrievalPolicy string `json:"retrieval_policy"`
|
||
}{subjectPublicID, subjectName, profileVersion, input.RequestedGoal, profileContext, recentSummaries, memoryContents, retrievalPolicy(options.MemoryPolicy)}
|
||
contextJSON, _ := json.Marshal(contextValue)
|
||
contextDigest := sha256.Sum256(contextJSON)
|
||
contextSnapshotID := uuid.NewString()
|
||
contextSnapshotPublicID := uuid.NewString()
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO context_snapshot (
|
||
id, public_id, tenant_id, subject_id, profile_version, consent_snapshot_id,
|
||
memory_versions, prompt_bundle_version, engine_mode, content_digest, sanitized_summary
|
||
) VALUES ($1::uuid, $2::uuid, $3::uuid, $4::uuid, $5, $6::uuid,
|
||
$7::jsonb, $8, $9, $10, $11)`,
|
||
contextSnapshotID, contextSnapshotPublicID, identity.TenantID, subjectID, profileVersion,
|
||
consentSnapshotID, memoryVersionsJSON, promptBundleVersion, engineMode, hex.EncodeToString(contextDigest[:]), contextSummary,
|
||
)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("create context snapshot: %w", err)
|
||
}
|
||
|
||
conversationID := uuid.NewString()
|
||
conversationPublicID := uuid.NewString()
|
||
providerTaskRef := "voice-task-" + uuid.NewString()
|
||
roomID := "r_" + strings.ReplaceAll(uuid.NewString(), "-", "")
|
||
deviceRTCUserID := "d_" + strings.ReplaceAll(identity.PublicID, "-", "")
|
||
agentRTCUserID := "a_" + strings.ReplaceAll(conversationPublicID, "-", "")
|
||
privacy := privacyFromConsent(effectiveConsents)
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation (
|
||
id, public_id, tenant_id, household_id, subject_id, device_id,
|
||
consent_snapshot_id, context_snapshot_id, provider_task_ref,
|
||
engine_mode, retention_mode, rtc_credential_renewal_supported,
|
||
status, allocation_expires_at, created_at, updated_at
|
||
) VALUES (
|
||
$1::uuid, $2::uuid, $3::uuid, $4::uuid, $5::uuid, $6::uuid,
|
||
$7::uuid, $8::uuid, $9, $10, $11, COALESCE($12, false), 'PREPARING', $13, $14, $14
|
||
)`, conversationID, conversationPublicID, identity.TenantID, householdID, subjectID, identity.ID,
|
||
consentSnapshotID, contextSnapshotID, providerTaskRef, engineMode,
|
||
privacy.RetentionMode, input.Capabilities.RTCCredentialRenewal, options.AllocationExpires, options.Now,
|
||
)
|
||
if err != nil {
|
||
var postgresError *pgconn.PgError
|
||
if errors.As(err, &postgresError) && postgresError.ConstraintName == "conversation_one_active_per_device" {
|
||
return CreateResponse{}, "", ErrConversationAlreadyActive
|
||
}
|
||
return CreateResponse{}, "", fmt.Errorf("create conversation: %w", err)
|
||
}
|
||
if err = billing.FreezeBudgetTx(ctx, tx, conversationID, identity.TenantID, budgetSnapshot); err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("freeze conversation usage budget: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (conversation_id, from_status, to_status, reason, source)
|
||
VALUES ($1::uuid, NULL, 'PREPARING', 'create request accepted', 'device')`, conversationID)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("record preparing state: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO rtc_task (
|
||
conversation_id, provider, provider_task_ref, room_id,
|
||
device_rtc_user_id, agent_rtc_user_id, status, lease_expires_at,
|
||
credential_issued_at, credential_expires_at
|
||
) VALUES ($1::uuid, $2, $3, $4, $5, $6, 'ALLOCATED', $7, $8, $9)`,
|
||
conversationID, options.VoiceProvider, providerTaskRef, roomID, deviceRTCUserID, agentRTCUserID,
|
||
options.AllocationExpires, options.Now, options.TokenExpires,
|
||
)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("allocate RTC task: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
UPDATE conversation SET status = 'ROOM_ISSUED', version = version + 1, updated_at = $2
|
||
WHERE id = $1::uuid AND status = 'PREPARING'`, conversationID, options.Now)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("issue conversation room: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (conversation_id, from_status, to_status, reason, source)
|
||
VALUES ($1::uuid, 'PREPARING', 'ROOM_ISSUED', 'room credentials allocated', 'platform')`, conversationID)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("record room issued state: %w", err)
|
||
}
|
||
|
||
_ = householdPublicID
|
||
return CreateResponse{
|
||
ConversationID: conversationPublicID,
|
||
Status: conversation.StatusRoomIssued,
|
||
RTC: RTCGrant{
|
||
AppID: options.RTCAppID, RoomID: roomID, UserID: deviceRTCUserID,
|
||
AgentUserID: agentRTCUserID, ExpiresAt: options.TokenExpires,
|
||
},
|
||
Privacy: privacy,
|
||
Session: Session{
|
||
EngineMode: engineMode, HeartbeatIntervalSecond: options.HeartbeatInterval,
|
||
AllocationExpiresAt: options.AllocationExpires,
|
||
},
|
||
}, conversationID, nil
|
||
}
|
||
|
||
func selectRuntimeConfigurationTx(ctx context.Context, tx pgx.Tx, tenantID, subjectID, deviceID, devicePublicID string) (string, string, error) {
|
||
// Rollouts already bind an immutable configuration version and its business
|
||
// type. ConfigKey is an operator-defined version family, not a runtime
|
||
// discriminator: filtering on a reserved key would make a successfully
|
||
// published custom Agent bundle silently ineffective. Prefer the newest
|
||
// matching Agent bundle at the most specific scope, then retain any direct
|
||
// ROUTING rollout as the compatibility path for installations that have not
|
||
// adopted bundles yet.
|
||
var configID, engineMode string
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT bundle.public_id::text,routing.payload->>'default_engine_mode'
|
||
FROM runtime_config_rollout rr
|
||
JOIN runtime_config_version bundle
|
||
ON bundle.id=rr.config_version_id AND bundle.tenant_id=rr.tenant_id
|
||
JOIN runtime_config_version routing
|
||
ON routing.tenant_id=bundle.tenant_id
|
||
AND routing.public_id::text=bundle.payload->>'routing_version_id'
|
||
AND routing.config_kind='ROUTING'
|
||
AND routing.status IN ('PUBLISHED','RETIRED')
|
||
WHERE rr.tenant_id=$1::uuid
|
||
AND rr.status='ACTIVE'
|
||
AND bundle.config_kind='AGENT_BUNDLE'
|
||
AND bundle.status='PUBLISHED'
|
||
AND (
|
||
rr.scope_type='TENANT'
|
||
OR (rr.scope_type='SUBJECT' AND rr.subject_id=$2::uuid)
|
||
OR (
|
||
rr.scope_type='DEVICE_GROUP'
|
||
AND EXISTS (
|
||
SELECT 1 FROM ops_device_group_member dgm
|
||
JOIN ops_device_group dg ON dg.id=dgm.group_id AND dg.tenant_id=dgm.tenant_id
|
||
WHERE dgm.group_id=rr.device_group_id AND dgm.tenant_id=rr.tenant_id
|
||
AND dgm.device_id=$3::uuid AND dg.status='ACTIVE'
|
||
)
|
||
)
|
||
)
|
||
AND (
|
||
rr.rollout_percent=100
|
||
OR mod((hashtextextended($4,rr.cohort_seed::bigint) & 9223372036854775807),100) < rr.rollout_percent
|
||
)
|
||
AND (
|
||
SELECT count(*) FROM conversation active_conversation
|
||
JOIN context_snapshot active_snapshot ON active_snapshot.id=active_conversation.context_snapshot_id
|
||
WHERE active_conversation.tenant_id=rr.tenant_id
|
||
AND active_conversation.status IN ('PREPARING','ROOM_ISSUED','STARTING_AGENT','ACTIVE','STOPPING')
|
||
AND active_snapshot.prompt_bundle_version='cfg:' || bundle.public_id::text
|
||
) < rr.max_concurrent_calls
|
||
ORDER BY CASE rr.scope_type WHEN 'SUBJECT' THEN 1 WHEN 'DEVICE_GROUP' THEN 2 ELSE 3 END,
|
||
rr.activated_at DESC,rr.id DESC
|
||
LIMIT 1`, tenantID, subjectID, deviceID, devicePublicID).Scan(&configID, &engineMode)
|
||
if err == nil {
|
||
return configID, engineMode, nil
|
||
}
|
||
if !errors.Is(err, pgx.ErrNoRows) {
|
||
return "", "", err
|
||
}
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT routing.public_id::text,routing.payload->>'default_engine_mode'
|
||
FROM runtime_config_rollout rr
|
||
JOIN runtime_config_version routing
|
||
ON routing.id=rr.config_version_id AND routing.tenant_id=rr.tenant_id
|
||
WHERE rr.tenant_id=$1::uuid
|
||
AND rr.status='ACTIVE'
|
||
AND routing.config_kind='ROUTING'
|
||
AND routing.status='PUBLISHED'
|
||
AND (
|
||
rr.scope_type='TENANT'
|
||
OR (rr.scope_type='SUBJECT' AND rr.subject_id=$2::uuid)
|
||
OR (
|
||
rr.scope_type='DEVICE_GROUP'
|
||
AND EXISTS (
|
||
SELECT 1 FROM ops_device_group_member dgm
|
||
JOIN ops_device_group dg ON dg.id=dgm.group_id AND dg.tenant_id=dgm.tenant_id
|
||
WHERE dgm.group_id=rr.device_group_id AND dgm.tenant_id=rr.tenant_id
|
||
AND dgm.device_id=$3::uuid AND dg.status='ACTIVE'
|
||
)
|
||
)
|
||
)
|
||
AND (
|
||
rr.rollout_percent=100
|
||
OR mod((hashtextextended($4,rr.cohort_seed::bigint) & 9223372036854775807),100) < rr.rollout_percent
|
||
)
|
||
AND (
|
||
SELECT count(*) FROM conversation active_conversation
|
||
JOIN context_snapshot active_snapshot ON active_snapshot.id=active_conversation.context_snapshot_id
|
||
WHERE active_conversation.tenant_id=rr.tenant_id
|
||
AND active_conversation.status IN ('PREPARING','ROOM_ISSUED','STARTING_AGENT','ACTIVE','STOPPING')
|
||
AND active_snapshot.prompt_bundle_version='cfg:' || routing.public_id::text
|
||
) < rr.max_concurrent_calls
|
||
ORDER BY CASE rr.scope_type WHEN 'SUBJECT' THEN 1 WHEN 'DEVICE_GROUP' THEN 2 ELSE 3 END,
|
||
rr.activated_at DESC,rr.id DESC
|
||
LIMIT 1`, tenantID, subjectID, deviceID, devicePublicID).Scan(&configID, &engineMode)
|
||
return configID, engineMode, err
|
||
}
|
||
|
||
func (r *Repository) createContextPreparationFailure(
|
||
ctx context.Context,
|
||
tx pgx.Tx,
|
||
identity deviceauth.Identity,
|
||
householdID, subjectID string,
|
||
profileVersion int64,
|
||
consentSnapshotID string,
|
||
memoryVersionsJSON json.RawMessage,
|
||
promptBundleVersion, engineMode string,
|
||
privacy Privacy,
|
||
options AllocationOptions,
|
||
) (CreateResponse, string, error) {
|
||
failureValue := struct {
|
||
FailureCode string `json:"failure_code"`
|
||
ProfileVersion int64 `json:"profile_version"`
|
||
PromptVersion string `json:"prompt_bundle_version"`
|
||
}{contextPreparationFailureCode, profileVersion, promptBundleVersion}
|
||
failureJSON, _ := json.Marshal(failureValue)
|
||
digest := sha256.Sum256(failureJSON)
|
||
contextSnapshotID := uuid.NewString()
|
||
contextSnapshotPublicID := uuid.NewString()
|
||
_, err := tx.Exec(ctx, `
|
||
INSERT INTO context_snapshot (
|
||
id, public_id, tenant_id, subject_id, profile_version, consent_snapshot_id,
|
||
memory_versions, prompt_bundle_version, engine_mode, content_digest, sanitized_summary,
|
||
preparation_status, preparation_failure_code
|
||
) VALUES ($1::uuid, $2::uuid, $3::uuid, $4::uuid, $5, $6::uuid,
|
||
$7::jsonb, $8, $9, $10, $11, 'FAILED', $12)`,
|
||
contextSnapshotID, contextSnapshotPublicID, identity.TenantID, subjectID, profileVersion,
|
||
consentSnapshotID, memoryVersionsJSON, promptBundleVersion, engineMode,
|
||
hex.EncodeToString(digest[:]), "背景资料准备失败，未创建房间", contextPreparationFailureCode)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("create failed context snapshot: %w", err)
|
||
}
|
||
|
||
conversationID := uuid.NewString()
|
||
conversationPublicID := uuid.NewString()
|
||
providerTaskRef := "context-failed-" + uuid.NewString()
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation (
|
||
id, public_id, tenant_id, household_id, subject_id, device_id,
|
||
consent_snapshot_id, context_snapshot_id, provider_task_ref,
|
||
engine_mode, retention_mode, status, allocation_expires_at,
|
||
ended_at, failure_code, created_at, updated_at
|
||
) VALUES (
|
||
$1::uuid, $2::uuid, $3::uuid, $4::uuid, $5::uuid, $6::uuid,
|
||
$7::uuid, $8::uuid, $9, $10, $11, 'FAILED', $12,
|
||
$13, $14, $13, $13
|
||
)`, conversationID, conversationPublicID, identity.TenantID, householdID, subjectID, identity.ID,
|
||
consentSnapshotID, contextSnapshotID, providerTaskRef, engineMode, privacy.RetentionMode,
|
||
options.AllocationExpires, options.Now, contextPreparationFailureCode)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("create failed conversation: %w", err)
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO conversation_state_history (conversation_id, from_status, to_status, reason, source)
|
||
VALUES
|
||
($1::uuid, NULL, 'PREPARING', 'create request accepted', 'device'),
|
||
($1::uuid, 'PREPARING', 'FAILED', 'authorized subject context preparation failed', 'platform')`, conversationID)
|
||
if err != nil {
|
||
return CreateResponse{}, "", fmt.Errorf("record context preparation failure: %w", err)
|
||
}
|
||
return CreateResponse{
|
||
ConversationID: conversationPublicID,
|
||
Status: conversation.StatusFailed,
|
||
FailureCode: contextPreparationFailureCode,
|
||
Privacy: privacy,
|
||
Session: Session{
|
||
EngineMode: engineMode, HeartbeatIntervalSecond: options.HeartbeatInterval,
|
||
AllocationExpiresAt: options.AllocationExpires,
|
||
},
|
||
}, conversationID, nil
|
||
}
|
||
|
||
func retrievalPolicy(policy string) string {
|
||
if policy == "" {
|
||
return "pgvector-normal-v1"
|
||
}
|
||
return policy
|
||
}
|
||
|
||
func truncateContextText(value string, limit int) string {
|
||
runes := []rune(strings.TrimSpace(value))
|
||
if len(runes) <= limit {
|
||
return string(runes)
|
||
}
|
||
return string(runes[:limit]) + "…"
|
||
}
|
||
|
||
type authorizedSubjectContext struct {
|
||
Background string `json:"background,omitempty"`
|
||
Interests []string `json:"interests,omitempty"`
|
||
ConversationGoal string `json:"conversation_goal,omitempty"`
|
||
Pace string `json:"pace,omitempty"`
|
||
PreferredTopics []string `json:"preferred_topics,omitempty"`
|
||
AvoidedTopics []string `json:"avoided_topics,omitempty"`
|
||
}
|
||
|
||
func authorizedProfileContext(profileDataJSON, preferencesJSON json.RawMessage) (authorizedSubjectContext, error) {
|
||
var profileData, preferences map[string]any
|
||
if err := json.Unmarshal(profileDataJSON, &profileData); err != nil {
|
||
return authorizedSubjectContext{}, fmt.Errorf("decode profile data: %w", err)
|
||
}
|
||
if err := json.Unmarshal(preferencesJSON, &preferences); err != nil {
|
||
return authorizedSubjectContext{}, fmt.Errorf("decode conversation preferences: %w", err)
|
||
}
|
||
background, err := strictContextString(profileData, "background", 500)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
interests, err := strictContextStringList(profileData, "interests", 8, 50)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
goal, err := strictContextString(preferences, "conversation_goal", 200)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
pace, err := strictContextString(preferences, "pace", 24)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
preferredTopics, err := strictContextStringList(preferences, "preferred_topics", 8, 50)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
avoidedTopics, err := strictContextStringList(preferences, "avoided_topics", 8, 50)
|
||
if err != nil {
|
||
return authorizedSubjectContext{}, err
|
||
}
|
||
if pace != "slow" && pace != "normal" && pace != "lively" {
|
||
pace = ""
|
||
}
|
||
return authorizedSubjectContext{
|
||
Background: background,
|
||
Interests: interests,
|
||
ConversationGoal: goal,
|
||
Pace: pace,
|
||
PreferredTopics: preferredTopics,
|
||
AvoidedTopics: avoidedTopics,
|
||
}, nil
|
||
}
|
||
|
||
func buildContextSummary(preferredName string, profile authorizedSubjectContext, recentSummaries, memoryContents []string) string {
|
||
parts := make([]string, 0, 10)
|
||
if len(profile.AvoidedTopics) > 0 {
|
||
parts = append(parts, "交流边界：不要主动提及“"+strings.Join(truncateContextItems(profile.AvoidedTopics, 24), "、")+"”；若长者主动提及，先简短共情，再自然转向；紧急安全响应不受此限制")
|
||
}
|
||
parts = append(parts, "称呼："+truncateContextField(preferredName, 60))
|
||
for index, summary := range recentSummaries {
|
||
parts = append(parts, fmt.Sprintf("最近第%d次通话摘要：%s", index+1, truncateContextField(summary, 140)))
|
||
}
|
||
if len(memoryContents) > 0 {
|
||
parts = append(parts, "已人工确认的长期记忆："+strings.Join(truncateContextItems(memoryContents, 30), "；"))
|
||
}
|
||
goal := profile.ConversationGoal
|
||
if goal == "" {
|
||
goal = "进行一次自然、尊重的陪伴交流"
|
||
}
|
||
parts = append(parts, "本次交流目标："+truncateContextField(goal, 80))
|
||
if profile.Pace != "" {
|
||
paceLabels := map[string]string{"slow": "放慢语速并留出回应时间", "normal": "自然语速", "lively": "轻快但保持清晰"}
|
||
parts = append(parts, "交流节奏："+paceLabels[profile.Pace])
|
||
}
|
||
if len(profile.PreferredTopics) > 0 {
|
||
parts = append(parts, "优先话题："+strings.Join(truncateContextItems(profile.PreferredTopics, 15), "、"))
|
||
}
|
||
if profile.Background != "" {
|
||
parts = append(parts, "背景资料："+truncateContextField(profile.Background, 80))
|
||
}
|
||
if len(profile.Interests) > 0 {
|
||
parts = append(parts, "兴趣："+strings.Join(truncateContextItems(profile.Interests, 15), "、"))
|
||
}
|
||
result := strings.Join(parts, "；")
|
||
return truncateContextField(result, contextSummaryLimit)
|
||
}
|
||
|
||
func truncateContextItems(values []string, itemLimit int) []string {
|
||
items := make([]string, 0, len(values))
|
||
for _, value := range values {
|
||
if value = truncateContextField(value, itemLimit); value != "" {
|
||
items = append(items, value)
|
||
}
|
||
}
|
||
return items
|
||
}
|
||
|
||
func truncateContextField(value string, limit int) string {
|
||
runes := []rune(strings.TrimSpace(value))
|
||
if len(runes) <= limit {
|
||
return string(runes)
|
||
}
|
||
if limit <= 1 {
|
||
return "…"
|
||
}
|
||
return string(runes[:limit-1]) + "…"
|
||
}
|
||
|
||
func consentAllowsConversationContinuity(consents json.RawMessage) bool {
|
||
values := map[string]bool{}
|
||
if json.Unmarshal(consents, &values) != nil {
|
||
return false
|
||
}
|
||
return values["personalized_memory"] && values["story_extraction"]
|
||
}
|
||
|
||
func selectRecentConversationSummaries(ctx context.Context, tx pgx.Tx, tenantID, subjectID string, limit int) ([]string, error) {
|
||
if limit <= 0 {
|
||
return nil, nil
|
||
}
|
||
rows, err := tx.Query(ctx, `
|
||
WITH latest_summary AS (
|
||
SELECT DISTINCT ON (summary.conversation_id)
|
||
summary.conversation_id, summary.content, conversation.ended_at, conversation.created_at
|
||
FROM conversation_summary summary
|
||
JOIN conversation ON conversation.id=summary.conversation_id
|
||
AND conversation.tenant_id=summary.tenant_id
|
||
WHERE summary.tenant_id=$1::uuid AND summary.subject_id=$2::uuid
|
||
AND summary.summary_type='conversation_summary' AND summary.status='READY'
|
||
AND conversation.status='COMPLETED'
|
||
ORDER BY summary.conversation_id, summary.generation_version DESC, summary.created_at DESC
|
||
)
|
||
SELECT content FROM latest_summary
|
||
ORDER BY COALESCE(ended_at,created_at) DESC, conversation_id DESC
|
||
LIMIT $3`, tenantID, subjectID, limit)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
result := make([]string, 0, limit)
|
||
for rows.Next() {
|
||
var content string
|
||
if err := rows.Scan(&content); err != nil {
|
||
return nil, err
|
||
}
|
||
if content = truncateContextText(content, recentSummaryContentLimit); content != "" {
|
||
result = append(result, content)
|
||
}
|
||
}
|
||
return result, rows.Err()
|
||
}
|
||
|
||
func strictContextString(values map[string]any, key string, limit int) (string, error) {
|
||
raw, exists := values[key]
|
||
if !exists || raw == nil {
|
||
return "", nil
|
||
}
|
||
value, ok := raw.(string)
|
||
if !ok {
|
||
return "", fmt.Errorf("%s must be a string", key)
|
||
}
|
||
return truncateContextText(value, limit), nil
|
||
}
|
||
|
||
func strictContextStringList(values map[string]any, key string, maxItems, itemLimit int) ([]string, error) {
|
||
raw, exists := values[key]
|
||
if !exists || raw == nil {
|
||
return nil, nil
|
||
}
|
||
rawItems, ok := raw.([]any)
|
||
if !ok {
|
||
return nil, fmt.Errorf("%s must be an array of strings", key)
|
||
}
|
||
items := make([]string, 0, min(len(rawItems), maxItems))
|
||
for _, raw := range rawItems {
|
||
value, ok := raw.(string)
|
||
if !ok {
|
||
return nil, fmt.Errorf("%s must contain only strings", key)
|
||
}
|
||
value = truncateContextText(value, itemLimit)
|
||
if value == "" {
|
||
continue
|
||
}
|
||
items = append(items, value)
|
||
if len(items) == maxItems {
|
||
break
|
||
}
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func privacyFromConsent(consents json.RawMessage) Privacy {
|
||
values := map[string]bool{}
|
||
_ = json.Unmarshal(consents, &values)
|
||
privacy := Privacy{
|
||
RecordNear: values["raw_audio_storage"] || values["audio_recording"],
|
||
RecordFar: values["raw_audio_storage"] || values["audio_recording"],
|
||
UploadAllowed: values["raw_audio_storage"] || values["audio_recording"], RetentionMode: "none",
|
||
}
|
||
if values["transcript_storage"] || values["transcript"] {
|
||
privacy.RetentionMode = "transcript_only"
|
||
}
|
||
if values["raw_audio_storage"] || values["audio_recording"] {
|
||
privacy.RetentionMode = "transcript_and_audio"
|
||
}
|
||
return privacy
|
||
}
|