feat(servidor): adiciona SSE, token, E-Stop real e trava de hardware

- comum.py: ler_json (Origin/JSON/tamanho), gravar_atomico e token
  (CARINHA_TOKEN ou seguranca.json), substituindo 6 copias nos modulos
- eventos.py: hub SSE em /eventos com heartbeat e Last-Event-ID;
  modulos publicam expressao, telemetria, conversa, reacao, ia, motores
- /estado agrega o snapshot inicial; /comando unifica gesto/rosto/cor/reacao
- /parar-tudo desliga o PCA9685, para reacoes e trava motores ate /rearmar
- braco: lease (ttl_ms) e interpolacao com limite de velocidade no servidor
- braco e diagnostico nao disputam o PCA9685 (409 para o segundo dono)
- fila limitada (20) com validade de 10 s; POST /face usa prefixo "rosto"
- /app/config passa a checar Origin e JSON; versao vem de version.json
- figurinhas automaticas do Gemini limitadas a 6 por hora
- rosto entende a ordem "calma" enviada pelo E-Stop

Token so e exigido quando configurado (docs/SEGURANCA.md).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.gitignore

@@ -22,3 +22,4 @@ api-privada.json
 api-privada.tmp
 
 ia-config.json
+seguranca.json

api_config.py

@@ -2,6 +2,7 @@
 import json,os,re,threading,time
 from pathlib import Path
 import gemini
+from comum import gravar_atomico, ler_json
 
 class ConfigAPI:
     def __init__(self,path):
@@ -25,9 +26,7 @@ class ConfigAPI:
             key=d.get('chave') or os.environ.get('GEMINI_API_KEY','')
             saved=dict(modelo=d['modelo'])
             if key:saved['chave']=key
-            tmp=self.path.with_suffix('.tmp');fd=os.open(str(tmp),os.O_WRONLY|os.O_CREAT|os.O_TRUNC,0o600)
-            with os.fdopen(fd,'w',encoding='utf-8') as f:json.dump(saved,f)
-            os.chmod(tmp,0o600);tmp.replace(self.path);os.chmod(self.path,0o600)
+            gravar_atomico(self.path,saved,privado=True)
             gemini.MODELO=d['modelo']
             if key:os.environ['GEMINI_API_KEY']=key
             self.ultimo_teste=None
@@ -47,11 +46,7 @@ class ConfigAPI:
         try:
             if method=='GET' and route.endswith('/status'):result=self.status()
             elif method=='POST':
-                origin=h.headers.get('Origin')
-                if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')):raise ValueError('Origem inválida')
-                size=int(h.headers.get('Content-Length',0))
-                if not 0<size<=2048 or h.headers.get('Content-Type','').split(';')[0]!='application/json':raise ValueError('Use JSON até 2048 bytes')
-                d=json.loads(h.rfile.read(size))
+                d=ler_json(h,2048)
                 if route.endswith('/salvar'):result=self.salvar(d)
                 elif route.endswith('/testar'):result=self.testar()
                 else:raise ValueError('Rota inválida')

braco.py

@@ -3,11 +3,25 @@ import copy
 import json
 import math
 from pathlib import Path
+import threading
 import time
+from comum import Conflito, gravar_atomico, ler_json
 from servos import integer
 
 NAMES=['Garra','Punho 1','Punho 2','Braço superior','Braço inferior','Antebraço','Base']
 CHANNELS=[0,1,2,3,4,5,7]
+LEASE_PADRAO=2000   # ms; o painel novo manda prazos curtos e renova enquanto o botão está pressionado
+PASSO=.02           # 50 Hz de interpolação no servidor
+
+def velocidade_valida(v):
+    if v is None: return None
+    if isinstance(v,bool) or not isinstance(v,(int,float)) or not math.isfinite(v) or not 5<=v<=360:
+        raise ValueError('Velocidade deve ficar entre 5 e 360 graus/s')
+    return float(v)
+
+def prazo_valido(ms):
+    if ms is None: return LEASE_PADRAO
+    return integer(ms,100,2000)
 
 class Braco:
     def __init__(self,path,servos):
@@ -15,6 +29,8 @@ class Braco:
         self.config=dict(calibracao=[dict(canal=i,minimo=1400,maximo=1600) for i in range(8)],posturas=[])
         try: self.config=self.validate(json.loads(self.path.read_text(encoding='utf-8')))
         except (OSError,ValueError,TypeError): pass
+        self.atual=None; self.alvo=None; self.velocidade=None; self.lease=0
+        threading.Thread(target=self.loop,daemon=True).start()
     def validate_pose(self,pose,calibration):
         if not isinstance(pose,list) or len(pose)!=7: raise ValueError('Esperava sete ângulos')
         pulses={}
@@ -53,27 +69,64 @@ class Braco:
         with self.servos.lock:
             result=copy.deepcopy(self.config)
             result.update(disponivel=self.servos.hw is not None,erro=self.servos.error,
+                          dono=self.servos.dono,travado=self.servos.travado,atual=self.atual,
+                          movendo=self.alvo is not None,
                           articulacoes=[dict(nome=name,canal=ch) for name,ch in zip(NAMES,CHANNELS)])
             return result
     def save(self,data):
         clean=self.validate(data)
         with self.servos.lock:
             if self.servos.active: raise ValueError('Pare os motores antes de salvar')
-            tmp=self.path.with_suffix('.tmp'); tmp.write_text(json.dumps(clean,ensure_ascii=False,indent=2),encoding='utf-8'); tmp.replace(self.path)
+            gravar_atomico(self.path,clean)
             self.config=clean
         return self.status()
-    def move(self,pose):
+    def escrever(self,pulses,prazo):
+        try:
+            for channel,pulse in pulses.items():
+                self.servos.hw.pulse(channel,pulse)
+                self.servos.active[channel]=prazo
+        except OSError:
+            self.alvo=None; self.servos.stop()
+            raise
+    def move(self,pose,velocidade=None,ttl_ms=None):
+        """Sem velocidade: vai direto (compatível). Com velocidade: o servidor interpola até o alvo.
+        Em ambos os casos os motores desligam sozinhos quando o prazo (ttl_ms) vence sem renovação."""
+        velocidade=velocidade_valida(velocidade); ttl=prazo_valido(ttl_ms)
         with self.servos.lock:
             pulses=self.validate_pose(pose,self.config['calibracao'])
             if self.servos.hw is None: raise OSError('PCA9685 indisponível')
-            try:
-                for channel,pulse in pulses.items():
-                    self.servos.hw.pulse(channel,pulse)
-                    self.servos.active[channel]=time.monotonic()+2
-            except OSError:
-                self.servos.stop()
-                raise
-        return dict(ok=True,pulsos=pulses)
+            self.servos.reservar('braco')
+            prazo=time.monotonic()+ttl/1000
+            if velocidade is None or self.atual is None:
+                self.alvo=None
+                self.escrever(pulses,prazo)
+                self.atual=[float(a) for a in pose]
+                return dict(ok=True,pulsos=pulses,movendo=False)
+            self.alvo=[float(a) for a in pose]; self.velocidade=velocidade; self.lease=prazo
+            for channel in list(self.servos.active): self.servos.active[channel]=prazo
+        return dict(ok=True,pulsos=pulses,movendo=True)
+    def cancelar(self):
+        with self.servos.lock: self.alvo=None
+    def passo(self,agora,dt):
+        with self.servos.lock:
+            if self.alvo is None: return
+            if agora>=self.lease or self.servos.travado or (self.servos.dono!='braco' and self.servos.active):
+                self.alvo=None; return
+            delta=self.velocidade*dt
+            nova=[a+max(-delta,min(delta,t-a)) for a,t in zip(self.atual,self.alvo)]
+            try: pulses=self.validate_pose(nova,self.config['calibracao'])
+            except ValueError:
+                self.alvo=None; return
+            self.escrever(pulses,self.lease)
+            self.atual=nova
+            if nova==self.alvo: self.alvo=None
+    def loop(self):
+        anterior=time.monotonic()
+        while not self.servos.closed.wait(PASSO):
+            agora=time.monotonic()
+            try: self.passo(agora,agora-anterior)
+            except OSError as e: self.servos.error=str(e)
+            anterior=agora
     @staticmethod
     def angle_pulse(angle,center):
         return round(500+angle*(center-500)/90 if angle<=90 else center+(angle-90)*(2500-center)/90)
@@ -86,6 +139,7 @@ class Braco:
             outputs[other]=integer(center_sum-pulse,500,2500)
         with self.servos.lock:
             if self.servos.hw is None: raise OSError('PCA9685 indisponível')
+            self.servos.reservar('braco'); self.alvo=None; self.atual=None
             for ch,us in outputs.items():
                 self.servos.hw.pulse(ch,us)
                 self.servos.active[ch]=time.monotonic()+2
@@ -96,21 +150,16 @@ class Braco:
         try:
             if method=='GET' and route=='/braco/status': result=self.status()
             elif method=='POST':
-                origin=h.headers.get('Origin')
-                if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')): raise ValueError('Origem inválida')
-                if h.headers.get('Content-Type','').split(';')[0]!='application/json': raise ValueError('Use JSON')
-                size=int(h.headers.get('Content-Length','0'))
-                if not 0<size<=65536: raise ValueError('Tamanho inválido')
-                data=json.loads(h.rfile.read(size))
-                if not isinstance(data,dict): raise ValueError('Objeto obrigatório')
+                data=ler_json(h,65536)
                 if route=='/braco/config': result=self.save(data)
