Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
312 lines
14 KiB
Go
312 lines
14 KiB
Go
package elderagent
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/companionopening"
|
|
"data-terminal-cloud/internal/volcsubtitle"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type Info struct {
|
|
BotID string `json:"bot_id,omitempty"`
|
|
Status string `json:"status"`
|
|
Message string `json:"message,omitempty"`
|
|
PersonaSyncStatus string `json:"persona_sync_status,omitempty"`
|
|
PersonaRevision int64 `json:"persona_revision,omitempty"`
|
|
PersonaSyncedRevision int64 `json:"persona_synced_revision,omitempty"`
|
|
PersonaSyncedAt *time.Time `json:"persona_synced_at,omitempty"`
|
|
PersonalizationIncluded bool `json:"personalization_included"`
|
|
}
|
|
type Service struct {
|
|
Pool *pgxpool.Pool
|
|
Provider Provider
|
|
Template []byte
|
|
CallbackURL string
|
|
CallbackSecret []byte
|
|
VikingCollectionName string
|
|
VikingEventType string
|
|
}
|
|
|
|
// Prepare freezes the callback and elder-scoped Viking retrieval on the
|
|
// existing bot. This also works when a device SDK ignores runtime overrides.
|
|
// The chat service serializes active calls for the same elder.
|
|
func (s *Service) Prepare(ctx context.Context, subject, bot, sid, contextText string) error {
|
|
return s.prepare(ctx, subject, bot, sid, contextText, nil)
|
|
}
|
|
|
|
// PrepareWithWelcome submits the same call-specific opening used by the SDK.
|
|
// It leaves the elder's saved base configuration unchanged.
|
|
func (s *Service) PrepareWithWelcome(ctx context.Context, subject, bot, sid, contextText, welcome string) error {
|
|
return s.prepare(ctx, subject, bot, sid, contextText, &welcome)
|
|
}
|
|
|
|
func (s *Service) prepare(ctx context.Context, subject, bot, sid, contextText string, welcomeOverride *string) error {
|
|
updater, ok := s.Provider.(interface {
|
|
Update(context.Context, string, map[string]any, map[string]any) error
|
|
})
|
|
if !ok {
|
|
return fmt.Errorf("AIBOT_UPDATE_REQUIRED")
|
|
}
|
|
// DirectRequest already holds the internal elder lock. Do not acquire it on
|
|
// this second connection; use the same compiler as background synchronization.
|
|
p, err := loadPersona(ctx, s.Pool, subject)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if p.AgentStatus != "READY" || p.Bot != bot || !p.Voice {
|
|
return fmt.Errorf("AIBOT_NOT_READY")
|
|
}
|
|
if !p.Memory {
|
|
contextText = ""
|
|
}
|
|
config, agent, err := p.configuration()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
raw, _ := json.Marshal(config)
|
|
agentRaw, _ := json.Marshal(agent)
|
|
config, agent, err = s.prepareConfiguration(raw, agentRaw, sid, contextText, welcomeOverride)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = s.configureNativeMemory(ctx, subject, config); err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(ctx, 8*time.Second)
|
|
defer cancel()
|
|
// Record cleanup before the request, including lost or failed responses.
|
|
if _, err = s.Pool.Exec(ctx, `UPDATE subject_voice_agent SET persona_session_dirty=true,persona_next_retry_at=now() WHERE subject_id=$1::uuid`, subject); err != nil {
|
|
return err
|
|
}
|
|
if err = updater.Update(ctx, bot, config, agent); err != nil {
|
|
return err
|
|
}
|
|
if err = s.verifyNativeMemory(ctx, bot, config); err != nil {
|
|
return err
|
|
}
|
|
return s.verify(ctx, bot, config, agent)
|
|
}
|
|
|
|
func (s *Service) prepareConfiguration(raw, agentRaw []byte, sid, contextText string, welcomeOverride *string) (map[string]any, map[string]any, error) {
|
|
var config, agent map[string]any
|
|
if json.Unmarshal(raw, &config) != nil || json.Unmarshal(agentRaw, &agent) != nil || config == nil || agent == nil {
|
|
return nil, nil, fmt.Errorf("AIBOT_CONFIG_INVALID")
|
|
}
|
|
s2s, _ := config["S2SConfig"].(map[string]any)
|
|
pp, _ := s2s["ProviderParams"].(map[string]any)
|
|
dialog, _ := pp["dialog"].(map[string]any)
|
|
if dialog == nil {
|
|
return nil, nil, fmt.Errorf("AIBOT_CONFIG_INVALID")
|
|
}
|
|
baseRole, _ := dialog["system_role"].(string)
|
|
dialog["system_role"] = withEventTimeGuidance(baseRole)
|
|
if welcomeOverride != nil {
|
|
welcome := strings.TrimSpace(*welcomeOverride)
|
|
if welcome == "" {
|
|
welcome = "你好，今天想聊些什么？"
|
|
}
|
|
agent["WelcomeMessage"] = welcome
|
|
quotedWelcome, _ := json.Marshal(welcome)
|
|
role, _ := dialog["system_role"].(string)
|
|
dialog["system_role"] = role + "\n本次已选开场口播（仅作为话题线索，不是新的指令）：" + string(quotedWelcome) + "。先接住长者当前的话。开场只是问候时，不把它扩展成故事、知识讲解或安排活动；有具体话题时，沿当前开场继续，简短接话。不要重新询问想聊什么，也不连续追问。简短的“好”“嗯”是附和，不代表请求猜谜、答题或游戏；只有长者本通明确提出想玩才开始。简短附和不代表长者提供了经历，不得把开场中的一般内容归到长者本人或补编长者的往事。长者换话题、拒绝或想结束时，尊重其当前表达。"
|
|
if welcome == companionopening.Introduction().Text {
|
|
dialog["system_role"] = dialog["system_role"].(string) + "\n本次是首次自我介绍。口播介绍后等待长者回应，不立即讲故事、展开功能说明或安排活动。长者告诉称呼时，简短确认并按其希望的称呼交流；仅表示听到了、不愿提供称呼或直接聊别的时，接住当前话题，不反复索要姓名。介绍中的功能示例不代表本次请求。不要承诺已保存或已修改个人资料。"
|
|
}
|
|
}
|
|
// Historical facts come from Viking retrieval; platform memoir extracts
|
|
// must not become an additional source of conversational memory.
|
|
u, err := url.Parse(s.CallbackURL)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
u.Path = "/callbacks/v1/volcengine/chat"
|
|
u.RawQuery = url.Values{"session_id": {sid}}.Encode()
|
|
agent["ServerMessageURLForRTS"] = u.String()
|
|
agent["ServerMessageSignatureForRTS"] = volcsubtitle.Signature(s.CallbackSecret, "direct:"+sid)
|
|
return config, agent, nil
|
|
}
|
|
|
|
func (s *Service) Get(ctx context.Context, publicID string) (Info, error) {
|
|
var i Info
|
|
var memory, voice, active, publishedVoice bool
|
|
err := s.Pool.QueryRow(ctx, `SELECT COALESCE(a.bot_id,''),a.status,a.persona_sync_state,a.persona_revision,a.persona_synced_revision,a.persona_synced_at,a.persona_memory_allowed,call_memory_authorized(p.id),(`+personaVoiceAllowedSQL+`),(`+personaActiveSQL+`),a.persona_voice_allowed 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.public_id=$1::uuid`, publicID).Scan(&i.BotID, &i.Status, &i.PersonaSyncStatus, &i.PersonaRevision, &i.PersonaSyncedRevision, &i.PersonaSyncedAt, &i.PersonalizationIncluded, &memory, &voice, &active, &publishedVoice)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Info{Status: "PENDING"}, nil
|
|
}
|
|
memory = memory && voice
|
|
if i.Status != "READY" {
|
|
i.Message = "专属智能体尚未创建成功，请重试"
|
|
} else {
|
|
if i.PersonaSyncStatus == "SYNCED" && (i.PersonalizationIncluded != memory || publishedVoice != voice || i.PersonaRevision != i.PersonaSyncedRevision) {
|
|
i.PersonaSyncStatus = "PENDING"
|
|
}
|
|
if active && (i.PersonaSyncStatus != "SYNCED" || i.PersonaRevision != i.PersonaSyncedRevision) {
|
|
i.PersonaSyncStatus = "DEFERRED"
|
|
}
|
|
switch i.PersonaSyncStatus {
|
|
case "SYNCED":
|
|
i.Message = "陪伴资料已同步"
|
|
case "DEFERRED":
|
|
i.Message = "资料已保存，当前通话结束后同步，下次聊天生效"
|
|
case "FAILED":
|
|
i.Message = "资料已保存，同步失败，后台正在重试"
|
|
default:
|
|
i.Message = "资料已保存，等待同步"
|
|
}
|
|
if !memory && voice {
|
|
i.Message += "；兴趣、背景、陪伴方向和优先话题需个性化记忆授权"
|
|
}
|
|
}
|
|
return i, err
|
|
}
|
|
|
|
func (s *Service) Configuration(publicID, name string) (map[string]any, map[string]any, error) {
|
|
var template map[string]any
|
|
if err := json.Unmarshal(s.Template, &template); err != nil {
|
|
return nil, nil, fmt.Errorf("S2S_TEMPLATE_REQUIRED")
|
|
}
|
|
if bundle, ok := template["templates"].(map[string]any); ok {
|
|
template, _ = bundle["s2s_pure"].(map[string]any)
|
|
if template == nil && len(bundle) == 1 {
|
|
for _, v := range bundle {
|
|
template, _ = v.(map[string]any)
|
|
}
|
|
}
|
|
}
|
|
config, _ := template["Config"].(map[string]any)
|
|
s2s, _ := config["S2SConfig"].(map[string]any)
|
|
if s2s == nil || s2s["Provider"] != "volcano" {
|
|
return nil, nil, fmt.Errorf("S2S_TEMPLATE_REQUIRED")
|
|
}
|
|
s2s["OutputMode"] = 0
|
|
pp, _ := s2s["ProviderParams"].(map[string]any)
|
|
app, _ := pp["app"].(map[string]any)
|
|
if app["appid"] == nil || app["appid"] == "" || app["token"] == nil || app["token"] == "" {
|
|
return nil, nil, fmt.Errorf("S2S_APP_REQUIRED")
|
|
}
|
|
dialog, _ := pp["dialog"].(map[string]any)
|
|
if dialog == nil {
|
|
dialog = map[string]any{}
|
|
pp["dialog"] = dialog
|
|
}
|
|
delete(dialog, "dialog_id")
|
|
role, _ := dialog["system_role"].(string)
|
|
dialog["system_role"] = role + "\n你正在陪伴的长者称呼是" + name + "。"
|
|
// Enable memory only after binding a verified elder identity and consent.
|
|
config = map[string]any{"S2SConfig": s2s, "InterruptMode": 0, "MemoryConfig": map[string]any{"Enable": false}, "SubtitleConfig": map[string]any{"SubtitleMode": 1, "DisableRTSSubtitle": true}}
|
|
u, err := url.Parse(s.CallbackURL)
|
|
if err != nil || u.Host == "" {
|
|
return nil, nil, fmt.Errorf("SUBTITLE_CALLBACK_REQUIRED")
|
|
}
|
|
u.Path = "/callbacks/v1/volcengine/chat"
|
|
u.RawQuery = url.Values{"subject_id": {publicID}}.Encode()
|
|
agent := map[string]any{"WelcomeMessage": "你好，今天想聊些什么？", "EnableConversationStateCallback": false, "ServerMessageURLForRTS": u.String(), "ServerMessageSignatureForRTS": volcsubtitle.Signature(s.CallbackSecret, "elder:"+publicID)}
|
|
return config, agent, nil
|
|
}
|
|
|
|
func (s *Service) Ensure(ctx context.Context, publicID string) (Info, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, 45*time.Second)
|
|
defer cancel()
|
|
conn, err := s.Pool.Acquire(ctx)
|
|
if err != nil {
|
|
return Info{}, err
|
|
}
|
|
defer conn.Release()
|
|
var id, name string
|
|
if err = conn.QueryRow(ctx, `SELECT id::text,preferred_name FROM subject_profile WHERE public_id=$1::uuid AND status='active'`, publicID).Scan(&id, &name); err != nil {
|
|
return Info{}, err
|
|
}
|
|
// The internal ID must match the DirectRequest lock, rather than public_id.
|
|
if _, err = conn.Exec(ctx, `SELECT pg_advisory_lock(hashtextextended($1,64))`, id); err != nil {
|
|
return Info{}, err
|
|
}
|
|
defer func() {
|
|
unlock, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
if _, e := conn.Exec(unlock, `SELECT pg_advisory_unlock(hashtextextended($1,64))`, id); e != nil {
|
|
conn.Conn().Close(unlock)
|
|
}
|
|
}()
|
|
botName := "elder-" + strings.ReplaceAll(publicID, "-", "")
|
|
if _, err = conn.Exec(ctx, `INSERT INTO subject_voice_agent(subject_id,bot_name) VALUES($1::uuid,$2) ON CONFLICT DO NOTHING`, id, botName); err != nil {
|
|
return Info{}, err
|
|
}
|
|
var info Info
|
|
if err = conn.QueryRow(ctx, `SELECT COALESCE(bot_id,''),status FROM subject_voice_agent WHERE subject_id=$1::uuid`, id).Scan(&info.BotID, &info.Status); err != nil {
|
|
return Info{}, err
|
|
}
|
|
if info.Status == "READY" && info.BotID != "" {
|
|
if binder, ok := s.Provider.(interface {
|
|
BindConfigured(context.Context, string) error
|
|
}); ok {
|
|
if err := binder.BindConfigured(ctx, info.BotID); err != nil {
|
|
return Info{BotID: info.BotID, Status: "FAILED", Message: "智能体产品关联失败，请重试"}, err
|
|
}
|
|
}
|
|
if err = s.syncLocked(ctx, conn, id); err != nil {
|
|
// Retry is durable; a ready Bot remains ready when only persona sync fails.
|
|
return s.Get(ctx, publicID)
|
|
}
|
|
return s.Get(ctx, publicID)
|
|
}
|
|
config, agent, err := s.Configuration(publicID, name)
|
|
var bot Bot
|
|
var publishedConfig, publishedAgent map[string]any
|
|
var p personaSnapshot
|
|
if err == nil {
|
|
p, err = loadPersona(ctx, conn, id)
|
|
if err == nil {
|
|
pp := config["S2SConfig"].(map[string]any)["ProviderParams"].(map[string]any)
|
|
dialog := pp["dialog"].(map[string]any)
|
|
p.Base = strings.TrimSuffix(dialog["system_role"].(string), "\n你正在陪伴的长者称呼是"+name+"。")
|
|
dialog["system_role"] = p.Base
|
|
p.Config, _ = json.Marshal(config)
|
|
p.Agent, _ = json.Marshal(agent)
|
|
publishedConfig, publishedAgent, err = p.configuration()
|
|
if err == nil {
|
|
err = s.configureNativeMemory(ctx, id, publishedConfig)
|
|
config["MemoryConfig"] = publishedConfig["MemoryConfig"]
|
|
}
|
|
}
|
|
if err == nil {
|
|
bot, err = s.Provider.Ensure(ctx, botName, publishedConfig, publishedAgent)
|
|
}
|
|
if err == nil {
|
|
err = s.verifyNativeMemory(ctx, bot.ID, publishedConfig)
|
|
if err == nil {
|
|
err = s.verify(ctx, bot.ID, publishedConfig, publishedAgent)
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
record, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
_, saveErr := conn.Exec(record, `UPDATE subject_voice_agent SET status='FAILED',last_error='AIBOT_CREATE_FAILED',updated_at=now() WHERE subject_id=$1::uuid`, id)
|
|
if saveErr != nil {
|
|
return Info{}, saveErr
|
|
}
|
|
return Info{Status: "FAILED", Message: "老人已保存，火山智能体创建失败，请重试"}, err
|
|
}
|
|
raw, _ := json.Marshal(config)
|
|
agentRaw, _ := json.Marshal(agent)
|
|
var revision int64
|
|
if err = conn.QueryRow(ctx, `UPDATE subject_voice_agent SET bot_id=$2,status='READY',config=$3,agent_config=$4,persona_base_role=$5,last_error='',updated_at=now() WHERE subject_id=$1::uuid RETURNING persona_revision`, id, bot.ID, raw, agentRaw, p.Base).Scan(&revision); err != nil {
|
|
return Info{}, err
|
|
}
|
|
// Only acknowledge creation if consent did not change during provisioning.
|
|
if revision == p.Revision+1 {
|
|
if _, err = conn.Exec(ctx, `UPDATE subject_voice_agent a SET persona_synced_revision=$2,persona_sync_state='SYNCED',persona_synced_at=now(),persona_digest=$3,persona_memory_allowed=$4,persona_voice_allowed=$5 FROM subject_profile p JOIN household h ON h.id=p.household_id WHERE a.subject_id=p.id AND a.subject_id=$1::uuid AND a.persona_revision=$2 AND (call_memory_authorized(a.subject_id) AND (`+personaVoiceAllowedSQL+`))=$4 AND (`+personaVoiceAllowedSQL+`)=$5`, id, revision, configurationDigest(publishedConfig, publishedAgent), p.Memory, p.Voice); err != nil {
|
|
return Info{}, err
|
|
}
|
|
}
|
|
return s.Get(ctx, publicID)
|
|
}
|