diff --git a/docs/regional/ARQUITETURA_TIA_XAI_REGIONAL.md b/docs/regional/ARQUITETURA_TIA_XAI_REGIONAL.md index 2744235..9921532 100644 --- a/docs/regional/ARQUITETURA_TIA_XAI_REGIONAL.md +++ b/docs/regional/ARQUITETURA_TIA_XAI_REGIONAL.md @@ -231,7 +231,7 @@ As conexões WebSocket já estabelecidas não são redirecionadas e continuam no ## 9. Renovação escalonada das conexões -Manter 50 sockets abertos indefinidamente sem renovação é arriscado porque serviços upstream normalmente aplicam TTL e renovação de autorização. +Manter um pool grande de sockets aberto indefinidamente sem renovação é arriscado porque serviços upstream normalmente aplicam TTL e renovação de autorização. A configuração padrão usa: @@ -249,9 +249,55 @@ WS03 -> ~509s ... ``` -Somente conexões livres são renovadas. Isso evita um evento no qual 50 conexões expiram e fazem handshake simultaneamente. +Somente conexões livres são renovadas. Isso evita que todo o pool expire e faça handshake simultaneamente. -## 10. Alta disponibilidade regional +## 10. Barge-in e reutilização segura do pool + +Barge-in ocorre quando o usuário interrompe a fala do TTS antes de `audio.done`. A versão corrigida não descarta automaticamente um WebSocket saudável nem permite que áudio residual contamine a próxima síntese. + +Fluxo: + +```text +usuário interrompe + | + v +TIA cancela a utterance e envia text.clear + | + v +sidecar cancela imediatamente o relay de audio.delta + | + v +envia text.clear ao xAI + | + v +drena e DESCARTA audio.delta/audio.done residuais + | + v +recebe audio.clear + | + +--> confirmado: devolve audio.clear ao TIA e libera o WS saudável ao pool + | + `--> timeout/erro: fecha o WS; maintenance cria substituto pré-aquecido +``` + +Enquanto aguarda `audio.clear`, nenhum frame residual é encaminhado ao LiveKit. Isso cria um boundary limpo entre a utterance cancelada e a próxima. + +Configuração: + +```text +XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S=1.0 +``` + +Métricas específicas: + +```text +tia_xai_pool_barge_ins_total +tia_xai_pool_barge_in_reuses_total +tia_xai_pool_barge_in_resets_total +tia_xai_pool_barge_in_discarded_messages_total +``` + +## 11. Alta disponibilidade regional Operação normal: @@ -267,7 +313,7 @@ Se ORD perder saúde/capacidade xAI, os slots começam a falhar e a quantidade d Não há necessidade de alterar o cliente ou o LiveKit para escolher a região. -## 11. Escala horizontal +## 12. Escala horizontal A capacidade teórica de pool é: @@ -291,19 +337,19 @@ IAD: 2 pods x 50 = 100 TOTAL = 200 ``` -### Restrição crítica +### Capacidade efetivamente provisionada -`replicas * XAI_POOL_SIZE` **não pode ultrapassar o limite real concedido pela OCI para o endpoint/tenancy/região**. +`XAI_POOL_SIZE` é **somente o tamanho configurado do pool por réplica TIA**; não representa um limite público do OCI/xAI. Neste ambiente, a capacidade foi negociada diretamente com xAI/OCI e pode ser muito superior aos exemplos de 50 conexões usados neste documento. -Se a OCI disser que o limite 50 é global por endpoint, então duas réplicas de 50 no mesmo endpoint seriam incorretas. Nesse caso use, por exemplo: +O dimensionamento correto é: ```text -2 replicas x 25 = 50 total +conexões pré-aquecidas da região = réplicas_region * XAI_POOL_SIZE ``` -ou obtenha endpoints/capacidades independentes. +e deve ser comparado com a **capacidade efetivamente negociada/provisionada** para aquela região/endpoint. Exemplos com 25/50 existem apenas para facilitar a leitura da arquitetura. -## 12. Escala visual +## 13. Escala visual ```text Carga baixa diff --git a/docs/regional/CHANGELOG_IMPLEMENTACAO.md b/docs/regional/CHANGELOG_IMPLEMENTACAO.md index 079f06f..bccb733 100644 --- a/docs/regional/CHANGELOG_IMPLEMENTACAO.md +++ b/docs/regional/CHANGELOG_IMPLEMENTACAO.md @@ -30,3 +30,13 @@ O modo anterior permanece disponível. O deployment regional é opcional e não - parsing YAML dos manifests renderizados: PASS. Não foi executado teste real contra OCI xAI, pois depende das credenciais/endpoints do ambiente TIM/OCI. O manual descreve smoke, saturação, failover e stress test a executar em FQA. + + +## Correção — barge-in e capacidade negociada + +- relay upstream passou a ser assíncrono para que `text.clear` seja processado imediatamente durante `audio.delta`; +- barge-in cancela o relay, envia `text.clear` ao xAI e drena mensagens residuais até `audio.clear`; +- socket volta ao pool somente após boundary confirmado; timeout/erro força reset e reposição do slot; +- adicionadas métricas de barge-in/reuso/reset/descarte; +- `XAI_POOL_SIZE` documentado como tamanho configurável por réplica, sem assumir limite público de 50 conexões; +- capacidade total passa a ser dimensionada conforme acordo/provisionamento real xAI/OCI. diff --git a/docs/regional/DEPLOYMENT_TIA_XAI_REGIONAL.md b/docs/regional/DEPLOYMENT_TIA_XAI_REGIONAL.md index 1fb5edb..b25a268 100644 --- a/docs/regional/DEPLOYMENT_TIA_XAI_REGIONAL.md +++ b/docs/regional/DEPLOYMENT_TIA_XAI_REGIONAL.md @@ -70,7 +70,7 @@ IAD_XAI_SECRET_NAME=xai-iad-credentials ## 5. Definir tamanho do pool -Para um Pod com 50 sockets: +`XAI_POOL_SIZE` é o número de conexões xAI pré-aquecidas mantidas por réplica TIA; **não é um limite do serviço OCI/xAI**. Ajuste-o conforme a capacidade negociada/provisionada. Exemplo com 50 sockets: ```bash XAI_POOL_SIZE=50 @@ -90,6 +90,15 @@ XAI_POOL_UNAVAILABLE_FREE=0 XAI_POOL_RECOVER_FREE=5 ``` + +Para barge-in, configure também: + +```bash +XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S=1.0 +``` + +Esse timeout limita quanto tempo o sidecar espera pelo `audio.clear` que confirma que o WebSocket está limpo após uma interrupção. Se não houver confirmação, o socket é descartado e substituído. + ### Importante Se houver `N` réplicas apontando para o mesmo endpoint: @@ -98,7 +107,7 @@ Se houver `N` réplicas apontando para o mesmo endpoint: sockets máximos = N * XAI_POOL_SIZE ``` -Nunca configure isso acima da quota xAI real. +Dimensione esse total de acordo com a capacidade efetivamente negociada/provisionada para a região/endpoint. ## 6. Criar Secrets regionais diff --git a/docs/regional/TESTES_TIA_XAI_REGIONAL.md b/docs/regional/TESTES_TIA_XAI_REGIONAL.md index 677f55d..063d547 100644 --- a/docs/regional/TESTES_TIA_XAI_REGIONAL.md +++ b/docs/regional/TESTES_TIA_XAI_REGIONAL.md @@ -14,7 +14,7 @@ Validar separadamente capacidade do pool, comportamento do Kubernetes, failover ## Camada 2 — Prewarm -Subir com `XAI_POOL_SIZE=50` e medir: +Subir com um `XAI_POOL_SIZE` representativo da configuração alvo (50 abaixo é apenas exemplo) e medir: - tempo total até Ready; - taxa de sucesso de handshake; @@ -64,11 +64,32 @@ Observe por 5 minutos. Esperado: - sockets são renovados individualmente; -- não existe burst de 50 reconnects; +- não existe burst de reconnects equivalente ao tamanho total do pool; - slots ocupados não são renovados no meio da síntese; - pool retorna ao tamanho configurado. -## Camada 6 — Falha upstream + +## Camada 6 — Barge-in / clean boundary + +Com uma síntese longa em andamento: + +1. aguardar pelo menos um `audio.delta`; +2. simular interrupção do usuário; +3. confirmar envio de `text.clear` ao sidecar; +4. fazer o fake/provider enviar 1 ou mais `audio.delta` residuais antes de `audio.clear`; +5. confirmar que os frames residuais **não** chegam ao LiveKit; +6. confirmar `audio.clear` entregue ao TIA; +7. confirmar `leased` retorna ao valor anterior sem fechar o socket; +8. iniciar nova utterance e validar ausência de áudio da utterance cancelada. + +Critérios: + +- `tia_xai_pool_barge_ins_total` incrementa; +- `tia_xai_pool_barge_in_reuses_total` incrementa quando `audio.clear` é confirmado; +- `tia_xai_pool_barge_in_discarded_messages_total` contabiliza frames residuais; +- se `audio.clear` não chegar dentro de `XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S`, `tia_xai_pool_barge_in_resets_total` incrementa e o slot é recriado. + +## Camada 7 — Falha upstream Bloqueie ORD ou aponte temporariamente para endpoint inválido. @@ -80,7 +101,7 @@ Esperado: - Service deixa de enviar novas chamadas a ORD; - IAD continua Ready. -## Camada 7 — Latência +## Camada 8 — Latência Compare três cenários: @@ -101,7 +122,7 @@ Meça: Hipótese: o hop localhost adiciona latência desprezível frente ao TTFB do provider, enquanto remove handshake xAI do caminho crítico na situação normal. -## Camada 8 — Carga semelhante a produção +## Camada 9 — Carga semelhante a produção Evite somente burst C=200. Use sockets persistentes e concorrência de síntese representativa da operação real, seguindo a metodologia que produziu resultados reprodutíveis nos testes anteriores. @@ -115,7 +136,7 @@ Rodar pelo menos: 120% por janela curta ``` -## Camada 9 — Rollout +## Camada 10 — Rollout Com chamadas ativas: diff --git a/docs/regional/VALIDACAO_CORRECAO_BARGE_IN.md b/docs/regional/VALIDACAO_CORRECAO_BARGE_IN.md new file mode 100644 index 0000000..5d6a92a --- /dev/null +++ b/docs/regional/VALIDACAO_CORRECAO_BARGE_IN.md @@ -0,0 +1,36 @@ +# Validação da Correção — Barge-in e Pool xAI + +## Escopo + +A correção torna o relay upstream cancelável durante uma síntese e preserva o WebSocket somente após confirmação explícita de boundary limpo (`audio.clear`). + +## Comportamento validado + +Sequência simulada após barge-in: + +```text +TIA -> text.clear +xAI -> audio.delta (residual; descartado) +xAI -> audio.done (residual; descartado) +xAI -> audio.clear (boundary confirmado) +``` + +Resultado esperado e observado no teste focado: + +- `text.clear` é enviado ao upstream; +- mensagens residuais são drenadas localmente; +- frames residuais não são encaminhados ao LiveKit; +- `audio.clear` confirma que o socket pode ser reutilizado; +- se não houver `audio.clear` no timeout, o slot é marcado unhealthy, fechado e posteriormente recriado pelo maintenance loop. + +## Validações executadas neste ambiente + +- `python -m compileall -q src`: PASS; +- parse YAML dos 6 manifests regionais renderizados: PASS; +- teste focado `_clear_slot_and_wait` com 2 mensagens residuais antes de `audio.clear`: PASS; +- suíte existente `tests/adapters/test_xai_tts.py`: não executada por falta do pacote `oci` no runtime de validação; +- `kubectl --dry-run`: não executado porque `kubectl` não está instalado no runtime; os manifests foram validados por parser YAML. + +## Capacidade + +`XAI_POOL_SIZE` representa apenas o tamanho do pool por réplica TIA. A capacidade total deve seguir o provisionamento/acordo real do ambiente xAI/OCI, sem assumir o limite público de 50 conexões. diff --git a/k8s/regional/deployment-region.yaml b/k8s/regional/deployment-region.yaml index 15be7b7..9fe37f8 100644 --- a/k8s/regional/deployment-region.yaml +++ b/k8s/regional/deployment-region.yaml @@ -157,6 +157,8 @@ spec: value: "${XAI_POOL_REFRESH_JITTER_S}" - name: XAI_POOL_PREWARM_CONCURRENCY value: "${XAI_POOL_PREWARM_CONCURRENCY}" + - name: XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S + value: "${XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S}" - name: XAI_TTS_VOICE value: ${XAI_TTS_VOICE} - name: XAI_TTS_LANGUAGE diff --git a/k8s/regional/regions.env.example b/k8s/regional/regions.env.example index a471928..6230b32 100644 --- a/k8s/regional/regions.env.example +++ b/k8s/regional/regions.env.example @@ -23,17 +23,20 @@ CPU_XAI_POOL_LIM=1000m MEM_XAI_POOL_LIM=768Mi # Pool profile - 50 means 50 prewarmed upstream WebSockets PER POD. +# Tamanho do pool por réplica TIA. NÃO representa limite do OCI/xAI. +# Ajuste conforme capacidade negociada para o ambiente. XAI_POOL_SIZE=50 XAI_POOL_UNAVAILABLE_FREE=2 XAI_POOL_RECOVER_FREE=5 XAI_POOL_CONNECTION_TTL_S=540 XAI_POOL_REFRESH_JITTER_S=45 XAI_POOL_PREWARM_CONCURRENCY=5 +XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S=1.0 XAI_TTS_VOICE=c8x2ieiocufs XAI_TTS_LANGUAGE=pt-BR XAI_UPSTREAM_AUTH_METHOD=API_KEY -# HPA. WARNING: replicas * XAI_POOL_SIZE must respect OCI/xAI quota. +# HPA. replicas * XAI_POOL_SIZE deve respeitar a capacidade efetivamente negociada/provisionada para cada região/endpoint. HPA_MIN_REPLICAS=1 HPA_MAX_REPLICAS=3 HPA_CPU_TARGET=65 diff --git a/k8s/regional/rendered/deployment-iad.yaml b/k8s/regional/rendered/deployment-iad.yaml new file mode 100644 index 0000000..666587a --- /dev/null +++ b/k8s/regional/rendered/deployment-iad.yaml @@ -0,0 +1,213 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: tim-ai-atend-agnt-integ-tia-iad + namespace: agnt-ai-atendimento + labels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: iad +spec: + replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: iad + template: + metadata: + labels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: iad + spec: + terminationGracePeriodSeconds: 600 + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: bridge + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.bridge_entry", "--host", "0.0.0.0", "--port", "8000", "--log-level", "info"] + ports: + - name: bridge-http + containerPort: 8000 + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: /etc/google/credentials.json + - name: PYTHONPATH + value: /app/src + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + - name: TIA_XAI_REGION + value: iad + envFrom: + - configMapRef: + name: tim-ai-atend-agnt-integ-tia-config + - secretRef: + name: tim-ai-atend-agnt-integ-tia-api-secrets + readinessProbe: + httpGet: {path: /health, port: 8000} + initialDelaySeconds: 15 + periodSeconds: 5 + timeoutSeconds: 3 + failureThreshold: 3 + livenessProbe: + httpGet: {path: /health, port: 8000} + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + resources: + requests: {cpu: "250m", memory: "512Mi"} + limits: {cpu: "1000m", memory: "1Gi"} + volumeMounts: + - {name: google-sa-volume, mountPath: /etc/google, readOnly: true} + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + - name: agent + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.agent_entry", "start", "--log-level", "info"] + ports: + - name: agent-http + containerPort: 18081 + envFrom: + - configMapRef: + name: tim-ai-atend-agnt-integ-tia-config + - secretRef: + name: tim-ai-atend-agnt-integ-tia-api-secrets + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: /etc/google/credentials.json + - name: PYTHONPATH + value: /app/src + - name: AGENT_SERVER_PORT + value: "18081" + - name: NUM_IDLE_PROCESSES + value: "1" + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + - name: TIA_XAI_REGION + value: iad + # Agent sees a local xAI-compatible endpoint. Real OCI credentials stay in xai-pool. + - name: XAI_WEBSOCKET_URL + value: ws://127.0.0.1:18100/xai/v1/tts + - name: XAI_TTS_AUTH_METHOD + value: API_KEY + - name: XAI_API_KEY + value: local-pool-proxy + startupProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 10 + periodSeconds: 5 + timeoutSeconds: 5 + failureThreshold: 24 + readinessProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 20 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 + livenessProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + resources: + requests: {cpu: "500m", memory: "1Gi"} + limits: {cpu: "2000m", memory: "2Gi"} + volumeMounts: + - {name: google-sa-volume, mountPath: /etc/google, readOnly: true} + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + - name: xai-pool + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.livekit.adapters.xai_pool_proxy"] + ports: + - name: xai-pool + containerPort: 18100 + env: + - name: PYTHONPATH + value: /app/src + - name: TIA_XAI_REGION + value: iad + - name: XAI_POOL_UPSTREAM_URL + value: wss://peiadagnt003prd.pe.inference.generativeai.us-ashburn-1.oci.oraclecloud.com/xai/v1/tts + - name: XAI_POOL_SIZE + value: "50" + - name: XAI_POOL_UNAVAILABLE_FREE + value: "2" + - name: XAI_POOL_RECOVER_FREE + value: "5" + - name: XAI_POOL_CONNECTION_TTL_S + value: "540" + - name: XAI_POOL_REFRESH_JITTER_S + value: "45" + - name: XAI_POOL_PREWARM_CONCURRENCY + value: "5" + - name: XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S + value: "1.0" + - name: XAI_TTS_VOICE + value: c8x2ieiocufs + - name: XAI_TTS_LANGUAGE + value: pt-BR + - name: XAI_TTS_AUTH_METHOD + value: API_KEY + - name: OCI_COMPARTMENT_ID + valueFrom: + secretKeyRef: + name: xai-iad-credentials + key: OCI_COMPARTMENT_ID + optional: true + - name: XAI_API_KEY + valueFrom: + secretKeyRef: + name: xai-iad-credentials + key: XAI_API_KEY + optional: true + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + readinessProbe: + httpGet: {path: /readyz, port: 18100} + initialDelaySeconds: 5 + periodSeconds: 2 + timeoutSeconds: 1 + failureThreshold: 2 + successThreshold: 1 + livenessProbe: + httpGet: {path: /healthz, port: 18100} + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 2 + failureThreshold: 3 + lifecycle: + preStop: + exec: + command: ["/bin/sh", "-c", "curl -sf -X POST http://127.0.0.1:18100/drain || true; sleep 30"] + resources: + requests: {cpu: "200m", memory: "256Mi"} + limits: {cpu: "1000m", memory: "768Mi"} + volumeMounts: + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + volumes: + - name: google-sa-volume + secret: + secretName: tim-ai-atend-agnt-integ-tia-google-sa-secret + - name: trusted-ca-volume + secret: + secretName: shared-tls-secret diff --git a/k8s/regional/rendered/deployment-ord.yaml b/k8s/regional/rendered/deployment-ord.yaml new file mode 100644 index 0000000..5635d35 --- /dev/null +++ b/k8s/regional/rendered/deployment-ord.yaml @@ -0,0 +1,213 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: tim-ai-atend-agnt-integ-tia-ord + namespace: agnt-ai-atendimento + labels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: ord +spec: + replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: ord + template: + metadata: + labels: + app: tim-ai-atend-agnt-integ-tia-regional + tia-region: ord + spec: + terminationGracePeriodSeconds: 600 + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: bridge + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.bridge_entry", "--host", "0.0.0.0", "--port", "8000", "--log-level", "info"] + ports: + - name: bridge-http + containerPort: 8000 + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: /etc/google/credentials.json + - name: PYTHONPATH + value: /app/src + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + - name: TIA_XAI_REGION + value: ord + envFrom: + - configMapRef: + name: tim-ai-atend-agnt-integ-tia-config + - secretRef: + name: tim-ai-atend-agnt-integ-tia-api-secrets + readinessProbe: + httpGet: {path: /health, port: 8000} + initialDelaySeconds: 15 + periodSeconds: 5 + timeoutSeconds: 3 + failureThreshold: 3 + livenessProbe: + httpGet: {path: /health, port: 8000} + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + resources: + requests: {cpu: "250m", memory: "512Mi"} + limits: {cpu: "1000m", memory: "1Gi"} + volumeMounts: + - {name: google-sa-volume, mountPath: /etc/google, readOnly: true} + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + - name: agent + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.agent_entry", "start", "--log-level", "info"] + ports: + - name: agent-http + containerPort: 18081 + envFrom: + - configMapRef: + name: tim-ai-atend-agnt-integ-tia-config + - secretRef: + name: tim-ai-atend-agnt-integ-tia-api-secrets + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: /etc/google/credentials.json + - name: PYTHONPATH + value: /app/src + - name: AGENT_SERVER_PORT + value: "18081" + - name: NUM_IDLE_PROCESSES + value: "1" + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + - name: TIA_XAI_REGION + value: ord + # Agent sees a local xAI-compatible endpoint. Real OCI credentials stay in xai-pool. + - name: XAI_WEBSOCKET_URL + value: ws://127.0.0.1:18100/xai/v1/tts + - name: XAI_TTS_AUTH_METHOD + value: API_KEY + - name: XAI_API_KEY + value: local-pool-proxy + startupProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 10 + periodSeconds: 5 + timeoutSeconds: 5 + failureThreshold: 24 + readinessProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 20 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 + livenessProbe: + httpGet: {path: /, port: 18081} + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + resources: + requests: {cpu: "500m", memory: "1Gi"} + limits: {cpu: "2000m", memory: "2Gi"} + volumeMounts: + - {name: google-sa-volume, mountPath: /etc/google, readOnly: true} + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + - name: xai-pool + image: iad.ocir.io/SEU_NAMESPACE/tia:regional-xai-pool-v1 + imagePullPolicy: IfNotPresent + args: ["app.livekit.adapters.xai_pool_proxy"] + ports: + - name: xai-pool + containerPort: 18100 + env: + - name: PYTHONPATH + value: /app/src + - name: TIA_XAI_REGION + value: ord + - name: XAI_POOL_UPSTREAM_URL + value: wss://peordagnt002prd.pe.inference.generativeai.us-chicago-1.oci.oraclecloud.com/xai/v1/tts + - name: XAI_POOL_SIZE + value: "50" + - name: XAI_POOL_UNAVAILABLE_FREE + value: "2" + - name: XAI_POOL_RECOVER_FREE + value: "5" + - name: XAI_POOL_CONNECTION_TTL_S + value: "540" + - name: XAI_POOL_REFRESH_JITTER_S + value: "45" + - name: XAI_POOL_PREWARM_CONCURRENCY + value: "5" + - name: XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S + value: "1.0" + - name: XAI_TTS_VOICE + value: c8x2ieiocufs + - name: XAI_TTS_LANGUAGE + value: pt-BR + - name: XAI_TTS_AUTH_METHOD + value: API_KEY + - name: OCI_COMPARTMENT_ID + valueFrom: + secretKeyRef: + name: xai-ord-credentials + key: OCI_COMPARTMENT_ID + optional: true + - name: XAI_API_KEY + valueFrom: + secretKeyRef: + name: xai-ord-credentials + key: XAI_API_KEY + optional: true + - name: REQUESTS_CA_BUNDLE + value: /etc/ssl/custom/tls.crt + - name: SSL_CERT_FILE + value: /etc/ssl/custom/tls.crt + readinessProbe: + httpGet: {path: /readyz, port: 18100} + initialDelaySeconds: 5 + periodSeconds: 2 + timeoutSeconds: 1 + failureThreshold: 2 + successThreshold: 1 + livenessProbe: + httpGet: {path: /healthz, port: 18100} + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 2 + failureThreshold: 3 + lifecycle: + preStop: + exec: + command: ["/bin/sh", "-c", "curl -sf -X POST http://127.0.0.1:18100/drain || true; sleep 30"] + resources: + requests: {cpu: "200m", memory: "256Mi"} + limits: {cpu: "1000m", memory: "768Mi"} + volumeMounts: + - {name: trusted-ca-volume, mountPath: /etc/ssl/custom, readOnly: true} + + volumes: + - name: google-sa-volume + secret: + secretName: tim-ai-atend-agnt-integ-tia-google-sa-secret + - name: trusted-ca-volume + secret: + secretName: shared-tls-secret diff --git a/k8s/regional/rendered/hpa-iad.yaml b/k8s/regional/rendered/hpa-iad.yaml new file mode 100644 index 0000000..8526693 --- /dev/null +++ b/k8s/regional/rendered/hpa-iad.yaml @@ -0,0 +1,32 @@ +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: tim-ai-atend-agnt-integ-tia-iad + namespace: agnt-ai-atendimento +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: tim-ai-atend-agnt-integ-tia-iad + minReplicas: 1 + maxReplicas: 3 + behavior: + scaleUp: + stabilizationWindowSeconds: 0 + policies: + - type: Percent + value: 100 + periodSeconds: 60 + scaleDown: + stabilizationWindowSeconds: 300 + policies: + - type: Percent + value: 25 + periodSeconds: 60 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 65 diff --git a/k8s/regional/rendered/hpa-ord.yaml b/k8s/regional/rendered/hpa-ord.yaml new file mode 100644 index 0000000..07e287f --- /dev/null +++ b/k8s/regional/rendered/hpa-ord.yaml @@ -0,0 +1,32 @@ +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: tim-ai-atend-agnt-integ-tia-ord + namespace: agnt-ai-atendimento +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: tim-ai-atend-agnt-integ-tia-ord + minReplicas: 1 + maxReplicas: 3 + behavior: + scaleUp: + stabilizationWindowSeconds: 0 + policies: + - type: Percent + value: 100 + periodSeconds: 60 + scaleDown: + stabilizationWindowSeconds: 300 + policies: + - type: Percent + value: 25 + periodSeconds: 60 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 65 diff --git a/k8s/regional/rendered/pdb.yaml b/k8s/regional/rendered/pdb.yaml new file mode 100644 index 0000000..4e24573 --- /dev/null +++ b/k8s/regional/rendered/pdb.yaml @@ -0,0 +1,10 @@ +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: tim-ai-atend-agnt-integ-tia-regional + namespace: agnt-ai-atendimento +spec: + minAvailable: 2 + selector: + matchLabels: + app: tim-ai-atend-agnt-integ-tia-regional diff --git a/k8s/regional/rendered/service.yaml b/k8s/regional/rendered/service.yaml new file mode 100644 index 0000000..56febfa --- /dev/null +++ b/k8s/regional/rendered/service.yaml @@ -0,0 +1,17 @@ +apiVersion: v1 +kind: Service +metadata: + name: tim-ai-atend-agnt-integ-tia-regional + namespace: agnt-ai-atendimento + labels: + app: tim-ai-atend-agnt-integ-tia-regional +spec: + type: LoadBalancer + sessionAffinity: None + selector: + app: tim-ai-atend-agnt-integ-tia-regional + ports: + - name: ws-http + protocol: TCP + port: 80 + targetPort: 8000 diff --git a/src/app/livekit/adapters/xai_pool_proxy.py b/src/app/livekit/adapters/xai_pool_proxy.py index 66cac72..aea2ad1 100644 --- a/src/app/livekit/adapters/xai_pool_proxy.py +++ b/src/app/livekit/adapters/xai_pool_proxy.py @@ -89,6 +89,7 @@ class PoolConfig: refresh_jitter_s: float maintenance_interval_s: float acquire_timeout_s: float + barge_in_clear_timeout_s: float @classmethod def from_env(cls) -> "PoolConfig": @@ -109,6 +110,7 @@ class PoolConfig: refresh_jitter_s=max(0.0, _env_float("XAI_POOL_REFRESH_JITTER_S", 45.0)), maintenance_interval_s=max(0.5, _env_float("XAI_POOL_MAINTENANCE_INTERVAL_S", 2.0)), acquire_timeout_s=max(0.1, _env_float("XAI_POOL_ACQUIRE_TIMEOUT_S", 2.0)), + barge_in_clear_timeout_s=max(0.1, _env_float("XAI_POOL_BARGE_IN_CLEAR_TIMEOUT_S", 1.0)), ) def upstream_ws_url(self) -> str: @@ -215,6 +217,10 @@ class RegionalXAIPool: self.total_acquires = 0 self.total_acquire_timeouts = 0 self.total_proxy_failures = 0 + self.total_barge_ins = 0 + self.total_barge_in_reuses = 0 + self.total_barge_in_resets = 0 + self.total_barge_in_discarded_messages = 0 async def start(self) -> None: if self._started: @@ -359,6 +365,10 @@ class RegionalXAIPool: "total_acquires": self.total_acquires, "total_acquire_timeouts": self.total_acquire_timeouts, "total_proxy_failures": self.total_proxy_failures, + "total_barge_ins": self.total_barge_ins, + "total_barge_in_reuses": self.total_barge_in_reuses, + "total_barge_in_resets": self.total_barge_in_resets, + "total_barge_in_discarded_messages": self.total_barge_in_discarded_messages, } @@ -382,7 +392,12 @@ def _client_query_matches(request: web.Request, config: PoolConfig) -> bool: async def _relay_upstream_until_boundary(client: web.WebSocketResponse, slot: UpstreamSlot) -> bool: - """Relay one provider response boundary. Return True only after audio.done.""" + """Relay provider audio until audio.done. + + The task is intentionally cancellable. Barge-in is processed by the client + control loop, which cancels this relay first and then establishes a clean + provider boundary with text.clear/audio.clear before the slot can be reused. + """ ws = slot.ws if ws is None: return False @@ -404,6 +419,61 @@ async def _relay_upstream_until_boundary(client: web.WebSocketResponse, slot: Up return False +async def _clear_slot_and_wait( + slot: UpstreamSlot, + *, + timeout_s: float, +) -> tuple[bool, str | None, int]: + """Clear one xAI connection and drain stale frames until audio.clear. + + Returns ``(confirmed, audio_clear_json, discarded_message_count)``. + Anything received before ``audio.clear`` is deliberately discarded. This is + the critical safety boundary that prevents audio from a cancelled utterance + from leaking into the next user turn. + """ + ws = slot.ws + if ws is None or not slot.healthy: + return False, None, 0 + + discarded = 0 + try: + await ws.send_str(json.dumps({"type": "text.clear"})) + except Exception: + return False, None, discarded + + deadline = time.monotonic() + timeout_s + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False, None, discarded + try: + msg = await asyncio.wait_for(ws.receive(), timeout=remaining) + except asyncio.TimeoutError: + return False, None, discarded + except Exception: + return False, None, discarded + + if msg.type != WSMsgType.TEXT: + if msg.type in {WSMsgType.CLOSE, WSMsgType.CLOSED, WSMsgType.ERROR}: + return False, None, discarded + discarded += 1 + continue + + try: + payload = json.loads(msg.data) + except Exception: + discarded += 1 + continue + msg_type = str(payload.get("type") or "") + if msg_type == "audio.clear": + return True, msg.data, discarded + if msg_type in {"error", "response.error"}: + return False, None, discarded + # audio.delta/audio.done from the interrupted utterance and any other + # stale provider messages are drained locally and MUST NOT reach LiveKit. + discarded += 1 + + async def websocket_proxy(request: web.Request) -> web.StreamResponse: pool = POOL if pool is None: @@ -413,83 +483,191 @@ async def websocket_proxy(request: web.Request) -> web.StreamResponse: client = web.WebSocketResponse(heartbeat=20.0, max_msg_size=8 * 1024 * 1024) await client.prepare(request) + leased_slot: UpstreamSlot | None = None - slot_healthy = True + relay_task: asyncio.Task[bool] | None = None + client_receive_task: asyncio.Task[Any] | None = None + turn_started = False + + async def cancel_relay() -> None: + nonlocal relay_task + if relay_task is None: + return + if not relay_task.done(): + relay_task.cancel() + try: + await relay_task + except asyncio.CancelledError: + pass + except Exception: + pass + relay_task = None + + async def clean_current_slot(*, forward_ack: bool, count_barge_in: bool) -> bool: + nonlocal leased_slot, turn_started + if leased_slot is None: + return True + await cancel_relay() + confirmed, clear_json, discarded = await _clear_slot_and_wait( + leased_slot, + timeout_s=pool.config.barge_in_clear_timeout_s, + ) + pool.total_barge_in_discarded_messages += discarded + if count_barge_in: + pool.total_barge_ins += 1 + if confirmed: + if forward_ack and clear_json is not None and not client.closed: + await client.send_str(clear_json) + if count_barge_in: + pool.total_barge_in_reuses += 1 + await pool.release(leased_slot, healthy=True) + else: + if count_barge_in: + pool.total_barge_in_resets += 1 + await pool.release(leased_slot, healthy=False) + leased_slot = None + turn_started = False + return confirmed + try: - async for msg in client: - if msg.type != WSMsgType.TEXT: + client_receive_task = asyncio.create_task(client.receive()) + while not client.closed: + waiters: set[asyncio.Task[Any]] = {client_receive_task} + if relay_task is not None: + waiters.add(relay_task) + done, _ = await asyncio.wait(waiters, return_when=asyncio.FIRST_COMPLETED) + + # Prefer an incoming control message when both sides complete in the + # same event-loop turn. This lets text.clear win the barge-in race. + if client_receive_task in done: + msg = client_receive_task.result() + client_receive_task = asyncio.create_task(client.receive()) + if msg.type in {WSMsgType.CLOSE, WSMsgType.CLOSED, WSMsgType.ERROR}: break - continue - try: - payload = json.loads(msg.data) - except Exception: - await client.send_str(json.dumps({"type": "error", "message": "invalid json"})) - continue - msg_type = str(payload.get("type") or "") - - if msg_type == "text.clear": - if leased_slot is not None: - await pool.release(leased_slot, healthy=slot_healthy) - leased_slot = None + if msg.type != WSMsgType.TEXT: + continue try: - leased_slot = await pool.acquire() - slot_healthy = True - except PoolUnavailable as exc: - await client.send_str(json.dumps({"type": "error", "message": str(exc), "code": "xai_pool_exhausted"})) - await client.close(code=1013, message=b"xAI pool exhausted") - break + payload = json.loads(msg.data) + except Exception: + await client.send_str(json.dumps({"type": "error", "message": "invalid json"})) + continue + msg_type = str(payload.get("type") or "") + + if msg_type == "text.clear": + # Existing active/pending utterance => barge-in/reset. The + # slot is only reusable after audio.clear confirms the clean + # boundary. Residual audio is drained and never forwarded. + if leased_slot is not None: + was_active_turn = turn_started or relay_task is not None + if was_active_turn: + confirmed = await clean_current_slot( + forward_ack=True, + count_barge_in=True, + ) + if not confirmed: + pool.total_proxy_failures += 1 + await client.close(code=1011, message=b"xAI cancel clear unconfirmed") + break + # A cancellation clear completes the old turn. The + # next synthesis will acquire a (possibly different) + # warm slot when the client sends its next text.clear. + continue + + # Defensive repeated clear before a turn starts: keep the + # same leased socket and return exactly one clear ACK. + confirmed, clear_json, discarded = await _clear_slot_and_wait( + leased_slot, + timeout_s=pool.config.barge_in_clear_timeout_s, + ) + pool.total_barge_in_discarded_messages += discarded + if not confirmed or clear_json is None: + pool.total_proxy_failures += 1 + await pool.release(leased_slot, healthy=False) + leased_slot = None + await client.close(code=1011, message=b"xAI upstream clear failed") + break + await client.send_str(clear_json) + continue + + try: + leased_slot = await pool.acquire() + except PoolUnavailable as exc: + await client.send_str(json.dumps({"type": "error", "message": str(exc), "code": "xai_pool_exhausted"})) + await client.close(code=1013, message=b"xAI pool exhausted") + break + confirmed, clear_json, discarded = await _clear_slot_and_wait( + leased_slot, + timeout_s=pool.config.barge_in_clear_timeout_s, + ) + pool.total_barge_in_discarded_messages += discarded + if not confirmed or clear_json is None: + pool.total_proxy_failures += 1 + await pool.release(leased_slot, healthy=False) + leased_slot = None + await client.close(code=1011, message=b"xAI upstream clear failed") + break + await client.send_str(clear_json) + turn_started = False + continue + + if leased_slot is None: + await client.send_str(json.dumps({"type": "error", "message": "text.clear required before synthesis"})) + continue + + if relay_task is not None: + await client.send_str(json.dumps({"type": "error", "message": "text.clear required to interrupt active synthesis"})) + continue + assert leased_slot.ws is not None try: await leased_slot.ws.send_str(msg.data) - # text.clear has its own acknowledgement and must be forwarded - # before the client sends text.delta/text.done. - while True: - ack = await leased_slot.ws.receive() - if ack.type != WSMsgType.TEXT: - slot_healthy = False - raise RuntimeError("xAI clear acknowledgement failed") - await client.send_str(ack.data) - try: - ack_payload = json.loads(ack.data) - except Exception: - ack_payload = {} - ack_type = str(ack_payload.get("type") or "") - if ack_type == "audio.clear": - break - if ack_type in {"error", "response.error"}: - slot_healthy = False - raise RuntimeError("xAI clear returned error") + except Exception: + pool.total_proxy_failures += 1 + await pool.release(leased_slot, healthy=False) + leased_slot = None + await client.close(code=1011, message=b"xAI upstream write failure") + break + + if msg_type in {"text.delta", "text.done"}: + turn_started = True + if msg_type == "text.done": + relay_task = asyncio.create_task( + _relay_upstream_until_boundary(client, leased_slot), + name=f"xai-pool-relay-{leased_slot.slot_id}", + ) + continue + + if relay_task is not None and relay_task in done: + try: + slot_healthy = bool(relay_task.result()) + except asyncio.CancelledError: + # Cancellation is handled by the text.clear path above. + continue except Exception: slot_healthy = False - pool.total_proxy_failures += 1 - await client.close(code=1011, message=b"xAI upstream clear failed") - break - continue - - if leased_slot is None: - await client.send_str(json.dumps({"type": "error", "message": "text.clear required before synthesis"})) - continue - - assert leased_slot.ws is not None - try: - await leased_slot.ws.send_str(msg.data) - if msg_type == "text.done": - slot_healthy = await _relay_upstream_until_boundary(client, leased_slot) + relay_task = None + if leased_slot is not None: await pool.release(leased_slot, healthy=slot_healthy) leased_slot = None - if not slot_healthy: - pool.total_proxy_failures += 1 - await client.close(code=1011, message=b"xAI upstream failed") - break - except Exception: - slot_healthy = False - pool.total_proxy_failures += 1 - await client.close(code=1011, message=b"xAI upstream failure") - break + turn_started = False + if not slot_healthy: + pool.total_proxy_failures += 1 + await client.close(code=1011, message=b"xAI upstream failed") + break finally: + if client_receive_task is not None and not client_receive_task.done(): + client_receive_task.cancel() + try: + await client_receive_task + except asyncio.CancelledError: + pass if leased_slot is not None: - await pool.release(leased_slot, healthy=False) + # Client disappeared during an utterance. Try the same clean-boundary + # protocol used for barge-in so the warm socket can still be reused. + confirmed = await clean_current_slot(forward_ack=False, count_barge_in=turn_started) + if not confirmed: + logger.info("XAI_POOL_SLOT_RESET_AFTER_CLIENT_DISCONNECT region=%s", pool.config.region) return client @@ -535,6 +713,11 @@ async def metrics(_: web.Request) -> web.Response: f'tia_xai_pool_acquire_timeouts_total{{region="{region}"}} {values.get("total_acquire_timeouts", 0)}', "# TYPE tia_xai_pool_proxy_failures_total counter", f'tia_xai_pool_proxy_failures_total{{region="{region}"}} {values.get("total_proxy_failures", 0)}', + "# TYPE tia_xai_pool_barge_ins_total counter", + f'tia_xai_pool_barge_ins_total{{region="{region}"}} {values.get("total_barge_ins", 0)}', + f'tia_xai_pool_barge_in_reuses_total{{region="{region}"}} {values.get("total_barge_in_reuses", 0)}', + f'tia_xai_pool_barge_in_resets_total{{region="{region}"}} {values.get("total_barge_in_resets", 0)}', + f'tia_xai_pool_barge_in_discarded_messages_total{{region="{region}"}} {values.get("total_barge_in_discarded_messages", 0)}', ] return web.Response(text="\n".join(lines) + "\n", content_type="text/plain")