-                elif route=='/braco/mover': result=self.move(data.get('angulos'))
+                elif route=='/braco/mover': result=self.move(data.get('angulos'),data.get('velocidade'),data.get('ttl_ms'))
                 elif route=='/braco/calibrar': result=self.calibrate(data.get('canal'),data.get('pulso'))
-                elif route=='/braco/parar': result=self.servos.stop()
+                elif route=='/braco/parar': self.cancelar(); result=self.servos.stop()
                 else: raise ValueError('Rota inválida')
             else:
                 h.enviar(json.dumps(dict(erro='Método inválido')),codigo=405); return True
             h.enviar(json.dumps(result,ensure_ascii=False))
+        except Conflito as e: h.enviar(json.dumps(dict(erro=str(e)),ensure_ascii=False),codigo=409)
         except (ValueError,TypeError) as e: h.enviar(json.dumps(dict(erro=str(e))),codigo=400)
         except OSError as e: h.enviar(json.dumps(dict(erro=str(e))),codigo=503)
         return True

carinha.html

@@ -424,6 +424,9 @@ function trata(ordem) {
     animar(LIVRES[Math.floor(Math.random() * LIVRES.length)]);
   } else if (SEQS[ordem]) {
     animar(ordem);
+  } else if (ordem === "calma") {
+    // E-Stop do painel: interrompe a animação e volta ao rosto neutro.
+    r.fila = []; r.kf = null; r.boca = "neutra"; r.dormindo = false;
   } else if (ordem === "parar") {
     try { window.close(); } catch (e) { r.fila = []; }
   }

cerebro.py

@@ -5,6 +5,7 @@ from pathlib import Path
 import threading
 import time
 from urllib.request import Request,urlopen
+from comum import gravar_atomico, ler_json
 
 GESTOS=('piscar','piscadinha','feliz','surpreso','coracao','olhar_esq','olhar_dir','olhar_baixo')
 EVENTOS=('som','movimento','saudacao','sozinho')
@@ -20,15 +21,17 @@ class Cerebro:
         self.executor='S10' if '192.168.1.16' in self.endpoint else 'Orange Pi'
         self.lock=threading.Lock();self.ativo=False;self.pending=None;self.busy=False
         self.conversando=False;self.history=[];self.result=None;self.error='';self.last=0;self.generation=0
+        self.publicar=lambda tipo,dados:None
         try:self.ativo=json.loads(self.path.read_text())['ativo'] is True
         except (OSError,ValueError,KeyError,TypeError):pass
         threading.Thread(target=self.loop,daemon=True).start()
     def config(self,ativo):
         if type(ativo) is not bool:raise ValueError('ativo deve ser booleano')
         with self.lock:
-            tmp=self.path.with_suffix('.tmp');tmp.write_text(json.dumps(dict(ativo=ativo)));tmp.replace(self.path)
+            gravar_atomico(self.path,dict(ativo=ativo))
             self.ativo=ativo;self.generation+=1;self.pending=None
-        return self.status()
+        status=self.status();self.publicar('ia',status)
+        return status
     def status(self):
         with self.lock:return dict(ativo=self.ativo,ocupado=self.busy,modelo='Qwen3-0.6B Q4_K_M',executor=self.executor,ultima=self.result,erro=self.error)
     def event(self,event):
@@ -77,6 +80,7 @@ class Cerebro:
             finally:
                 self.last=time.monotonic()
                 with self.lock:self.busy=False
+                self.publicar('ia',self.status())
     def rotas(self,h,method):
         route=h.path.split('?',1)[0]
         if route=='/ia/modelo' and method=='GET':
@@ -95,13 +99,7 @@ class Cerebro:
         try:
             if method=='GET' and route=='/ia/status':result=self.status()
             elif method=='POST':
-                origin=h.headers.get('Origin')
-                if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')):raise ValueError('Origem inválida')
-                if h.headers.get('Content-Type','').split(';')[0]!='application/json':raise ValueError('Use JSON')
-                size=int(h.headers.get('Content-Length','0'))
-                if not 0<size<=2048:raise ValueError('Tamanho inválido')
-                data=json.loads(h.rfile.read(size))
-                if not isinstance(data,dict):raise ValueError('Objeto obrigatório')
+                data=ler_json(h,2048)
                 if route=='/ia/config':result=self.config(data.get('ativo'))
                 elif route=='/ia/evento':result=self.event(data.get('evento'))
                 else:raise ValueError('Rota inválida')

comum.py

@@ -0,0 +1,76 @@
+"""Peças compartilhadas pelos módulos HTTP: leitura de JSON, gravação atômica e token."""
+import hmac
+import json
+import os
+from pathlib import Path
+from urllib.parse import parse_qs, urlsplit
+
+
+class Conflito(Exception):
+    """Pedido válido, mas o estado atual não permite (vira HTTP 409)."""
+
+
+def origem_valida(h):
+    origin = h.headers.get('Origin')
+    host = h.headers.get('Host', '')
+    return not origin or origin in ('http://' + host, 'https://' + host)
+
+
+def ler_json(h, limite):
+    """Corpo JSON de um POST do painel; ValueError com mensagem legível se inválido."""
+    if not origem_valida(h):
+        raise ValueError('Origem inválida')
+    if h.headers.get('Content-Type', '').split(';')[0] != 'application/json':
+        raise ValueError('Use JSON')
+    tamanho = int(h.headers.get('Content-Length', '0'))
+    if not 0 < tamanho <= limite:
+        raise ValueError(f'Corpo vazio ou maior que {limite} bytes')
+    dados = json.loads(h.rfile.read(tamanho))
+    if not isinstance(dados, dict):
+        raise ValueError('Objeto JSON obrigatório')
+    return dados
+
+
+def gravar_atomico(caminho, obj, privado=False):
+    caminho = Path(caminho)
+    tmp = caminho.with_suffix('.tmp')
+    texto = json.dumps(obj, ensure_ascii=False, indent=2)
+    if privado:
+        fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(fd, 'w', encoding='utf-8') as f:
+            f.write(texto)
+    else:
+        tmp.write_text(texto, encoding='utf-8')
+    tmp.replace(caminho)
+    if privado:
+        os.chmod(caminho, 0o600)
+
+
+def carregar_token(pasta):
+    """CARINHA_TOKEN do ambiente ou {"token": ...} em seguranca.json; None = sem autenticação."""
+    token = os.environ.get('CARINHA_TOKEN', '').strip()
+    if not token:
+        try:
+            token = str(json.loads((Path(pasta) / 'seguranca.json').read_text(encoding='utf-8')).get('token', '')).strip()
+        except (OSError, ValueError, AttributeError):
+            token = ''
+    if token and len(token) < 8:
+        raise ValueError('CARINHA_TOKEN precisa ter pelo menos 8 caracteres')
+    return token or None
+
+
+def token_do_pedido(h):
+    auth = h.headers.get('Authorization', '')
+    if auth.lower().startswith('bearer '):
+        return auth[7:].strip()
+    if h.headers.get('X-Carinha-Token'):
+        return h.headers['X-Carinha-Token'].strip()
+    # EventSource e o app Android não mandam cabeçalhos próprios com facilidade.
+    return parse_qs(urlsplit(h.path).query).get('t', [''])[0]
+
+
+def autorizado(h):
+    esperado = getattr(h.server, 'token', None)
+    if not esperado:
+        return True
+    return hmac.compare_digest(token_do_pedido(h).encode(), esperado.encode())

controle_pc.py

@@ -7,8 +7,16 @@ from pathlib import Path
 import subprocess
 import threading
 import time
+from comum import origem_valida
 from sinalizacao import Sinalizacao
 
