217 lines
8.8 KiB
Go
217 lines
8.8 KiB
Go
|
package elderagent
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/sha256"
|
||
|
|
"encoding/hex"
|
||
|
|
"encoding/json"
|
||
|
|
"fmt"
|
||
|
|
"log/slog"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
)
|
||
|
|
|
||
|
|
const personaVoiceAllowedSQL = `p.status='active' AND h.status='active'
|
||
|
|
AND COALESCE((SELECT effective_consents @> '{"voice_chat":true}'::jsonb FROM consent_snapshot WHERE subject_id=p.id ORDER BY snapshot_version DESC LIMIT 1),false)
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM privacy_request WHERE subject_id=p.id AND request_type='DELETE' AND status NOT IN ('FAILED','REJECTED','CANCELLED'))`
|
||
|
|
|
||
|
|
const personaActiveSQL = `EXISTS(SELECT 1 FROM call_session c WHERE c.subject_id=p.id
|
||
|
|
AND (c.manifest_status IN ('ISSUED','REVOKED') OR c.budget_snapshot->>'protocol'='DIRECT-CHAT-1')
|
||
|
|
AND NOT(c.local_audio_off AND c.sdk_stopped AND (c.remote_state IN ('NOT_STARTED','STOPPED_OBSERVED')
|
||
|
|
OR (COALESCE(c.budget_snapshot->>'protocol','')='DIRECT-CHAT-1' AND c.ended_at IS NOT NULL))))
|
||
|
|
OR EXISTS(SELECT 1 FROM conversation c WHERE c.subject_id=p.id AND c.status IN ('ROOM_ISSUED','ACTIVE','STOPPING'))`
|
||
|
|
|
||
|
|
type personaSnapshot struct {
|
||
|
|
ID, PublicID, Bot, Base, Name, Language, State, Digest, AgentStatus string
|
||
|
|
Config, Agent, Profile, Preferences []byte
|
||
|
|
Revision, Synced int64
|
||
|
|
Attempts int
|
||
|
|
Voice, Memory, PublishedVoice, PublishedMemory, Active, Dirty bool
|
||
|
|
}
|
||
|
|
|
||
|
|
type personaQuerier interface {
|
||
|
|
QueryRow(context.Context, string, ...any) pgx.Row
|
||
|
|
}
|
||
|
|
|
||
|
|
func loadPersona(ctx context.Context, q personaQuerier, subject string) (personaSnapshot, error) {
|
||
|
|
var p personaSnapshot
|
||
|
|
err := q.QueryRow(ctx, `SELECT p.id::text,p.public_id::text,COALESCE(a.bot_id,''),a.persona_base_role,p.preferred_name,p.language,
|
||
|
|
a.config,a.agent_config,p.profile_data,p.conversation_preferences,a.persona_revision,a.persona_synced_revision,
|
||
|
|
a.persona_sync_state,a.persona_sync_attempts,a.persona_digest,(`+personaVoiceAllowedSQL+`),call_memory_authorized(p.id),
|
||
|
|
a.persona_voice_allowed,a.persona_memory_allowed,(`+personaActiveSQL+`),a.persona_session_dirty,a.status
|
||
|
|
FROM subject_voice_agent a JOIN subject_profile p ON p.id=a.subject_id JOIN household h ON h.id=p.household_id
|
||
|
|
WHERE p.id=$1::uuid`, subject).Scan(&p.ID, &p.PublicID, &p.Bot, &p.Base, &p.Name, &p.Language, &p.Config, &p.Agent, &p.Profile, &p.Preferences, &p.Revision, &p.Synced, &p.State, &p.Attempts, &p.Digest, &p.Voice, &p.Memory, &p.PublishedVoice, &p.PublishedMemory, &p.Active, &p.Dirty, &p.AgentStatus)
|
||
|
|
p.Memory = p.Memory && p.Voice
|
||
|
|
return p, err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (p personaSnapshot) configuration() (map[string]any, map[string]any, error) {
|
||
|
|
role, err := compilePersona(p.Base, p.Name, p.Language, p.Profile, p.Preferences, p.Voice, p.Memory)
|
||
|
|
if err != nil {
|
||
|
|
return nil, nil, err
|
||
|
|
}
|
||
|
|
return personaConfiguration(p.Config, p.Agent, role)
|
||
|
|
}
|
||
|
|
|
||
|
|
func configurationDigest(config, agent map[string]any) string {
|
||
|
|
raw, _ := json.Marshal([]any{config, agent})
|
||
|
|
sum := sha256.Sum256(raw)
|
||
|
|
return hex.EncodeToString(sum[:])
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) verify(ctx context.Context, bot string, config, agent map[string]any) error {
|
||
|
|
// The production client reads the provider configuration back; embedded test
|
||
|
|
// providers can use the smaller provisioning interface.
|
||
|
|
if verifier, ok := s.Provider.(interface {
|
||
|
|
VerifyConfiguration(context.Context, string, map[string]any, map[string]any) error
|
||
|
|
}); ok {
|
||
|
|
return verifier.VerifyConfiguration(ctx, bot, config, agent)
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func unlockPersona(conn *pgxpool.Conn, id string) {
|
||
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if _, err := conn.Exec(ctx, `SELECT pg_advisory_unlock(hashtextextended($1,64))`, id); err != nil {
|
||
|
|
_ = conn.Conn().Close(ctx)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Sync attempts immediately without delaying a save behind an active preparation.
|
||
|
|
// The revision and retry time persist independently of this request's lifetime.
|
||
|
|
func (s *Service) Sync(ctx context.Context, publicID string) (Info, error) {
|
||
|
|
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
conn, err := s.Pool.Acquire(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return Info{}, err
|
||
|
|
}
|
||
|
|
defer conn.Release()
|
||
|
|
var id string
|
||
|
|
if err = conn.QueryRow(ctx, `SELECT id::text FROM subject_profile WHERE public_id=$1::uuid`, publicID).Scan(&id); err != nil {
|
||
|
|
return Info{}, err
|
||
|
|
}
|
||
|
|
var locked bool
|
||
|
|
if err = conn.QueryRow(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,64))`, id).Scan(&locked); err != nil {
|
||
|
|
return Info{}, err
|
||
|
|
}
|
||
|
|
if !locked {
|
||
|
|
return s.Get(ctx, publicID)
|
||
|
|
}
|
||
|
|
defer unlockPersona(conn, id)
|
||
|
|
err = s.syncLocked(ctx, conn, id)
|
||
|
|
info, getErr := s.Get(ctx, publicID)
|
||
|
|
if getErr != nil {
|
||
|
|
return info, getErr
|
||
|
|
}
|
||
|
|
return info, err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) syncLocked(ctx context.Context, conn *pgxpool.Conn, id string) error {
|
||
|
|
p, err := loadPersona(ctx, conn, id)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if p.AgentStatus != "READY" {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if p.Active {
|
||
|
|
_, err = conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_sync_state=CASE WHEN persona_revision<>persona_synced_revision OR persona_voice_allowed<>$2 OR persona_memory_allowed<>$3 THEN 'DEFERRED' ELSE persona_sync_state END,persona_next_retry_at=now()+interval '5 seconds' WHERE subject_id=$1::uuid`, id, p.Voice, p.Memory)
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
config, agent, err := p.configuration()
|
||
|
|
if err == nil {
|
||
|
|
err = s.configureNativeMemory(ctx, id, config)
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return s.recordPersonaFailure(conn, p, err)
|
||
|
|
}
|
||
|
|
digest := configurationDigest(config, agent)
|
||
|
|
// Equality includes all provider settings, not only the prompt.
|
||
|
|
if p.Revision == p.Synced && digest == p.Digest && p.Voice == p.PublishedVoice && p.Memory == p.PublishedMemory && !p.Dirty {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
updater, ok := s.Provider.(interface {
|
||
|
|
Update(context.Context, string, map[string]any, map[string]any) error
|
||
|
|
})
|
||
|
|
if !ok {
|
||
|
|
return s.recordPersonaFailure(conn, p, fmt.Errorf("AIBOT_UPDATE_REQUIRED"))
|
||
|
|
}
|
||
|
|
if err = updater.Update(ctx, p.Bot, config, agent); err == nil {
|
||
|
|
err = s.verifyNativeMemory(ctx, p.Bot, config)
|
||
|
|
}
|
||
|
|
if err == nil {
|
||
|
|
err = s.verify(ctx, p.Bot, config, agent)
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return s.recordPersonaFailure(conn, p, err)
|
||
|
|
}
|
||
|
|
// A concurrent consent change may advance the revision while the external call
|
||
|
|
// runs. Never acknowledge that newer version, and leave it queued for cleanup.
|
||
|
|
_, err = conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_synced_revision=$2,persona_sync_state='SYNCED',persona_sync_attempts=0,persona_error_code='',persona_synced_at=now(),persona_digest=$3,persona_memory_allowed=$4,persona_voice_allowed=$5,persona_session_dirty=false WHERE subject_id=$1::uuid AND persona_revision=$2`, id, p.Revision, digest, p.Memory, p.Voice)
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) recordPersonaFailure(conn *pgxpool.Conn, p personaSnapshot, cause error) error {
|
||
|
|
// Provider errors may contain credentials or user text. Persist a fixed code.
|
||
|
|
attempt := min(p.Attempts, 8)
|
||
|
|
delay := min(time.Second*time.Duration(1<<attempt), 5*time.Minute)
|
||
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
_, err := conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_sync_state='FAILED',persona_sync_attempts=LEAST(persona_sync_attempts+1,1000000),persona_error_code='PERSONA_SYNC_FAILED',persona_next_retry_at=$3 WHERE subject_id=$1::uuid AND persona_revision=$2`, p.ID, p.Revision, time.Now().Add(delay))
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
return fmt.Errorf("PERSONA_SYNC_FAILED")
|
||
|
|
}
|
||
|
|
|
||
|
|
// RunPersonaSync also cleans call-specific context after a call has fully stopped,
|
||
|
|
// and detects permission expiry without needing a new user edit.
|
||
|
|
func (s *Service) RunPersonaSync(ctx context.Context, logger *slog.Logger) {
|
||
|
|
ticker := time.NewTicker(2 * time.Second)
|
||
|
|
defer ticker.Stop()
|
||
|
|
for {
|
||
|
|
if err := s.SyncPending(ctx); err != nil && ctx.Err() == nil {
|
||
|
|
logger.Warn("elder persona sync scan failed")
|
||
|
|
}
|
||
|
|
select {
|
||
|
|
case <-ctx.Done():
|
||
|
|
return
|
||
|
|
case <-ticker.C:
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) SyncPending(ctx context.Context) error {
|
||
|
|
rows, err := s.Pool.Query(ctx, `SELECT p.public_id::text FROM subject_voice_agent a JOIN subject_profile p ON p.id=a.subject_id JOIN household h ON h.id=p.household_id
|
||
|
|
WHERE a.status='READY' AND a.persona_next_retry_at<=now() AND (a.persona_sync_state<>'SYNCED' OR a.persona_session_dirty
|
||
|
|
OR a.persona_memory_allowed IS DISTINCT FROM (call_memory_authorized(p.id) AND (`+personaVoiceAllowedSQL+`)) OR a.persona_voice_allowed IS DISTINCT FROM (`+personaVoiceAllowedSQL+`)) ORDER BY a.persona_next_retry_at,a.subject_id LIMIT 10`)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
ids := []string{}
|
||
|
|
for rows.Next() {
|
||
|
|
var id string
|
||
|
|
if err = rows.Scan(&id); err != nil {
|
||
|
|
rows.Close()
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
ids = append(ids, id)
|
||
|
|
}
|
||
|
|
err = rows.Err()
|
||
|
|
rows.Close()
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
for _, id := range ids {
|
||
|
|
if ctx.Err() != nil {
|
||
|
|
return ctx.Err()
|
||
|
|
}
|
||
|
|
_, _ = s.Sync(ctx, id)
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|