460 lines
21 KiB
Go
460 lines
21 KiB
Go
|
package callmemory
|
||
|
|
|
||
|
|
// Direct chat keeps audio and subtitles off the device/platform data path:
|
||
|
|
// the device obtains SDK parameters; Volcano sends signed subtitles to us.
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/subtle"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"io"
|
||
|
|
"net/http"
|
||
|
|
"net/url"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"data-terminal-cloud/internal/billing"
|
||
|
|
"data-terminal-cloud/internal/deviceactivation"
|
||
|
|
"data-terminal-cloud/internal/deviceauth"
|
||
|
|
"data-terminal-cloud/internal/volcsubtitle"
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
type DirectOptions struct {
|
||
|
|
Prepare func(context.Context, string, string, string, string) error
|
||
|
|
PrepareWithWelcome func(context.Context, string, string, string, string, string) error
|
||
|
|
Template []byte
|
||
|
|
BotID, CallbackURL string
|
||
|
|
CallbackSecret []byte
|
||
|
|
}
|
||
|
|
type DirectInput struct {
|
||
|
|
DeviceID string `json:"device_id,omitempty"`
|
||
|
|
SessionID string `json:"session_id"`
|
||
|
|
BootID string `json:"boot_id,omitempty"` // Accepted for old firmware only; no startup registration required.
|
||
|
|
Reason string `json:"reason,omitempty"`
|
||
|
|
SDKStopped bool `json:"sdk_stopped,omitempty"`
|
||
|
|
RoomID string `json:"room_id,omitempty"`
|
||
|
|
RTCAppID string `json:"rtc_app_id,omitempty"`
|
||
|
|
RTCUserID string `json:"rtc_user_id,omitempty"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) ConfigureDirect(o DirectOptions) { s.direct = o }
|
||
|
|
|
||
|
|
func (s *Service) RegisterDirectRoutes(mux *http.ServeMux, auth DeviceAuth) {
|
||
|
|
for _, op := range []string{"start", "end", "recording"} {
|
||
|
|
mux.HandleFunc("POST /api/v1/device/chat/"+op, func(w http.ResponseWriter, r *http.Request) {
|
||
|
|
d, _, raw, e := auth(r)
|
||
|
|
if e != nil {
|
||
|
|
s.write(w, r, nil, fault(401, "DEVICE_AUTHENTICATION_FAILED"))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
var in DirectInput
|
||
|
|
if len(raw) > 2048 || StrictJSON(raw, &in) != nil {
|
||
|
|
s.write(w, r, nil, fault(400, "INVALID_CHAT_REQUEST"))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
out, e := s.DirectRequest(r.Context(), d, op, in)
|
||
|
|
s.write(w, r, out, e)
|
||
|
|
})
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) DirectRequest(ctx context.Context, d deviceauth.Identity, op string, in DirectInput) (any, error) {
|
||
|
|
if (op != "recording" && (in.RoomID != "" || in.RTCAppID != "" || in.RTCUserID != "")) || len(in.RoomID) > 128 || len(in.RTCAppID) > 128 || len(in.RTCUserID) > 128 {
|
||
|
|
return nil, fault(400, "INVALID_CHAT_REQUEST")
|
||
|
|
}
|
||
|
|
// Legacy firmware already supplies its factory ID in the authenticated header.
|
||
|
|
// New firmware also includes it explicitly in the minimal chat body.
|
||
|
|
if in.DeviceID != "" && in.DeviceID != strings.TrimPrefix(d.CredentialRef, "onboarding://") {
|
||
|
|
return nil, fault(400, "DEVICE_ID_MISMATCH")
|
||
|
|
}
|
||
|
|
if !validUUID(in.SessionID) || (op == "start" && ((in.BootID != "" && !validUUID(in.BootID)) || in.Reason != "" || in.SDKStopped)) || (op == "end" && !oneOf(in.Reason, "hangup", "start_failed", "disconnected", "reboot", "timeout")) {
|
||
|
|
return nil, fault(400, "INVALID_CHAT_REQUEST")
|
||
|
|
}
|
||
|
|
ctx, cancel := context.WithTimeout(ctx, 12*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
tx, e := s.pool.Begin(ctx)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
defer tx.Rollback(context.Background())
|
||
|
|
if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,11))`, d.ID); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if op == "recording" {
|
||
|
|
out, err := s.prepareMixedRecording(ctx, tx, d, in)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return out, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
if op == "end" {
|
||
|
|
// End is durable even when it wins a race against start. No media claim is
|
||
|
|
// inferred from a lost HTTP response or a local SDK return code.
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_request_result(actor,request_id,path,body_digest,response) VALUES($1,$2::uuid,'direct-end',$3,'{}') ON CONFLICT DO NOTHING`, "device:"+d.ID, in.SessionID, hash(jsonBytes(in)))
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET call_state=CASE WHEN $3='hangup' THEN 'ENDED' ELSE 'FAILED' END,manifest_status='DISCARDED',manifest_b64=NULL,local_audio_off=true,sdk_stopped=$4,ended_at=COALESCE(ended_at,now()),drain_started_at=COALESCE(drain_started_at,now()),transcript_state=CASE WHEN transcript_state='COLLECTING' THEN 'DRAINING' ELSE transcript_state END WHERE id=$1::uuid AND device_id=$2::uuid AND budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND ended_at IS NULL`, in.SessionID, d.ID, in.Reason, in.SDKStopped)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if e = s.activateDirect(ctx, tx, in.SessionID); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
s.stopMixedRecording(ctx, tx, in.SessionID, d.ID)
|
||
|
|
return map[string]any{"session_id": in.SessionID, "ended": true}, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
if op != "start" {
|
||
|
|
return nil, fault(404, "NOT_FOUND")
|
||
|
|
}
|
||
|
|
var cancelled bool
|
||
|
|
if e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_request_result WHERE actor=$1 AND request_id=$2::uuid AND path='direct-end')`, "device:"+d.ID, in.SessionID).Scan(&cancelled); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if cancelled {
|
||
|
|
return nil, fault(409, "CHAT_ENDED")
|
||
|
|
}
|
||
|
|
var subject, tenant, name, bot string
|
||
|
|
var generation, epoch int64
|
||
|
|
var bound bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding b JOIN device d ON d.id=b.device_id AND d.binding_generation=b.generation JOIN subject_profile sp ON sp.id=b.subject_id AND sp.status='active' WHERE d.id=$1::uuid AND b.unbound_at IS NULL)`, d.ID).Scan(&bound)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !bound {
|
||
|
|
return nil, &Error{409, "DEVICE_UNBOUND", "绑定错误：设备没有绑定人员", "REBIND"}
|
||
|
|
}
|
||
|
|
// Registration belongs to the platform. Local SDK initialization is the
|
||
|
|
// device's responsibility, not a second remote activation handshake.
|
||
|
|
e = tx.QueryRow(ctx, `SELECT b.subject_id::text,b.tenant_id::text,d.binding_generation,j.provider_device_name,COALESCE(a.bot_id,'') FROM device d JOIN device_binding b ON b.device_id=d.id AND b.generation=d.binding_generation AND b.unbound_at IS NULL 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 LEFT JOIN subject_voice_agent a ON a.subject_id=sp.id AND a.status='READY' WHERE d.id=$1::uuid AND d.status='registered' AND d.activation_state NOT IN ('REVOKED','LOST') AND j.sdk_registered_at IS NOT NULL AND j.status IN ('REGISTERED_OBSERVED','READY')`, d.ID).Scan(&subject, &tenant, &generation, &name, &bot)
|
||
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return nil, fault(409, "DEVICE_NOT_READY")
|
||
|
|
}
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if generation != d.BindingGeneration {
|
||
|
|
return nil, fault(409, "DEVICE_BINDING_CHANGED")
|
||
|
|
}
|
||
|
|
if bot == "" {
|
||
|
|
return nil, &Error{409, "VOICE_AGENT_NOT_READY", "该老人的智能体尚未创建成功，请重试创建", "NONE"}
|
||
|
|
}
|
||
|
|
if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,64))`, subject); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
var allowed, memory, storyExtraction 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'),call_memory_authorized($1::uuid)`, subject).Scan(&allowed, &memory)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !allowed {
|
||
|
|
return nil, fault(403, "VOICE_CHAT_CONSENT_REQUIRED")
|
||
|
|
}
|
||
|
|
_, 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
|
||
|
|
}
|
||
|
|
if e = tx.QueryRow(ctx, `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, subject).Scan(&epoch); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
// Read again after acquiring the privacy lock, rather than retaining a
|
||
|
|
// consent result obtained before a concurrent withdrawal committed.
|
||
|
|
if e = tx.QueryRow(ctx, `SELECT NOT EXISTS(SELECT 1 FROM (VALUES ('transcript_storage'),('story_extraction')) required(kind)
|
||
|
|
WHERE NOT EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=$1::uuid AND cr.consent_type=required.kind AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())))`, subject).Scan(&storyExtraction); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
var frozen []byte
|
||
|
|
var state, owner, protocol string
|
||
|
|
var oldGeneration, oldEpoch int64
|
||
|
|
var expiry, lease time.Time
|
||
|
|
e = tx.QueryRow(ctx, `SELECT device_id::text,binding_generation,privacy_epoch,call_state,COALESCE(manifest_b64,''),expires_at,lease_expires_at,COALESCE(budget_snapshot->>'protocol','') FROM call_session WHERE id=$1::uuid FOR UPDATE`, in.SessionID).Scan(&owner, &oldGeneration, &oldEpoch, &state, &frozen, &expiry, &lease, &protocol)
|
||
|
|
if e == nil {
|
||
|
|
if owner != d.ID || protocol != "DIRECT-CHAT-1" {
|
||
|
|
return nil, fault(404, "CHAT_NOT_FOUND")
|
||
|
|
}
|
||
|
|
if oldGeneration != generation || oldEpoch != epoch || len(frozen) == 0 || !oneOf(state, "STARTING", "TALKING") || !expiry.After(s.now()) || !lease.After(s.now()) {
|
||
|
|
return nil, fault(409, "CHAT_ENDED")
|
||
|
|
}
|
||
|
|
var wasMemory bool
|
||
|
|
if e = tx.QueryRow(ctx, `SELECT memory_mode<>'OFF' FROM call_session WHERE id=$1::uuid`, in.SessionID).Scan(&wasMemory); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if wasMemory && !memory {
|
||
|
|
return nil, fault(403, "MEMORY_CONSENT_REVOKED")
|
||
|
|
}
|
||
|
|
} else if errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
var busy bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE (device_id=$1::uuid OR subject_id=$2::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','ACTIVE','STOPPING'))`, d.ID, subject).Scan(&busy)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if busy {
|
||
|
|
return nil, fault(409, "CHAT_ALREADY_ACTIVE")
|
||
|
|
}
|
||
|
|
budget, err := billing.CheckQuotaTx(ctx, tx, d, s.now())
|
||
|
|
if errors.Is(err, billing.ErrQuotaExceeded) {
|
||
|
|
return nil, fault(429, "CALL_QUOTA_EXCEEDED")
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
max := 3600
|
||
|
|
if budget != nil && budget.MaxCallDurationSec < max {
|
||
|
|
max = budget.MaxCallDurationSec
|
||
|
|
}
|
||
|
|
if max < 1 {
|
||
|
|
return nil, fault(429, "CALL_QUOTA_EXCEEDED")
|
||
|
|
}
|
||
|
|
expiry = s.now().Add(time.Duration(max) * time.Second)
|
||
|
|
mode := "OFF"
|
||
|
|
contextText := ""
|
||
|
|
story := storyContext{Version: storyContextVersion}
|
||
|
|
if memory && (s.direct.Prepare != nil || s.direct.PrepareWithWelcome != nil) {
|
||
|
|
mode = "NATIVE_VIKING"
|
||
|
|
}
|
||
|
|
welcome, opening, err := s.selectDirectOpening(ctx, tx, subject, bot, in.SessionID, d.ID, generation, epoch, memory)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if opening != nil && s.direct.PrepareWithWelcome != nil {
|
||
|
|
e = s.direct.PrepareWithWelcome(ctx, subject, bot, in.SessionID, contextText, welcome)
|
||
|
|
if e != nil {
|
||
|
|
return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
|
||
|
|
}
|
||
|
|
} else if s.direct.Prepare != nil {
|
||
|
|
if e = s.direct.Prepare(ctx, subject, bot, in.SessionID, contextText); e != nil {
|
||
|
|
return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
|
||
|
|
}
|
||
|
|
} else if s.direct.PrepareWithWelcome != nil {
|
||
|
|
if e = s.direct.PrepareWithWelcome(ctx, subject, bot, in.SessionID, contextText, welcome); e != nil {
|
||
|
|
return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
frozen, e = s.directParameters(in.SessionID, bot, welcome)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
lease = s.now().Add(45 * time.Second)
|
||
|
|
if lease.After(expiry) {
|
||
|
|
lease = expiry
|
||
|
|
}
|
||
|
|
// The shared legacy table requires a boot_id. Use the session's own UUID
|
||
|
|
// as an internal namespace without inventing a device startup event.
|
||
|
|
snapshot := map[string]any{"protocol": "DIRECT-CHAT-1", "device_name": name, "bot_id": bot, "quota": budget, "story_extraction_allowed": storyExtraction}
|
||
|
|
if memory {
|
||
|
|
snapshot["story_context"] = story
|
||
|
|
}
|
||
|
|
if opening != nil {
|
||
|
|
snapshot["opening"] = opening
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_session(id,device_id,tenant_id,subject_id,boot_id,hook_event_id,call_generation,binding_generation,privacy_epoch,memory_mode,manifest_status,manifest_b64,expires_at,lease_expires_at,call_state,budget_snapshot,text_gap) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$1::uuid,$1,0,$5,$6,$7,'ISSUED',$8,$9,$10,'STARTING',$11,true)`, in.SessionID, d.ID, tenant, subject, generation, epoch, mode, string(frozen), expiry, lease, jsonBytes(snapshot))
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
lease = s.now().Add(45 * time.Second)
|
||
|
|
if lease.After(expiry) {
|
||
|
|
lease = expiry
|
||
|
|
}
|
||
|
|
if s.mixedRecorder != nil {
|
||
|
|
var recordingRevoked bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT COALESCE(budget_snapshot->>'mixed_recording','false')='true' AND NOT EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=call_session.subject_id AND cr.tenant_id=call_session.tenant_id AND cr.consent_type='raw_audio_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())) FROM call_session WHERE id=$1::uuid`, in.SessionID).Scan(&recordingRevoked)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if recordingRevoked {
|
||
|
|
s.stopMixedRecording(ctx, tx, in.SessionID, d.ID)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if _, e = tx.Exec(ctx, `UPDATE call_session SET lease_expires_at=$2,authorization_renewed_at=now() WHERE id=$1::uuid`, in.SessionID, lease); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
var out map[string]any
|
||
|
|
if json.Unmarshal(frozen, &out) != nil {
|
||
|
|
return nil, fault(503, "CHAT_CONFIG_INVALID")
|
||
|
|
}
|
||
|
|
out["session_id"] = in.SessionID
|
||
|
|
out["lease_expires_at_ms"] = lease.UnixMilli()
|
||
|
|
out["expires_at_ms"] = expiry.UnixMilli()
|
||
|
|
return out, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) directParameters(sid, bot, welcome string) ([]byte, error) {
|
||
|
|
u, e := url.Parse(s.direct.CallbackURL)
|
||
|
|
if e != nil || u.Scheme != "https" || u.Host == "" || len(s.direct.CallbackSecret) < 32 || bot == "" {
|
||
|
|
return nil, fault(503, "CHAT_CONFIG_REQUIRED")
|
||
|
|
}
|
||
|
|
// S2S credentials and model settings are configured on the elder's bot once.
|
||
|
|
// Freeze the elder's welcome alongside the per-call callback address.
|
||
|
|
u.Path = "/callbacks/v1/volcengine/chat"
|
||
|
|
q := url.Values{}
|
||
|
|
q.Set("session_id", sid)
|
||
|
|
u.RawQuery = q.Encode()
|
||
|
|
if strings.TrimSpace(welcome) == "" {
|
||
|
|
welcome = "你好，今天想聊些什么？"
|
||
|
|
}
|
||
|
|
agent := map[string]any{"WelcomeMessage": welcome, "EnableConversationStateCallback": false, "ServerMessageURLForRTS": u.String(), "ServerMessageSignatureForRTS": volcsubtitle.Signature(s.direct.CallbackSecret, "direct:"+sid)}
|
||
|
|
params := map[string]any{"AgentConfig": agent, "PlatformTaskID": sid}
|
||
|
|
raw := jsonBytes(params)
|
||
|
|
if len(raw) > 16384 {
|
||
|
|
return nil, fault(503, "CHAT_CONFIG_TOO_LARGE")
|
||
|
|
}
|
||
|
|
return jsonBytes(map[string]any{"bot_id": bot, "params": params}), nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) DirectCallback(w http.ResponseWriter, r *http.Request) {
|
||
|
|
sid := r.URL.Query().Get("session_id")
|
||
|
|
var envelope volcsubtitle.Envelope
|
||
|
|
raw, e := io.ReadAll(http.MaxBytesReader(w, r.Body, 384*1024))
|
||
|
|
if !validUUID(sid) || e != nil || json.Unmarshal(raw, &envelope) != nil {
|
||
|
|
s.write(w, r, nil, fault(400, "INVALID_SUBTITLE"))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
expected := volcsubtitle.Signature(s.direct.CallbackSecret, "direct:"+sid)
|
||
|
|
if len(s.direct.CallbackSecret) < 32 || subtle.ConstantTimeCompare([]byte(envelope.Signature), []byte(expected)) != 1 {
|
||
|
|
s.write(w, r, nil, fault(401, "SUBTITLE_AUTHENTICATION_FAILED"))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
packet, e := volcsubtitle.Decode(envelope.Message)
|
||
|
|
if e != nil {
|
||
|
|
s.write(w, r, nil, fault(400, "INVALID_SUBTITLE"))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
e = s.DirectSubtitle(r.Context(), sid, packet)
|
||
|
|
if e != nil {
|
||
|
|
s.write(w, r, nil, e)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
w.Header().Set("Content-Type", "text/plain")
|
||
|
|
_, _ = w.Write([]byte("ok"))
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) DirectSubtitle(ctx context.Context, sid string, p volcsubtitle.Packet) error {
|
||
|
|
tx, e := s.pool.Begin(ctx)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
defer tx.Rollback(context.Background())
|
||
|
|
var subject, name, bot string
|
||
|
|
var epoch, current int64
|
||
|
|
var expires time.Time
|
||
|
|
var storyRaw []byte
|
||
|
|
var created time.Time
|
||
|
|
var storyAllowed bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT c.subject_id::text,c.budget_snapshot->>'device_name',c.budget_snapshot->>'bot_id',c.privacy_epoch,us.privacy_epoch,c.expires_at,COALESCE(c.budget_snapshot->'story_context','{}'::jsonb),c.created_at,c.memory_mode<>'OFF' AND call_memory_authorized(c.subject_id) FROM call_session c JOIN memory_user_state us ON us.subject_id=c.subject_id WHERE c.id=$1::uuid AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' FOR UPDATE OF c`, sid).Scan(&subject, &name, &bot, &epoch, ¤t, &expires, &storyRaw, &created, &storyAllowed)
|
||
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return fault(404, "CHAT_NOT_FOUND")
|
||
|
|
}
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
allowed, e := transcriptPermission(ctx, tx, subject)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if !allowed || epoch != current {
|
||
|
|
return fault(403, "TRANSCRIPT_STORAGE_REVOKED")
|
||
|
|
}
|
||
|
|
if s.now().After(expires.Add(24 * time.Hour)) {
|
||
|
|
return fault(410, "SUBTITLE_EXPIRED")
|
||
|
|
}
|
||
|
|
if p.DeviceName != "" && p.DeviceName != name {
|
||
|
|
return fault(409, "SUBTITLE_DEVICE_MISMATCH")
|
||
|
|
}
|
||
|
|
changed := false
|
||
|
|
var story storyContext
|
||
|
|
_ = json.Unmarshal(storyRaw, &story)
|
||
|
|
storyAllowed = storyAllowed && story.Version == storyContextVersion
|
||
|
|
for _, part := range p.Data {
|
||
|
|
if !part.Definite || strings.TrimSpace(part.Text) == "" {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
speaker := "UNKNOWN"
|
||
|
|
if part.UserID == "input" || part.UserID == name {
|
||
|
|
speaker = "USER"
|
||
|
|
} else if part.UserID == "output" || part.UserID == bot || isRTCAgentUser(part.UserID) {
|
||
|
|
speaker = "ASSISTANT"
|
||
|
|
}
|
||
|
|
key := hash(jsonBytes([]any{part.UserID, *part.RoundID, *part.Sequence}))
|
||
|
|
if isRTCAgentUser(part.UserID) {
|
||
|
|
// Keep the authenticated, session-specific RTC identity for recording lookup.
|
||
|
|
// A conflicting UID must never reassign a call to a different room.
|
||
|
|
tag, err := tx.Exec(ctx, `UPDATE call_session SET budget_snapshot=jsonb_set(budget_snapshot,'{rtc_agent_user_id}',to_jsonb($2::text)) WHERE id=$1::uuid AND (budget_snapshot->>'rtc_agent_user_id' IS NULL OR budget_snapshot->>'rtc_agent_user_id'=$2)`, sid, part.UserID)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if tag.RowsAffected() != 1 {
|
||
|
|
return fault(409, "RECORDING_IDENTITY_CONFLICT")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
var previous string
|
||
|
|
e = tx.QueryRow(ctx, `SELECT text_content FROM call_utterance WHERE session_id=$1::uuid AND utterance_id=$2`, sid, key).Scan(&previous)
|
||
|
|
if e == nil {
|
||
|
|
if previous != part.Text {
|
||
|
|
return fault(409, "SUBTITLE_CONFLICT")
|
||
|
|
}
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if !errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,$2,0,$3,$4,false,now())`, sid, key, speaker, part.Text)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if speaker == "USER" {
|
||
|
|
if e = s.applySpokenVolume(ctx, tx, sid, part); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
}
|
||
|
|
changed = true
|
||
|
|
if storyAllowed {
|
||
|
|
story.observe(sid, key, speaker, part.Text, s.now().Sub(created).Milliseconds())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if changed {
|
||
|
|
if storyAllowed {
|
||
|
|
if _, e = tx.Exec(ctx, `UPDATE call_session SET budget_snapshot=jsonb_set(budget_snapshot,'{story_context}',$2::jsonb) WHERE id=$1::uuid`, sid, jsonBytes(story)); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET last_text_at=now(),transcript_revision=transcript_revision+1,call_state=CASE WHEN call_state='STARTING' AND ended_at IS NULL THEN 'TALKING' ELSE call_state END,transcript_state=CASE WHEN ended_at IS NULL THEN 'COLLECTING' ELSE 'DRAINING' END WHERE id=$1::uuid`, sid)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if e = s.activateDirect(ctx, tx, sid); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
// The hardware RTC gateway uses bot_<17-digit timestamp> for the room's AI
|
||
|
|
// participant, not the persistent Aibot ID. Verified against real callbacks.
|
||
|
|
func isRTCAgentUser(id string) bool {
|
||
|
|
return len(id) == 21 && strings.HasPrefix(id, "bot_") && strings.Trim(id[4:], "0123456789") == ""
|
||
|
|
}
|
||
|
|
|
||
|
|
// First successful direct chat replaces the old separate conversation-test
|
||
|
|
// handshake. Activation still needs an authenticated device stop AND actual
|
||
|
|
// supplier subtitle evidence; sending a config alone does not activate it.
|
||
|
|
func (s *Service) activateDirect(ctx context.Context, tx pgx.Tx, sid string) error {
|
||
|
|
var device, tenant, job string
|
||
|
|
var generation int64
|
||
|
|
e := tx.QueryRow(ctx, `SELECT d.id::text,d.tenant_id::text,j.id::text,d.binding_generation FROM call_session c JOIN device d ON d.id=c.device_id JOIN device_activation_job j ON j.device_id=d.id AND j.binding_generation=d.binding_generation JOIN device_binding b ON b.device_id=d.id AND b.generation=d.binding_generation AND b.unbound_at IS NULL WHERE c.id=$1::uuid AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND c.binding_generation=d.binding_generation AND c.call_state='ENDED' AND c.sdk_stopped AND d.activation_state='PROVISIONING' AND j.status='REGISTERED_OBSERVED' AND j.sdk_registered_at IS NOT NULL AND EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='ASSISTANT')`, sid).Scan(&device, &tenant, &job, &generation)
|
||
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if _, e = deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: device, TenantID: tenant, Target: deviceactivation.Active, Reason: "直连聊天收到供应商字幕且设备已结束", Source: "onboarding", ActorType: "system", RequestID: sid}); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE device_activation_job SET status='READY',ready_binding_generation=$2,version=version+1,updated_at=now() WHERE id=$1::uuid`, job, generation)
|
||
|
|
return e
|
||
|
|
}
|