+
+def versao():
+    try:
+        return json.loads((Path(__file__).with_name('version.json')).read_text(encoding='utf-8'))['versao']
+    except (OSError, ValueError, KeyError):
+        return 'desconhecida'
+
 PADRAO = dict(preview=False, auto=True, audio=True, volume=0.35,
               ouvir=True, reagir=True, somlim=0.35, somganho=0.4,
               somaltolim=0.55, sompausa=20, somcooldown=5,
@@ -61,6 +69,7 @@ class ControlePC:
         self.frame_at = 0
         self.frame_seq = 0
         self.rtc = Sinalizacao()
+        self.publicar = lambda tipo, dados: None
         try:
             self.config.update(validar(json.loads(self.arquivo.read_text(encoding='utf-8'))))
         except (OSError, ValueError):
@@ -80,6 +89,17 @@ class ControlePC:
                 self.frame_at = 0
             return novo.copy()
 
+    def resumo(self, host=''):
+        """Estado do celular para o painel (GET /app/status, /estado e evento 'telemetria')."""
+        with self.lock:
+            agora = time.monotonic()
+            idade = agora - self.status_at
+            return dict(online=idade < 5, idade=round(idade, 1) if self.status_at else None,
+                        celular_ip=self.celular_ip, servidor_ip=host, recarga=self.recarga.copy(),
+                        estado=self.estado.copy(), preview=self.config['preview'],
+                        camera_recente=bool(self.frame) and agora - self.frame_at < 3 and self.config['preview'],
+                        frame_seq=self.frame_seq)
+
     def rotas(self, h, metodo):
         if self.rtc.atender(h, metodo, self.config['preview']):
             return True
@@ -92,6 +112,10 @@ class ControlePC:
 
         try:
             if metodo == 'POST':
+                if not origem_valida(h):
+                    raise ValueError('Origem inválida')
+                if rota == '/app/config' and h.headers.get('Content-Type', '').split(';')[0] != 'application/json':
+                    raise ValueError('Use JSON')
                 tamanho = int(h.headers.get('Content-Length', '0'))
                 limite = 1_000_000 if rota == '/camera/frame' else 16384
                 if tamanho <= 0 or tamanho > limite:
@@ -105,10 +129,12 @@ class ControlePC:
                 dados = None
 
             if rota == '/app/health':
-                enviar({'projeto': 'carinha-pc', 'versao': '0.7.0'})
+                enviar({'projeto': 'carinha-pc', 'versao': versao()})
             elif rota == '/app/config':
                 if metodo == 'POST':
-                    enviar({'ok': True, 'config': self.atualizar(dados)})
+                    config = self.atualizar(dados)
+                    self.publicar('app_config', config)
+                    enviar({'ok': True, 'config': config})
                 else:
                     with self.lock:
                         config = self.config.copy()
@@ -123,17 +149,10 @@ class ControlePC:
                         if self.recarga["id"] and str(dados.get("tela_atualizada", "")) == self.recarga["id"]:
                             self.recarga["confirmada"] = True
                         self.status_at = time.monotonic()
+                    self.publicar('telemetria', self.resumo(h.headers.get("Host", "").split(":")[0]))
                     enviar({'ok': True})
                 else:
-                    with self.lock:
-                        idade = time.monotonic() - self.status_at
-                        frame_idade = time.monotonic() - self.frame_at
-                        resultado = dict(online=idade < 5, idade=round(idade, 1) if self.status_at else None,
-                            celular_ip=self.celular_ip, servidor_ip=h.headers.get("Host", "").split(":")[0], recarga=self.recarga.copy(),
-                            estado=self.estado.copy(), preview=self.config['preview'],
-                            camera_recente=bool(self.frame) and frame_idade < 3 and self.config['preview'],
-                            frame_seq=self.frame_seq)
-                    enviar(resultado)
+                    enviar(self.resumo(h.headers.get("Host", "").split(":")[0]))
             elif rota == '/camera/frame':
                 with self.lock:
                     if metodo == 'POST':

conversa.py

@@ -6,14 +6,18 @@ import threading
 import time
 import unicodedata
 import re
+from collections import deque
 from datetime import datetime, timezone, timedelta
 from contextlib import contextmanager
 from urllib.request import Request, urlopen
 from cerebro import GESTOS
 from pathlib import Path
 from gemini import responder_gemini
+from comum import gravar_atomico, ler_json
 import gemini
 
+FIGURINHAS_POR_HORA = 6   # cada figurinha automática é mais uma chamada paga ao Gemini
+
 def resposta_factual(texto,agora=None):
     normal=''.join(c for c in unicodedata.normalize('NFD',texto.lower()) if not unicodedata.combining(c))
     normal=' '.join(normal.strip(' ?!.').split())
@@ -93,6 +97,14 @@ class Conversa:
             if saved.get('provedor') in ('local','gemini'):self.provedor=saved['provedor']
         except (OSError,ValueError,AttributeError):pass
         self.lock=threading.Lock();self.busy=False;self.ultima=None;self.erro='';self.generation=0
+        self.publicar=lambda tipo,dados:None;self.figurinhas_auto=deque()
+    def avisar(self):
+        self.publicar('conversa',self.status())
+    def pode_gerar_figurinha(self,agora=None):
+        agora=agora or time.time()
+        while self.figurinhas_auto and agora-self.figurinhas_auto[0]>3600:self.figurinhas_auto.popleft()
+        if len(self.figurinhas_auto)>=FIGURINHAS_POR_HORA:return False
+        self.figurinhas_auto.append(agora);return True
     def status(self):
         with self.lock:return dict(provedor=self.provedor,modelo=gemini.MODELO if self.provedor=='gemini' else 'Qwen3-0.6B',gemini_configurado=bool(os.environ.get('GEMINI_API_KEY')),ocupado=self.busy,ultima=self.ultima,erro=self.erro,historico=self.memoria.historico(),lembrancas=self.memoria.listar())
     def enviar(self,texto):
@@ -108,7 +120,9 @@ class Conversa:
                 self.memoria.registrar('assistant',factual['texto'],2 if self.provedor=='gemini' else 1)
                 self.ultima=dict(factual,id=time.time_ns(),horario=time.time(),segundos=0,fonte='relogio')
                 self.busy=False;self.cerebro.conversando=False
-                return dict(ok=True)
+                factual=True
+        self.avisar()
+        if factual is True:return dict(ok=True)
         threading.Thread(target=self.responder,args=(generation,),daemon=True).start()
         return dict(ok=True)
     def responder(self,generation):
@@ -140,32 +154,28 @@ class Conversa:
                 self.ultima=dict(result,id=time.time_ns(),horario=time.time(),segundos=round(time.monotonic()-inicio,2))
                 if self.provedor=='gemini' and treino:
                     if type(reaction) is int and reaction in [r['id'] for r in catalog]:treino.testar(reaction)
-                    elif not treino.status()['ocupado']:
+                    elif not treino.status()['ocupado'] and self.pode_gerar_figurinha():
                         treino.gerar_figurinha(history[-1]['content'])
         except Exception as e:
             with self.lock:self.erro=str(e)[:200] if generation==self.generation else ''
         finally:
             with self.lock,self.cerebro.lock:
                 self.busy=False;self.cerebro.conversando=False;self.cerebro.last=time.monotonic()
+            self.avisar()
     def rotas(self,h,method):
         route=h.path.split('?',1)[0]
         if not route.startswith('/conversa/'):return False
         try:
             if method=='GET' and route=='/conversa/status':result=self.status()
             elif method=='POST':
-                origin=h.headers.get('Origin')
-                if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')):raise ValueError('Origem inválida')
-                size=int(h.headers.get('Content-Length','0'))
-                if not 0<size<=4096 or h.headers.get('Content-Type','').split(';')[0]!='application/json':raise ValueError('Use JSON até 4096 bytes')
-                data=json.loads(h.rfile.read(size))
-                if not isinstance(data,dict):raise ValueError('Objeto obrigatório')
+                data=ler_json(h,4096)
                 if route=='/conversa/config':
                     with self.lock:
                         if self.busy:raise ValueError('Aguarde a resposta antes de trocar a IA')
                         provider=data.get('provedor')
                         if provider not in ('local','gemini'):raise ValueError('Provedor inválido')
                         if provider=='gemini' and not os.environ.get('GEMINI_API_KEY'):raise ValueError('Chave Gemini não configurada')
-                        tmp=self.config_path.with_suffix('.tmp');tmp.write_text(json.dumps(dict(provedor=provider)));tmp.replace(self.config_path)
+                        gravar_atomico(self.config_path,dict(provedor=provider))
                         self.provedor=provider
                     result=self.status()
                 elif route=='/conversa/enviar':result=self.enviar(data.get('texto'))
@@ -183,6 +193,7 @@ class Conversa:
                         else:self.memoria.apagar(data.get('id'))
                     result=dict(ok=True)
                 else:raise ValueError('Rota inválida')
+                if route!='/conversa/enviar':self.avisar()
             else:raise ValueError('Método inválido')
             h.enviar(json.dumps(result,ensure_ascii=False))
         except (ValueError,TypeError) as e:h.enviar(json.dumps(dict(erro=str(e)),ensure_ascii=False),codigo=400)

docs/GITEA.md

@@ -1,9 +1,21 @@
 # Repositório do Carinha
 
-Código em https://git.roterizacao.cloud/PSAutomation/painel-ia-gemini, branch `carinha-orange-pi`, pasta `carinha/`.
-O repositório já contém outro sistema; sua branch principal foi preservada.
+Código em https://git.roterizacao.cloud/julio/carinha-control-panel, branch `main`.
+(Antes vivia em `PSAutomation/painel-ia-gemini`, branch `carinha-orange-pi`, pasta `carinha/`; aquele repositório não é mais a fonte.)
 
-Credenciais, chave de assinatura Android, bases de conversa, logs, configurações do aparelho e arquivos gerados não são publicados.
+## Credenciais do Git
+
+Nunca coloque usuário/senha na URL do remote (`https://usuario:senha@...`): ela fica em texto claro em `.git/config`.
+Use a URL limpa e deixe o Git Credential Manager guardar o login:
+
+```
+git remote set-url origin https://git.roterizacao.cloud/julio/carinha-control-panel.git
+git config --global credential.helper manager
+```
+
+No primeiro `git push` o GCM pede o login (ou um token de acesso do Gitea, preferível à senha) e o guarda no cofre do Windows.
+
+## O que não é publicado
+
+Credenciais, chave de assinatura Android, bases de conversa, logs, configurações do aparelho e arquivos gerados.
 Modelos MediaPipe podem ser obtidos com `python obter-rastreamento.py`. As bibliotecas nativas llama.cpp ARM64 e a chave de assinatura Android precisam ser fornecidas separadamente para compilar o APK; permanecem na instalação atual. O servidor Python e a interface HTML estão no código versionado. Gemini usa GEMINI_API_KEY no ambiente do serviço.
-
-Esta publicação não altera a instalação ativa no Orange Pi.

docs/SEGURANCA.md

@@ -0,0 +1,43 @@
+# Segurança do servidor
+
+## Token de controle
+
+Com um token configurado, o servidor (porta 8100) exige `Authorization: Bearer <token>`
+em todo POST e nas leituras de dados (`/eventos`, `/estado`, `/fila`, `/app/*`, `/rtc/*`,
+`/camera/*`, `/conversa/*`, `/treinamento/*`, `/braco/*`, `/servos/*`, `/ia/status`, `/api/*`).
+EventSource e o app Android podem mandar `?t=<token>` na URL.
+
+Continuam públicos apenas: páginas estáticas, `/app/health`, `/faces`, `/gestos`,
+`GET /config` (aparência), `/ia/modelo` e `/app.apk`.
+
+Configure **um** dos dois (mínimo 8 caracteres):
+
+```
+# /etc/carinha.env, carregado pelo carinha-pc.service (EnvironmentFile=)
+CARINHA_TOKEN=troque-por-um-segredo-longo
+```
+
+ou `seguranca.json` na pasta do servidor (fora do git):
+
+```json
+{"token": "troque-por-um-segredo-longo"}
+```
+
+Sem token configurado o servidor fica aberto na rede local, como antes.
+
+> **Ordem de ativação.** O painel antigo e o `carinha.html`/APK atuais ainda não enviam token.
+> Ative o token na placa só depois de publicar o painel novo e o rosto/APK com suporte a token
+> (fases 2–5 do plano de modernização). Até lá o servidor já aplica as demais proteções:
+> checagem de Origin em todos os POST, comandos validados e E-Stop.
+
+## E-Stop
+
+`POST /parar-tudo` desliga as 16 saídas do PCA9685, cancela a interpolação do braço,
+para reações e manda `calma` ao rosto. Braço e servos ficam **travados** (HTTP 409) até
+`POST /rearmar`.
+
+## Trava de hardware
+
+Braço (`/braco/*`) e diagnóstico (`/servos/*`) usam o mesmo PCA9685 com limites
+diferentes. Quem estiver com motores ativos é o dono; o outro recebe 409 até os motores
+desligarem (parar, ou ~2 s sem comando).

eventos.py

@@ -0,0 +1,70 @@
+"""Hub Server-Sent Events: um único fluxo /eventos substitui os pollings do painel."""
+from collections import deque
+import json
+import threading
+import time
+from urllib.parse import parse_qs, urlsplit
+
+PING = 15
+
+
+class Eventos:
+    def __init__(self, buffer=200):
+        self.cond = threading.Condition()
+        self.seq = 0
+        self.buffer = deque(maxlen=buffer)
+        self.fechado = False
+
+    def publicar(self, tipo, dados):
+        corpo = json.dumps(dados, ensure_ascii=False)
+        with self.cond:
+            self.seq += 1
+            self.buffer.append((self.seq, tipo, corpo))
+            self.cond.notify_all()
+        return self.seq
+
+    def desde(self, seq):
+        with self.cond:
+            return [e for e in self.buffer if e[0] > seq]
+
+    def fechar(self):
+        with self.cond:
+            self.fechado = True
+            self.cond.notify_all()
+
+    def atender(self, h):
+        url = urlsplit(h.path)
+        try:
+            ultimo = int(h.headers.get('Last-Event-ID') or parse_qs(url.query).get('desde', ['0'])[0])
+        except ValueError:
+            ultimo = 0
+        with self.cond:
+            # Cliente novo (ou que perdeu demais): começa do agora e lê /estado.
+            if ultimo <= 0 or (self.buffer and ultimo < self.buffer[0][0] - 1) or ultimo > self.seq:
+                ultimo = self.seq
+        h.send_response(200)
+        h.send_header('Content-Type', 'text/event-stream; charset=utf-8')
+        h.send_header('Cache-Control', 'no-store')
+        h.send_header('X-Accel-Buffering', 'no')
+        h.end_headers()
+        try:
+            h.wfile.write(f'retry: 3000\nevent: pronto\nid: {ultimo}\ndata: {{"seq": {ultimo}}}\n\n'.encode())
+            while True:
+                limite = time.monotonic() + PING
+                with self.cond:
+                    while not self.fechado and (not self.buffer or self.buffer[-1][0] <= ultimo):
+                        resta = limite - time.monotonic()
+                        if resta <= 0:
+                            break
+                        self.cond.wait(resta)
+                    if self.fechado:
+                        return
+                    novos = [e for e in self.buffer if e[0] > ultimo]
+                if not novos:
+                    h.wfile.write(b': ping\n\n')
+                    continue
+                for seq, tipo, corpo in novos:
+                    h.wfile.write(f'id: {seq}\nevent: {tipo}\ndata: {corpo}\n\n'.encode())
+                    ultimo = seq
+        except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError, OSError):
+            return

kiosk-server.py

@@ -1,38 +1,39 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
-kiosk-server.py — servidor local do PC: painel, carinha e fila de ordens.
+kiosk-server.py — servidor do Carinha (Orange Pi, porta 8100): painel, rosto e controle.
 
-Alem de entregar os arquivos da pasta (padrao /root/carinha), ele expõe:
+Estado em tempo real:
+    GET  /eventos        -> Server-Sent Events (expressao, telemetria, conversa, reacao,
+                            ia, motores, estop, aparencia, app_config, treinamento)
+    GET  /estado         -> foto de tudo para o primeiro carregamento (inclui 'seq')
 
-    GET  /fila          -> devolve e limpa a fila, em JSON: ["feliz","piscar"]
-    GET  /gesto/<nome>  -> poe <nome> na fila   (ex.: /gesto/coracao)
-    POST /gesto         -> corpo = nome do gesto
+Comandos:
+    POST /comando        -> {"tipo": "gesto"|"rosto"|"cor"|"reacao"|"parar_reacao"|"recarregar", "valor": ...}
+    POST /parar-tudo     -> E-Stop: desliga todos os servos, para reações e acalma o rosto
+    POST /rearmar        -> libera os motores depois do E-Stop
+    GET  /fila?aparelho=1 -> ordens pendentes para o celular (até o rosto migrar para /eventos)
 
-A pagina da carinha pergunta /fila a cada meio segundo; assim da para mandar
-um gesto do PC com um curl, igual ao antigo FIFO /tmp/eilik.fifo:
+Autenticação: se CARINHA_TOKEN (ambiente) ou seguranca.json {"token": ...} existir, toda
+rota de controle/dados exige "Authorization: Bearer <token>" (ou ?t=<token> no EventSource
+e no app). Sem token configurado o servidor fica aberto como antes.
 
-    curl http://orangepi:8099/gesto/feliz
-
-Alem disso guarda a APARENCIA da carinha (rosto + cor dos olhos), para o painel
-personalizar e a pagina aplicar sozinha ao carregar:
-
-    GET  /config        -> {"rosto":"carinha","cor":[92,214,255]}
-    POST /config        -> corpo JSON {"rosto":"miudo","cor":[255,120,0]}
-
-E serve o painel de controle em http://<maquina>:8099/painel .
-
-So usa a biblioteca padrao (http.server). Nada de instalacao extra.
+Só usa a biblioteca padrão. Nada de instalação extra.
 """
 
 import argparse
+from collections import deque
 import json
 import os
 import threading
+import time
+import uuid
 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import unquote
 from pathlib import Path
-from controle_pc import ControlePC
+from comum import autorizado, carregar_token, gravar_atomico, ler_json
+from controle_pc import ControlePC, versao
+from eventos import Eventos
 from servos import ControleServos
 from braco import Braco
 from cerebro import Cerebro
@@ -41,7 +42,6 @@ from treinamento import Treinamento
 from api_config import ConfigAPI
 
 TRAVA = threading.Lock()
-FILA = []
 
 # Espelho da biblioteca de rostos do carinha.html (para /faces funcionar)
 ROSTOS = ["eilik", "carinha", "miudo"]
@@ -50,12 +50,65 @@ ROSTOS = ["eilik", "carinha", "miudo"]
 GESTOS = ["piscar", "piscar2", "olhar_esq", "olhar_dir", "olhar_baixo", "feliz",
           "surpreso", "bravo", "coracao", "triste", "piscadinha", "alegre",
           "dormir", "acordar", "aleatorio", "parar"]
+# Ordens internas que o rosto entende mas não viram botão.
+ORDENS_INTERNAS = {"calma"}
 
-# Apar┌┐ncia atual da carinha (personalizacao pelo painel). Fica gravada num
+# Aparência atual da carinha (personalizacao pelo painel). Fica gravada num
 # arquivo ao lado do servidor, para sobreviver a reinicios.
 APARENCIA = {"rosto": "carinha", "cor": [92, 214, 255]}
 ARQUIVO_APARENCIA = "aparencia.json"
 
+# Rotas que não exigem token: só o necessário para a página do rosto abrir e o app se achar.
+PUBLICAS_GET = {"/app/health", "/faces", "/gestos", "/config", "/ia/modelo", "/app.apk"}
+PREFIXOS_API = ("/eventos", "/estado", "/fila", "/rtc/", "/camera/", "/conversa/", "/treinamento/",
+                "/app/", "/braco/", "/servos/", "/ia/", "/api/", "/gesto/", "/face/", "/diag/")
+
+
+class Fila:
+    """Ordens para o celular: limitada e com validade, para não repetir ordens velhas
+    quando o aparelho volta depois de um tempo offline."""
+
+    def __init__(self, maximo=20, validade=10.0):
+        self.lock = threading.Lock()
+        self.itens = deque(maxlen=maximo)
+        self.validade = validade
+
+    def adicionar(self, ordem, agora=None):
+        with self.lock:
+            if ordem == "recarregar" and any(o == ordem for _, o in self.itens):
+                return
+            self.itens.append((agora if agora is not None else time.monotonic(), ordem))
+
+    def consumir(self, aparelho=True, agora=None):
+        """O celular leva tudo; um rosto aberto no navegador (kiosk) leva tudo menos 'recarregar'."""
+        agora = agora if agora is not None else time.monotonic()
+        with self.lock:
+            validos = [(t, o) for t, o in self.itens if agora - t <= self.validade]
+            pendentes = [o for _, o in validos if aparelho or o != "recarregar"]
+            self.itens.clear()
+            if not aparelho:
+                self.itens.extend((t, o) for t, o in validos if o == "recarregar")
+            return pendentes
+
+    def ordens(self):
+        with self.lock:
+            return [o for _, o in self.itens]
+
+    def limpar(self):
+        with self.lock:
+            self.itens.clear()
+
+
+FILA = Fila()
+EVENTOS = Eventos()
+ESTOP = {"ativo": False, "desde": None}
+
+
+def enfileirar(ordem):
+    """Caminho único de uma ordem para o rosto: fila do celular + evento para os painéis."""
+    FILA.adicionar(ordem)
+    EVENTOS.publicar("expressao", {"ordem": ordem})
+
 
 def carregar_aparencia():
     try:
@@ -73,14 +126,20 @@ def carregar_aparencia():
 
 def salvar_aparencia():
     try:
-        with open(ARQUIVO_APARENCIA, "w", encoding="utf-8") as f:
-            json.dump(APARENCIA, f, ensure_ascii=False, indent=2)
-    except Exception:
+        gravar_atomico(ARQUIVO_APARENCIA, APARENCIA)
+    except OSError:
         pass
 
 
+def validar_cor(valor):
+    cor = [max(0, min(255, int(v))) for v in list(valor)[:3]]
+    if len(cor) != 3:
+        raise ValueError("Cor precisa de três números")
+    return cor
+
+
 def aplicar_aparencia(dados):
-    """Guarda rosto/cor e poe as ordens na fila para a pagina aplicar na hora."""
+    """Guarda rosto/cor e manda as ordens para a página aplicar na hora."""
     ordens = []
     with TRAVA:
         if "rosto" in dados:
@@ -90,24 +149,43 @@ def aplicar_aparencia(dados):
                 ordens.append("rosto " + nome)
         if "cor" in dados:
             try:
-                cor = [max(0, min(255, int(v))) for v in list(dados["cor"])[:3]]
-                if len(cor) == 3:
-                    APARENCIA["cor"] = cor
-                    ordens.append("cor %d,%d,%d" % tuple(cor))
-            except Exception:
+                APARENCIA["cor"] = validar_cor(dados["cor"])
+                ordens.append("cor %d,%d,%d" % tuple(APARENCIA["cor"]))
+            except (TypeError, ValueError):
                 pass
-        FILA.extend(ordens)
-    salvar_aparencia()
+        aparencia = dict(APARENCIA)
+    for ordem in ordens:
+        enfileirar(ordem)
+    if ordens:
+        salvar_aparencia()
+        EVENTOS.publicar("aparencia", aparencia)
     return ordens
 
 
+def ordem_de_comando(tipo, valor):
+    """Traduz {tipo, valor} para a ordem de texto que o rosto entende; ValueError se inválido."""
+    if tipo == "gesto":
+        nome = str(valor or "").strip().lower()
+        if nome not in GESTOS and nome not in ORDENS_INTERNAS:
+            raise ValueError("Gesto desconhecido: " + nome)
+        return nome
+    if tipo == "rosto":
+        nome = str(valor or "").strip().lower()
+        if nome not in ROSTOS:
+            raise ValueError("Rosto desconhecido: " + nome)
+        return "rosto " + nome
+    raise ValueError("Tipo de comando inválido")
+
+
 class Handler(SimpleHTTPRequestHandler):
     extensoes_map = {".html": "text/html; charset=utf-8",
                      ".js": "text/javascript; charset=utf-8",
+                     ".mjs": "text/javascript; charset=utf-8",
                      ".css": "text/css; charset=utf-8",
                      ".json": "application/json; charset=utf-8",
                      ".svg": "image/svg+xml",
                      ".png": "image/png",
+                     ".woff2": "font/woff2",
                      ".mp4": "video/mp4"}
 
     def guess_type(self, caminho):
@@ -120,32 +198,71 @@ class Handler(SimpleHTTPRequestHandler):
         self.send_header("Content-Type", tipo)
         self.send_header("Content-Length", str(len(dados)))
         self.send_header("Cache-Control", "no-store")
+        if codigo == 401:
+            self.send_header("WWW-Authenticate", 'Bearer realm="carinha"')
         self.end_headers()
         self.wfile.write(dados)
 
+    def json(self, dados, codigo=200):
+        self.enviar(json.dumps(dados, ensure_ascii=False), codigo=codigo)
+
+    def rota(self):
+        return self.path.split("?", 1)[0]
+
+    def acesso_negado(self, metodo):
+        """True (e já respondeu 401) se a rota exige token e o pedido não trouxe o certo."""
+        rota = self.rota()
+        protegida = metodo == "POST" or (rota not in PUBLICAS_GET and rota.startswith(PREFIXOS_API))
+        if protegida and not autorizado(self):
+            self.json({"erro": "Token ausente ou inválido"}, 401)
+            return True
+        return False
+
+    def modulos(self):
+        s = self.server
+        return [getattr(s, nome) for nome in ("api_config", "treinamento", "conversa", "cerebro",
+                                              "braco", "servos", "controle") if hasattr(s, nome)]
+
+    def estado(self):
+        s = self.server
+        host = self.headers.get("Host", "").split(":")[0]
+        with TRAVA:
+            aparencia = dict(APARENCIA)
+        dados = dict(versao=versao(), seq=EVENTOS.seq, autenticacao=bool(getattr(s, "token", None)),
+                     estop=dict(ESTOP), aparencia=aparencia, gestos=GESTOS, rostos=ROSTOS)
+        if hasattr(s, "controle"):
+            dados.update(app=s.controle.resumo(host), app_config=s.controle.config.copy())
+        if hasattr(s, "servos"):
+            dados["motores"] = s.servos.motores()
+        if hasattr(s, "conversa"):
+            dados["conversa"] = s.conversa.status()
+        if hasattr(s, "cerebro"):
+            dados["ia"] = s.cerebro.status()
+        if hasattr(s, "treinamento"):
+            dados["reacao"] = s.treinamento.expressao
+            dados["treinamento"] = dict(ocupado=s.treinamento.busy, erro=s.treinamento.erro)
+        if hasattr(s, "api_config"):
+            dados["api"] = s.api_config.status()
+        return dados
+
     def do_GET(self):
-        if hasattr(self.server, "api_config") and self.server.api_config.rotas(self, "GET"):return
-        if hasattr(self.server, 'treinamento') and self.server.treinamento.rotas(self, 'GET'):
+        if self.acesso_negado("GET"):
             return
-        if hasattr(self.server, 'conversa') and self.server.conversa.rotas(self, 'GET'):
-            return
-        if hasattr(self.server, 'cerebro') and self.server.cerebro.rotas(self, 'GET'):
-            return
-        if hasattr(self.server, 'braco') and self.server.braco.rotas(self, 'GET'):
-            return
-        if hasattr(self.server, 'servos') and self.server.servos.rotas(self, 'GET'):
-            return
-        if self.server.controle.rotas(self, 'GET'):
-            return
-        caminho = self.path.split("?", 1)[0]
+        for modulo in self.modulos():
+            if modulo.rotas(self, "GET"):
+                return
+        caminho = self.rota()
         partes = [p for p in caminho.split("/") if p]
 
+        if caminho == "/eventos":
+            return EVENTOS.atender(self)
+
+        if caminho == "/estado":
+            return self.json(self.estado())
+
         if partes and partes[0] == "fila":
-            with TRAVA:
-                aparelho = 'aparelho=1' in self.path
-                pendentes = [x for x in FILA if x!='recarregar' or aparelho]
-                FILA[:] = [x for x in FILA if x=='recarregar' and not aparelho]
-            return self.enviar(json.dumps(pendentes))
+            # Painéis não leem a fila: observam o rosto por /eventos.
+            return self.enviar(json.dumps(FILA.consumir(aparelho="aparelho=1" in self.path)))
 
         if partes and partes[0] == "faces":
             return self.enviar(json.dumps({"rostos": ROSTOS}))
@@ -156,12 +273,6 @@ class Handler(SimpleHTTPRequestHandler):
         if partes and partes[0] == "gestos":
             return self.enviar(json.dumps({"gestos": GESTOS}))
 
-        if len(partes) == 2 and partes[0] == "face":
-            nome = unquote(partes[1]).lower()
-            with TRAVA:
-                FILA.append("rosto " + nome)
-            return self.enviar(json.dumps({"ok": True, "rosto": nome}))
-
         if partes and partes[0] == "diag":
             # a propria pagina relata o tamanho que recebeu (ajuda a achar
             # barras/enquadramento errado no kiosk). Aparece no journal.
@@ -169,11 +280,15 @@ class Handler(SimpleHTTPRequestHandler):
             print("DIAG kiosk:", info, flush=True)
             return self.enviar(json.dumps({"ok": True}))
 
-        if len(partes) == 2 and partes[0] == "gesto":
-            nome = unquote(partes[1]).lower()   # aceita %20 para espacos
-            with TRAVA:
-                FILA.append(nome)
-            return self.enviar(json.dumps({"ok": True, "gesto": nome}))
+        # Obsoletos (GET que muda estado); ficam até o painel antigo sair. Use POST /comando.
+        if len(partes) == 2 and partes[0] in ("gesto", "face"):
+            tipo = "gesto" if partes[0] == "gesto" else "rosto"
+            try:
+                ordem = ordem_de_comando(tipo, unquote(partes[1]))
+            except ValueError as e:
+                return self.json({"erro": str(e)}, 400)
+            enfileirar(ordem)
+            return self.json({"ok": True, partes[0]: ordem})
 
         if caminho in ("/dashboard", "/emojis", "/api"):
             self.path = caminho + ".html"
@@ -201,60 +316,103 @@ class Handler(SimpleHTTPRequestHandler):
         elif caminho in ('/dashboard.html','/emojis.html','/api.html','/dashboard.css','/dashboard.js','/overview.js','/biblioteca.js','/api-painel.js','/painel.css','/conversa.css','/treinamento.css','/servos.css','/braco.css','/audio-robot.js'):
             pass
         elif caminho not in ('/conversa.html', '/treinamento.html', '/treinamento.js', '/reacoes-render.js', '/painel-conversa.js', '/braco-calibracao.js', '/movimentos-core.js', '/braco-movimentos.js', '/braco.html', '/braco.js', '/servos.html', '/servos.js', '/carinha.html', '/camera.html', '/painel.html', '/painel-controle.js', '/painel-video.js',
-            '/rastreamento-core.js', '/rastreador-worker.js', '/painel-rastreamento.js', '/audio-robot.js'):
+            '/rastreamento-core.js', '/rastreador-worker.js', '/painel-rastreamento.js'):
             return self.enviar(json.dumps({'erro': 'Arquivo não disponível'}), codigo=404)
         return super().do_GET()
 
     def do_POST(self):
-        if hasattr(self.server, "api_config") and self.server.api_config.rotas(self, "POST"):return
-        if hasattr(self.server, 'treinamento') and self.server.treinamento.rotas(self, 'POST'):
+        if self.acesso_negado("POST"):
             return
-        if hasattr(self.server, 'conversa') and self.server.conversa.rotas(self, 'POST'):
-            return
-        if hasattr(self.server, 'cerebro') and self.server.cerebro.rotas(self, 'POST'):
-            return
-        if hasattr(self.server, 'braco') and self.server.braco.rotas(self, 'POST'):
-            return
-        if hasattr(self.server, 'servos') and self.server.servos.rotas(self, 'POST'):
-            return
-        if self.server.controle.rotas(self, 'POST'):
-            return
-        rota = self.path.split("?", 1)[0].rstrip("/")
-        tamanho = int(self.headers.get("Content-Length") or 0)
+        for modulo in self.modulos():
+            if modulo.rotas(self, "POST"):
+                return
+        rota = self.rota().rstrip("/")
+        try:
+            if rota == "/comando":
+                return self.comando(ler_json(self, 2048))
+            if rota == "/parar-tudo":
+                ler_json(self, 2048)
+                return self.json(self.parar_tudo())
+            if rota == "/rearmar":
+                ler_json(self, 2048)
+                return self.json(self.rearmar())
+            if rota == '/app/recarregar':
+                ler_json(self, 2048)
+                return self.recarregar()
+            if rota == "/config":
+                return self.json({"ok": True, "ordens": aplicar_aparencia(ler_json(self, 2048)),
+                                  "aparencia": APARENCIA})
+            if rota in ("/gesto", "/face"):
+                # Obsoleto: corpo é o nome cru. Use POST /comando.
+                tamanho = int(self.headers.get("Content-Length") or 0)
+                nome = self.rfile.read(min(tamanho, 200)).decode("utf-8", "ignore")
+                ordem = ordem_de_comando("gesto" if rota == "/gesto" else "rosto", nome)
+                enfileirar(ordem)
+                return self.json({"ok": True, "ordem": ordem})
+        except ValueError as e:
+            return self.json({"erro": str(e)}, 400)
+        return self.json({"erro": "Rota POST desconhecida"}, 404)
 
-        if rota == '/app/recarregar':
-            origin=self.headers.get('Origin')
-            if origin and origin not in ('http://'+self.headers.get('Host',''),'https://'+self.headers.get('Host','')):
-                return self.enviar(json.dumps(dict(erro='Origem inválida')),codigo=400)
-            import time
-            with self.server.controle.lock:
-                online=time.monotonic()-self.server.controle.status_at<5
-                import uuid
-                self.server.controle.recarga = dict(id=uuid.uuid4().hex, confirmada=False)
-            if not online:return self.enviar(json.dumps(dict(erro='Abra o Carinha e conecte ao servidor primeiro')),codigo=409)
-            with TRAVA:
-                if 'recarregar' not in FILA:FILA.append('recarregar')
-            return self.enviar(json.dumps(dict(ok=True,mensagem='Solicitação enviada. Aguardando o celular recarregar a tela.')))
+    def comando(self, dados):
+        tipo, valor = dados.get("tipo"), dados.get("valor")
+        s = self.server
+        if tipo == "cor":
+            if not aplicar_aparencia({"cor": valor}):
+                raise ValueError("Cor inválida")
+            return self.json({"ok": True})
+        if tipo == "recarregar":
+            return self.recarregar()
+        if tipo in ("reacao", "parar_reacao"):
+            if not hasattr(s, "treinamento"):
+                raise ValueError("Reações indisponíveis")
+            if tipo == "reacao":
+                if type(valor) is not int:
+                    raise ValueError("ID de reação inválido")
+                s.treinamento.testar(valor)
+            else:
+                with s.treinamento.lock:
+                    s.treinamento.expressao = dict(id=time.time_ns(), horario=time.time(), receita=dict(limpar=True))
+            return self.json({"ok": True})
+        ordem = ordem_de_comando(tipo, valor)
+        enfileirar(ordem)
+        return self.json({"ok": True, "ordem": ordem})
 
-        if rota == "/config":
-            try:
-                dados = json.loads(self.rfile.read(tamanho).decode("utf-8", "ignore") or "{}")
-                if not isinstance(dados, dict):
-                    raise ValueError("esperava um objeto JSON")
-            except Exception as e:
-                return self.enviar(json.dumps({"erro": "json invalido: %s" % e}), codigo=400)
-            ordens = aplicar_aparencia(dados)
-            return self.enviar(json.dumps({"ok": True, "aparencia": APARENCIA,
-                                           "ordens": ordens}))
+    def parar_tudo(self):
+        s = self.server
+        if hasattr(s, "braco"):
+            s.braco.cancelar()
+        if hasattr(s, "servos"):
+            with s.servos.lock:
+                s.servos.travado = True
+            s.servos.stop()
+        if hasattr(s, "treinamento"):
+            with s.treinamento.lock:
+                s.treinamento.expressao = dict(id=time.time_ns(), horario=time.time(), receita=dict(limpar=True))
+        FILA.limpar()
+        enfileirar("calma")
+        ESTOP.update(ativo=True, desde=time.time())
+        EVENTOS.publicar("estop", dict(ESTOP))
+        return {"ok": True, "estop": dict(ESTOP)}
 
-        if rota not in ("/gesto", "/face"):
-            return self.enviar(json.dumps({"erro": "rotas POST: /gesto, /face, /config"}), codigo=404)
-        corpo = self.rfile.read(tamanho).decode("utf-8", "ignore").strip().lower()
-        if not corpo:
-            return self.enviar(json.dumps({"erro": "corpo vazio"}), codigo=400)
-        with TRAVA:
-            FILA.append(corpo)
-        return self.enviar(json.dumps({"ok": True, "ordem": corpo}))
+    def rearmar(self):
+        s = self.server
+        if hasattr(s, "servos"):
+            with s.servos.lock:
+                s.servos.travado = False
+            EVENTOS.publicar("motores", s.servos.motores())
+        ESTOP.update(ativo=False, desde=None)
+        EVENTOS.publicar("estop", dict(ESTOP))
+        return {"ok": True, "estop": dict(ESTOP)}
+
+    def recarregar(self):
+        controle = self.server.controle
+        with controle.lock:
+            online = time.monotonic() - controle.status_at < 5
+            controle.recarga = dict(id=uuid.uuid4().hex, confirmada=False)
+        if not online:
+            return self.json({"erro": "Abra o Carinha e conecte ao servidor primeiro"}, 409)
+        FILA.adicionar("recarregar")
+        return self.json({"ok": True, "mensagem": "Solicitação enviada. Aguardando o celular recarregar a tela."})
 
     def log_message(self, formato, *args):
         # silencio por padrao (roda em segundo plano); --log liga o registro
@@ -270,6 +428,13 @@ class Servidor(ThreadingHTTPServer):
     so o desligamos no Windows."""
     daemon_threads = True
     allow_reuse_address = (os.name != "nt")
+    token = None
+
+
+def ligar_eventos(servidor):
+    for nome in ("controle", "servos", "cerebro", "conversa", "treinamento"):
+        if hasattr(servidor, nome):
+            getattr(servidor, nome).publicar = EVENTOS.publicar
 
 
 def main():
@@ -282,21 +447,24 @@ def main():
     os.chdir(args.pasta)
     carregar_aparencia()
     servidor = Servidor(("0.0.0.0", args.porta), Handler)
+    servidor.token = carregar_token(Path.cwd())
     servidor.api_config = ConfigAPI(Path.cwd() / 'api-privada.json')
     servidor.controle = ControlePC(Path.cwd() / 'app-config.json')
     servidor.servos = ControleServos(Path.cwd() / 'servos.json')
     servidor.braco = Braco(Path.cwd() / 'braco.json', servidor.servos)
-    def expressao_ia(gesto):
-        with TRAVA: FILA.append(gesto)
-    servidor.cerebro = Cerebro(Path.cwd() / 'ia-config.json', servidor.controle, expressao_ia)
+    servidor.cerebro = Cerebro(Path.cwd() / 'ia-config.json', servidor.controle, enfileirar)
     servidor.conversa = Conversa(Path.cwd() / 'memoria.db', servidor.cerebro)
     servidor.treinamento = Treinamento(Path.cwd() / 'treinamento.db')
     servidor.conversa.memoria.on_registrar = servidor.treinamento.registrar
     servidor.conversa.treinamento = servidor.treinamento
+    ligar_eventos(servidor)
     servidor.registrar = args.log
-    print(f"servindo {args.pasta} em http://0.0.0.0:{args.porta}/carinha.html")
-    print(f"painel de controle em http://0.0.0.0:{args.porta}/painel")
-    servidor.serve_forever()
+    print(f"servindo {args.pasta} em http://0.0.0.0:{args.porta}/")
+    print("autenticação:", "token obrigatório" if servidor.token else "DESLIGADA (defina CARINHA_TOKEN)")
+    try:
+        servidor.serve_forever()
+    finally:
+        EVENTOS.fechar()
 
 
 if __name__ == "__main__":

servos.py

@@ -4,6 +4,7 @@ import os
 from pathlib import Path
 import threading
 import time
+from comum import Conflito, gravar_atomico, ler_json
 
 def integer(value, lo, hi):
     if type(value) is not int or not lo <= value <= hi:
@@ -46,6 +47,9 @@ class ControleServos:
         self.hw = hardware
         self.error = ''
         self.active = {}
+        self.dono = None          # 'braco' ou 'diagnostico': quem está com os motores
+        self.travado = False      # E-Stop: nada se move até rearmar
+        self.publicar = lambda tipo, dados: None
         self.closed = threading.Event()
         self.config = [dict(nome=f'Servo {i+1}', canal=i, minimo=1400, maximo=1600, centro=1500) for i in range(6)]
         try:
@@ -78,14 +82,24 @@ class ControleServos:
     def status(self):
         with self.lock:
             return dict(disponivel=self.hw is not None, erro=self.error,
-                        servos=[dict(row) for row in self.config], ativos=list(self.active), prazo=2)
+                        servos=[dict(row) for row in self.config], ativos=list(self.active), prazo=2,
+                        dono=self.dono, travado=self.travado)
+    def motores(self):
+        with self.lock:
+            return dict(disponivel=self.hw is not None, erro=self.error, ativos=sorted(self.active),
+                        dono=self.dono, travado=self.travado)
+    def reservar(self, nome):
+        """Braço e diagnóstico usam o mesmo PCA9685 com limites diferentes: um de cada vez."""
+        with self.lock:
+            if self.travado: raise Conflito('E-Stop ativo: rearme antes de mover')
+            if self.dono not in (None, nome) and self.active:
+                raise Conflito(f'Motores em uso por {self.dono}; pare-os primeiro')
+            self.dono = nome
     def save(self, rows):
         rows = self.validate(rows)
         with self.lock:
             if self.active: raise ValueError('Desligue os sinais antes de salvar')
-            tmp = self.path.with_suffix('.tmp')
-            tmp.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding='utf-8')
-            tmp.replace(self.path)
+            gravar_atomico(self.path, rows)
             self.config = rows
         return self.status()
     def move(self, index, pulse):
@@ -94,20 +108,28 @@ class ControleServos:
             row = self.config[index]
             pulse = integer(pulse, row['minimo'], row['maximo'])
             if self.hw is None: raise OSError('PCA9685 indisponível neste servidor')
+            self.reservar('diagnostico')
             self.hw.pulse(row['canal'], pulse)
             self.active[row['canal']] = time.monotonic() + 2
         return self.status()
     def expire(self, now):
         with self.lock:
+            liberou = False
             for channel, deadline in list(self.active.items()):
                 if now >= deadline:
                     self.hw.off(channel)
                     del self.active[channel]
+                    liberou = True
+            if liberou and not self.active:
+                self.dono = None
+                self.publicar('motores', self.motores())
     def stop(self):
         with self.lock:
             if self.hw:
                 for channel in range(16): self.hw.off(channel)
             self.active.clear()
+            self.dono = None
+            self.publicar('motores', self.motores())
         return self.status()
     def watch(self):
         while not self.closed.wait(.1):
@@ -122,14 +144,7 @@ class ControleServos:
         try:
             if method == 'GET' and route == '/servos/status': result = self.status()
             elif method == 'POST':
-                origin = h.headers.get('Origin')
-                if origin and origin not in ('http://' + h.headers.get('Host', ''), 'https://' + h.headers.get('Host', '')):
-                    raise ValueError('Origem não permitida')
-                if h.headers.get('Content-Type', '').split(';')[0] != 'application/json': raise ValueError('Use JSON')
-                size = int(h.headers.get('Content-Length', '0'))
-                if not 0 < size <= 8192: raise ValueError('Corpo inválido')
-                data = json.loads(h.rfile.read(size))
-                if not isinstance(data, dict): raise ValueError('Objeto JSON obrigatório')
+                data = ler_json(h, 8192)
                 if route == '/servos/config': result = self.save(data.get('servos'))
                 elif route == '/servos/mover': result = self.move(data.get('indice'), data.get('pulso'))
                 elif route == '/servos/parar': result = self.stop()
@@ -138,6 +153,8 @@ class ControleServos:
                 h.enviar(json.dumps({'erro':'Método não permitido'}), codigo=405)
                 return True
             h.enviar(json.dumps(result, ensure_ascii=False))
+        except Conflito as e:
+            h.enviar(json.dumps({'erro':str(e)}, ensure_ascii=False), codigo=409)
         except (ValueError, TypeError) as e:
             h.enviar(json.dumps({'erro':str(e)}), codigo=400)
         except OSError as e:

tests/test_controle_pc.py

@@ -46,7 +46,7 @@ class ControleTest(unittest.TestCase):
         self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
         self.thread.start()
         self.url = 'http://127.0.0.1:%d' % self.server.server_port
-        kiosk.FILA.clear()
+        kiosk.FILA.limpar()
 
     def tearDown(self):
         self.server.shutdown()
@@ -64,12 +64,12 @@ class ControleTest(unittest.TestCase):
             return e.code, e.read()
 
     def test_config_is_persisted_without_consuming_gestures(self):
-        kiosk.FILA.append('feliz')
+        kiosk.FILA.adicionar('feliz')
         code, body = self.request('/app/config', {'volume': 0.7, 'sompausa': 40})
         self.assertEqual(code, 200)
         self.assertEqual(json.loads(body)['config']['volume'], 0.7)
         self.assertEqual(self.request('/app/config')[0], 200)
-        self.assertEqual(kiosk.FILA, ['feliz'])
+        self.assertEqual(kiosk.FILA.ordens(), ['feliz'])
         self.assertEqual(json.loads((Path(self.tmp.name) / 'app.json').read_text())['sompausa'], 40)
 
     def test_invalid_configuration_does_not_partially_apply(self):

tests/test_servidor.py

@@ -0,0 +1,226 @@
+"""Servidor base: token, /eventos (SSE), /estado, /comando, E-Stop, trava de hardware e fila."""
+import http.client
+import importlib.util
+import json
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+spec = importlib.util.spec_from_file_location('kiosk', ROOT / 'kiosk-server.py')
+kiosk = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(kiosk)
+from braco import Braco
+from comum import carregar_token
+from servos import ControleServos
+
+TOKEN = 'segredo-de-teste'
+
+
+class HW:
+    def __init__(self): self.calls = []
+    def pulse(self, ch, us): self.calls.append((ch, us))
+    def off(self, ch): self.calls.append((ch, None))
+
+
+class ServidorTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        pasta = Path(self.tmp.name)
+        self.hw = HW()
+        self.server = kiosk.Servidor(('127.0.0.1', 0), kiosk.Handler)
+        self.server.token = TOKEN
+        self.server.controle = kiosk.ControlePC(pasta / 'app.json')
+        self.server.servos = ControleServos(pasta / 'servos.json', self.hw)
+        self.server.braco = Braco(pasta / 'braco.json', self.server.servos)
+        kiosk.ligar_eventos(self.server)
+        kiosk.FILA.limpar()
+        kiosk.ESTOP.update(ativo=False, desde=None)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.porta = self.server.server_port
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.server.servos.close()
+        self.thread.join()
+        self.tmp.cleanup()
+
+    def pedir(self, metodo, rota, dados=None, token=TOKEN, cabecalhos=None):
+        conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
+        h = {'Content-Type': 'application/json'}
+        if token:
+            h['Authorization'] = 'Bearer ' + token
+        h.update(cabecalhos or {})
+        corpo = None if dados is None else json.dumps(dados)
+        conn.request(metodo, rota, body=corpo, headers=h)
+        r = conn.getresponse()
+        texto = r.read()
+        conn.close()
+        return r.status, json.loads(texto) if texto else None
+
+    # ------------------------------------------------------------- token
+    def test_rotas_de_controle_exigem_token(self):
+        for metodo, rota, dados in [('POST', '/braco/mover', {'angulos': [90] * 7}),
+                                    ('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'}),
+                                    ('POST', '/parar-tudo', {}),
+                                    ('GET', '/estado', None), ('GET', '/fila?aparelho=1', None),
+                                    ('GET', '/app/status', None), ('GET', '/gesto/feliz', None)]:
+            for token in (None, 'errado'):
+                codigo, corpo = self.pedir(metodo, rota, dados, token=token)
+                self.assertEqual(codigo, 401, (metodo, rota, token))
+        self.assertEqual(self.hw.calls, [])
+        self.assertEqual(kiosk.FILA.ordens(), [])
+
+    def test_rotas_publicas_e_token_na_query(self):
+        self.assertEqual(self.pedir('GET', '/gestos', token=None)[0], 200)
+        self.assertEqual(self.pedir('GET', '/app/health', token=None)[0], 200)
+        self.assertEqual(self.pedir('GET', '/estado?t=' + TOKEN, token=None)[0], 200)
+
+    def test_sem_token_configurado_fica_aberto(self):
+        self.server.token = None
+        self.assertEqual(self.pedir('GET', '/estado', token=None)[0], 200)
+
+    def test_carregar_token(self):
+        pasta = Path(self.tmp.name)
+        self.assertIsNone(carregar_token(pasta))
+        (pasta / 'seguranca.json').write_text(json.dumps({'token': 'abcdefgh123'}))
+        self.assertEqual(carregar_token(pasta), 'abcdefgh123')
+        (pasta / 'seguranca.json').write_text(json.dumps({'token': 'curto'}))
+        with self.assertRaises(ValueError):
+            carregar_token(pasta)
+
+    def test_origem_estrangeira_recusada(self):
+        codigo, _ = self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'},
+                               cabecalhos={'Origin': 'http://malicioso.example'})
+        self.assertEqual(codigo, 400)
+
+    # ------------------------------------------------------------- comando e fila
+    def test_comando_valida_e_enfileira(self):
+        self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'})[0], 200)
+        self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'miudo'})[0], 200)
+        self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'rm -rf'})[0], 400)
+        self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'nada'})[0], 400)
+        self.assertEqual(kiosk.FILA.ordens(), ['feliz', 'rosto miudo'])
+        self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], ['feliz', 'rosto miudo'])
+        self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], [])
+
+    def test_post_face_usa_prefixo_rosto(self):
+        conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
+        conn.request('POST', '/face', body='miudo', headers={'Authorization': 'Bearer ' + TOKEN})
+        self.assertEqual(conn.getresponse().status, 200)
+        conn.close()
+        self.assertEqual(kiosk.FILA.ordens(), ['rosto miudo'])
+
+    def test_fila_limitada_e_com_validade(self):
+        fila = kiosk.Fila(maximo=3, validade=10)
+        for i in range(5):
+            fila.adicionar(f'g{i}', agora=0)
+        self.assertEqual(fila.ordens(), ['g2', 'g3', 'g4'])
+        self.assertEqual(fila.consumir(agora=11), [])
+        fila.adicionar('recarregar'); fila.adicionar('recarregar'); fila.adicionar('feliz')
+        self.assertEqual(fila.ordens(), ['recarregar', 'feliz'])
+        # Rosto no navegador não rouba o 'recarregar' do celular.
+        self.assertEqual(fila.consumir(aparelho=False), ['feliz'])
+        self.assertEqual(fila.consumir(aparelho=True), ['recarregar'])
+
+    # ------------------------------------------------------------- SSE
+    def test_eventos_entrega_comando(self):
+        conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
+        conn.request('GET', '/eventos?t=' + TOKEN)
+        r = conn.getresponse()
+        self.assertEqual(r.status, 200)
+        self.assertIn('text/event-stream', r.getheader('Content-Type'))
+        linhas = []
+        while not linhas or linhas[-1] != '':
+            linhas.append(r.fp.readline().decode().rstrip('\n'))
+        self.assertIn('event: pronto', linhas)
+        self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'coracao'})
+        bloco = []
+        while not bloco or bloco[-1] != '':
+            bloco.append(r.fp.readline().decode().rstrip('\n'))
+        conn.close()
+        self.assertIn('event: expressao', bloco)
+        self.assertIn('data: {"ordem": "coracao"}', bloco)
+
+    def test_estado_agrega(self):
+        codigo, estado = self.pedir('GET', '/estado')
+        self.assertEqual(codigo, 200)
+        for chave in ('versao', 'seq', 'estop', 'aparencia', 'gestos', 'app', 'motores'):
+            self.assertIn(chave, estado)
+        self.assertTrue(estado['autenticacao'])
+
+    # ------------------------------------------------------------- E-Stop e trava
+    def test_parar_tudo_desliga_e_trava_ate_rearmar(self):
+        self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
+        self.hw.calls.clear()
+        codigo, corpo = self.pedir('POST', '/parar-tudo', {})
+        self.assertEqual(codigo, 200)
+        self.assertTrue(corpo['estop']['ativo'])
+        self.assertEqual({c for c, us in self.hw.calls if us is None}, set(range(16)))
+        self.assertEqual(kiosk.FILA.ordens(), ['calma'])
+        self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 409)
+        self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 409)
+        self.assertEqual(self.pedir('POST', '/rearmar', {})[0], 200)
+        self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
+
+    def test_braco_e_diagnostico_nao_disputam_motores(self):
+        self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
+        codigo, corpo = self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})
+        self.assertEqual(codigo, 409)
+        self.assertIn('braco', corpo['erro'])
+        self.pedir('POST', '/braco/parar', {})
+        self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 200)
+
+
+class BracoMovimentoTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.hw = HW()
+        self.servos = ControleServos(Path(self.tmp.name) / 's.json', self.hw)
+        self.servos.closed.set()  # sem threads: os passos são chamados à mão
+        self.arm = Braco(Path(self.tmp.name) / 'b.json', self.servos)
+        cal = self.arm.status()
+        for row in cal['calibracao']:
+            row.update(minimo=500, maximo=2500, centro=1500)
+        self.arm.save(dict(calibracao=cal['calibracao'], posturas=[]))
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_interpolacao_no_servidor_respeita_velocidade(self):
+        self.arm.move([90] * 7)
+        resposta = self.arm.move([90, 90, 90, 90, 90, 90, 130], velocidade=100, ttl_ms=2000)
+        self.assertTrue(resposta['movendo'])
+        agora = self.arm.lease - 1.5
+        self.arm.passo(agora, .1)          # 100 °/s * 0,1 s = 10°
+        self.assertEqual(self.arm.atual[6], 100)
+        for i in range(10):
+            self.arm.passo(agora + .1 * (i + 1), .1)
+        self.assertEqual(self.arm.atual[6], 130)
+        self.assertIsNone(self.arm.alvo)
+
+    def test_lease_vencido_interrompe_movimento(self):
+        self.arm.move([90] * 7)
+        self.arm.move([90] * 6 + [170], velocidade=50, ttl_ms=300)
+        self.arm.passo(self.arm.lease + .01, .1)
+        self.assertIsNone(self.arm.alvo)
+        self.assertEqual(self.arm.atual[6], 90)
+        # O watchdog desliga os canais no mesmo prazo.
+        self.servos.expire(self.arm.lease + .01)
+        self.assertEqual(self.servos.active, {})
+        self.assertIsNone(self.servos.dono)
+
+    def test_parametros_invalidos(self):
+        for kwargs in (dict(velocidade=1000), dict(velocidade=True), dict(ttl_ms=50), dict(ttl_ms=5000)):
+            with self.assertRaises(ValueError):
+                self.arm.move([90] * 7, **kwargs)
+        self.assertEqual(self.hw.calls, [])
+
+
+if __name__ == '__main__':
+    unittest.main()

treinamento.py

@@ -1,6 +1,7 @@
 """Rotina, diário e reações procedurais. Não treina pesos nem controla motores."""
 import json,sqlite3,threading,time,math
 from contextlib import contextmanager
+from comum import ler_json
 from gemini import gerar_json
 from figurinhas import criar_desenho,validar_desenho,montar_prompt
 
@@ -28,8 +29,20 @@ def validar_receita(r):
     return r
 
 class Treinamento:
+    # Mudanças de expressão/ocupação viram eventos SSE sem caçar cada atribuição.
+    @property
+    def expressao(self):return self._expressao
+    @expressao.setter
+    def expressao(self,valor):
+        self._expressao=valor;self.publicar('reacao',valor)
+    @property
+    def busy(self):return self._busy
+    @busy.setter
+    def busy(self,valor):
+        self._busy=valor;self.publicar('treinamento',dict(ocupado=valor,erro=self.erro))
     def __init__(self,path,iniciar=True):
-        self.path=str(path);self.lock=threading.RLock();self.busy=False;self.erro='';self.expressao=None
+        self.publicar=lambda tipo,dados:None
+        self.path=str(path);self.lock=threading.RLock();self.erro='';self.busy=False;self.expressao=None
         with self.db() as db:
             db.execute('CREATE TABLE IF NOT EXISTS rotina(id INTEGER PRIMARY KEY,texto TEXT)')
             db.execute('CREATE TABLE IF NOT EXISTS diario(id INTEGER PRIMARY KEY,horario REAL,papel TEXT,texto TEXT)')
@@ -178,12 +191,7 @@ class Treinamento:
             elif method=='GET' and route=='/treinamento/expressao':
                 with self.lock:result=self.expressao
             elif method=='POST':
-                origin=h.headers.get('Origin')
-                if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')):raise ValueError('Origem inválida')
-                size=int(h.headers.get('Content-Length','0'))
-                if not 0<size<=4096 or h.headers.get('Content-Type','').split(';')[0]!='application/json':raise ValueError('Use JSON até 4096 bytes')
-                d=json.loads(h.rfile.read(size))
-                if not isinstance(d,dict):raise ValueError('Objeto obrigatório')
+                d=ler_json(h,4096)
                 with self.lock,self.db() as db:
                     if route=='/treinamento/biblioteca':self.alterar_reacao(d.get('acao'),d.get('id'),d)
                     elif route=='/treinamento/parar':self.expressao=dict(id=time.time_ns(),horario=time.time(),receita=dict(limpar=True))