merge: MAJ silencieuse scaffold
This commit is contained in:
Dom
@@ -103,6 +103,16 @@ LOG_SHIP_INTERVAL_S = float(os.environ.get("RPA_LOG_SHIP_INTERVAL_S", "30"))
|
||||
AUTO_UPDATE_ENABLED = os.environ.get("RPA_AUTO_UPDATE_ENABLED", "false").lower() in (
|
||||
"true", "1", "yes", "on",
|
||||
)
|
||||
# Intervalle entre deux interrogations serveur pour une MAJ (secondes).
|
||||
# Défaut 1 h : une MAJ n'est jamais urgente ; on interroge peu pour ne pas
|
||||
# charger le réseau clinique. Le check ne fait de toute façon aucun swap.
|
||||
AUTO_UPDATE_INTERVAL_S = float(os.environ.get("RPA_AUTO_UPDATE_INTERVAL_S", "3600"))
|
||||
# Dossier de STAGING des ZIP d'update (jamais les fichiers vivants). Équivalent
|
||||
# de `Lea_next\\`. Sous LOCALAPPDATA en prod Windows, sinon à côté de l'agent.
|
||||
AUTO_UPDATE_STAGING_DIR = os.environ.get(
|
||||
"RPA_AUTO_UPDATE_STAGING_DIR",
|
||||
str(BASE_DIR / "_update_staging"),
|
||||
)
|
||||
|
||||
# Monitoring
|
||||
PERF_MONITOR_INTERVAL_S = 30
|
||||
|
||||
@@ -18,6 +18,7 @@ from .config import (
|
||||
SESSIONS_ROOT, AGENT_VERSION, SERVER_URL, MACHINE_ID, LOG_RETENTION_DAYS, LOG_FILE,
|
||||
SCREEN_RESOLUTION, DPI_SCALE, OS_THEME, API_TOKEN, MAX_SESSION_DURATION_S,
|
||||
STREAMING_ENDPOINT, LOG_SHIP_ENABLED, LOG_SHIP_INTERVAL_S,
|
||||
AUTO_UPDATE_ENABLED, AUTO_UPDATE_INTERVAL_S, AUTO_UPDATE_STAGING_DIR,
|
||||
)
|
||||
from .core.captor import EventCaptorV1
|
||||
from .core.executor import ActionExecutorV1
|
||||
@@ -158,6 +159,15 @@ class AgentV1:
|
||||
threading.Thread(target=self._replay_poll_loop, daemon=True).start()
|
||||
threading.Thread(target=self._background_heartbeat_loop, daemon=True).start()
|
||||
|
||||
# DETTE-022 v2 : MAJ silencieuse — boucle de check GATED (défaut OFF).
|
||||
# Interroge le serveur (canary-aware) et télécharge en STAGING ; le swap
|
||||
# réel reste réservé révision humaine (updater.apply_update = stub no-op).
|
||||
# Activable poste par poste via RPA_AUTO_UPDATE_ENABLED, sans rebuild.
|
||||
if AUTO_UPDATE_ENABLED:
|
||||
threading.Thread(
|
||||
target=self._auto_update_loop, daemon=True, name="lea-auto-update"
|
||||
).start()
|
||||
|
||||
# Mini-serveur HTTP pour captures a la demande (port 5006)
|
||||
self._capture_server = CaptureServer()
|
||||
self._capture_server.start()
|
||||
@@ -441,6 +451,67 @@ class AgentV1:
|
||||
logger.debug(f"[HEARTBEAT] Erreur: {e}")
|
||||
time.sleep(5)
|
||||
|
||||
def _auto_update_loop(self):
|
||||
"""DETTE-022 v2 — boucle de MAJ silencieuse GATED (défaut OFF).
|
||||
|
||||
Interroge périodiquement le serveur (endpoint canary-aware), et si une
|
||||
MAJ est proposée pour CE poste, la télécharge dans le STAGING après
|
||||
vérif SHA256. Le swap réel N'EST PAS fait ici : `updater.run_update_cycle`
|
||||
s'arrête au staging (apply_update = stub réservé révision humaine + swap
|
||||
hors-process par Lea.bat au prochain démarrage).
|
||||
|
||||
SÉCURITÉ — « au bon moment » : on NE stage PAS pendant un enregistrement
|
||||
ou un replay actif (self.session_id / self._replay_active), pour ne pas
|
||||
perturber le travail utilisateur ni consommer du réseau au mauvais
|
||||
moment. Best-effort : aucune exception ne remonte (ne casse jamais Léa).
|
||||
"""
|
||||
try:
|
||||
from .network.updater import run_update_cycle
|
||||
except Exception as e:
|
||||
logger.warning("[UPDATE] Module updater indisponible : %s", e)
|
||||
return
|
||||
|
||||
logger.info(
|
||||
"[UPDATE] Boucle MAJ silencieuse démarrée (intervalle=%.0fs, "
|
||||
"version=%s) — check seul, swap réservé révision humaine",
|
||||
AUTO_UPDATE_INTERVAL_S, AGENT_VERSION,
|
||||
)
|
||||
|
||||
while self.running:
|
||||
# Découpe l'attente pour réagir vite à l'arrêt.
|
||||
waited = 0.0
|
||||
step = 1.0
|
||||
while self.running and waited < AUTO_UPDATE_INTERVAL_S:
|
||||
time.sleep(step)
|
||||
waited += step
|
||||
if not self.running:
|
||||
break
|
||||
|
||||
# « Au bon moment » : jamais en plein travail (enregistrement/replay).
|
||||
if self.session_id or getattr(self, "_replay_active", False):
|
||||
logger.debug("[UPDATE] Report du check (session/replay active)")
|
||||
continue
|
||||
|
||||
try:
|
||||
result = run_update_cycle(
|
||||
local_version=AGENT_VERSION,
|
||||
machine_id=self.machine_id,
|
||||
staging_dir=AUTO_UPDATE_STAGING_DIR,
|
||||
)
|
||||
status = result.get("status")
|
||||
if status == "staged":
|
||||
logger.info(
|
||||
"[UPDATE] MAJ %s téléchargée en staging (SHA256=%s) — "
|
||||
"swap réservé révision humaine, non appliqué",
|
||||
result.get("target_version"),
|
||||
result.get("sha256_verified"),
|
||||
)
|
||||
elif status not in ("up_to_date", "disabled"):
|
||||
logger.debug("[UPDATE] Cycle: %s", result)
|
||||
except Exception as e:
|
||||
# run_update_cycle est déjà best-effort ; double filet ici.
|
||||
logger.debug("[UPDATE] Erreur boucle MAJ : %s", e)
|
||||
|
||||
def stop_session(self):
|
||||
# Sauvegarder le session_id avant de l'annuler (pour les logs)
|
||||
ended_session_id = self.session_id
|
||||
|
||||
@@ -238,6 +238,125 @@ def download_update(
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Interrogation serveur — checker INJECTABLE (GET /agents/update/check)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _default_update_checker(local_version: str, machine_id: str):
|
||||
"""Interroge le serveur : y a-t-il une MAJ ? (best-effort, INJECTABLE).
|
||||
|
||||
GET SERVER_URL/agents/update/check?current_version=..&machine_id=..
|
||||
(endpoint gated côté serveur — 503 si RPA_AUTO_UPDATE_SERVER_ENABLED OFF,
|
||||
auquel cas on renvoie None : pas de MAJ). Bearer si présent. Pattern aligné
|
||||
sur `log_shipper._default_sender`. INJECTABLE : remplacé par un fake en test.
|
||||
|
||||
Returns:
|
||||
Le dict réponse serveur (`should_update` sait le lire), ou None si
|
||||
indisponible / gated / erreur (jamais d'exception ne remonte).
|
||||
"""
|
||||
try:
|
||||
import requests # import tardif
|
||||
|
||||
headers = {}
|
||||
try:
|
||||
from ..config import SERVER_URL, API_TOKEN
|
||||
|
||||
base = SERVER_URL
|
||||
if API_TOKEN:
|
||||
headers["Authorization"] = f"Bearer {API_TOKEN}"
|
||||
except Exception:
|
||||
base = ""
|
||||
url = f"{base}/agents/update/check"
|
||||
resp = requests.get(
|
||||
url,
|
||||
params={"current_version": local_version, "machine_id": machine_id},
|
||||
headers=headers,
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
# 503 = endpoint gated OFF côté serveur → pas de MAJ (silencieux).
|
||||
if resp.status_code == 503:
|
||||
return None
|
||||
if not resp.ok:
|
||||
logger.debug("update/check HTTP %s", resp.status_code)
|
||||
return None
|
||||
return resp.json()
|
||||
except Exception as e:
|
||||
logger.debug("update/check indisponible : %s", e)
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Orchestrateur GATED — check → décide → download (staging) → stub apply
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def run_update_cycle(
|
||||
local_version: str,
|
||||
machine_id: str,
|
||||
staging_dir,
|
||||
checker: Optional[Callable[[str, str], object]] = None,
|
||||
downloader: Optional[Callable[[str], bytes]] = None,
|
||||
) -> dict:
|
||||
"""Un cycle complet de MAJ silencieuse — GATED, best-effort, SANS swap.
|
||||
|
||||
Enchaîne :
|
||||
1. GATE `auto_update_enabled()` (RPA_AUTO_UPDATE_ENABLED, défaut OFF) —
|
||||
si OFF, ne fait STRICTEMENT rien (aucun appel réseau).
|
||||
2. `checker(local_version, machine_id)` → réponse serveur (canary-aware).
|
||||
3. `should_update(...)` → plan (double garde semver, jamais de downgrade).
|
||||
4. `download_update(...)` → ZIP dans le STAGING + vérif SHA256. Ne touche
|
||||
JAMAIS les fichiers vivants.
|
||||
5. Le swap réel N'EST PAS FAIT ici : `apply_update` reste un stub no-op
|
||||
(réservé révision humaine + Lea.bat hors-process). Le résultat porte
|
||||
`applied: False`.
|
||||
|
||||
Jamais d'exception ne remonte (ne doit JAMAIS casser Léa). Retourne un dict
|
||||
d'état pour le diagnostic / le log :
|
||||
status ∈ {disabled, check_failed, up_to_date, download_failed, staged}
|
||||
|
||||
Args:
|
||||
checker : callable `(local_version, machine_id) -> dict|None`
|
||||
INJECTABLE (défaut = HTTP réel vers l'endpoint gated).
|
||||
downloader : callable `(url) -> bytes` INJECTABLE (défaut = HTTP réel).
|
||||
"""
|
||||
if not auto_update_enabled():
|
||||
return {"status": "disabled", "applied": False}
|
||||
|
||||
chk = checker if checker is not None else _default_update_checker
|
||||
|
||||
try:
|
||||
server_response = chk(local_version, machine_id)
|
||||
except Exception as e:
|
||||
logger.warning("update check a levé : %s", e)
|
||||
return {"status": "check_failed", "applied": False, "error": str(e)}
|
||||
|
||||
plan = should_update(local_version, server_response)
|
||||
if plan is None:
|
||||
return {"status": "up_to_date", "applied": False}
|
||||
|
||||
staged = download_update(plan, staging_dir, downloader=downloader)
|
||||
if not staged.get("ok"):
|
||||
return {
|
||||
"status": "download_failed",
|
||||
"applied": False,
|
||||
"error": staged.get("error"),
|
||||
}
|
||||
|
||||
# ⚠️ Le swap réel est réservé révision humaine : on APPELLE le stub (no-op)
|
||||
# pour matérialiser le point d'extension, mais rien n'est écrasé/redémarré.
|
||||
applied = apply_update(staged)
|
||||
|
||||
return {
|
||||
"status": "staged",
|
||||
"applied": bool(applied.get("applied", False)),
|
||||
"target_version": staged.get("target_version"),
|
||||
"update_type": staged.get("update_type"),
|
||||
"staged_zip": staged.get("staged_zip"),
|
||||
"sha256_verified": staged.get("sha256_verified", False),
|
||||
"apply_reason": applied.get("reason"),
|
||||
}
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# ⚠️ ZONE DANGEREUSE — STUBS RÉSERVÉS RÉVISION HUMAINE (NE PAS IMPLÉMENTER
|
||||
# PAR UN AGENT). Points d'extension explicites, no-op pour l'instant.
|
||||
|
||||
@@ -7848,6 +7848,9 @@ async def lea_screen_analyze(payload: _Phase25ScreenRequest, request: Request):
|
||||
# client + Lea.bat).
|
||||
# =========================================================================
|
||||
from .update_check import decide_update as _decide_update # noqa: E402
|
||||
from .update_policy import ( # noqa: E402
|
||||
resolve_target_version_from_env as _resolve_target_version_from_env,
|
||||
)
|
||||
|
||||
|
||||
def _auto_update_server_enabled() -> bool:
|
||||
@@ -7857,14 +7860,25 @@ def _auto_update_server_enabled() -> bool:
|
||||
)
|
||||
|
||||
|
||||
def _latest_agent_version() -> str:
|
||||
"""Dernière version d'agent disponible côté serveur.
|
||||
def _latest_agent_version(machine_id: Optional[str] = None) -> str:
|
||||
"""Version d'agent cible POUR CE POSTE (canary-aware, DETTE-022 v2).
|
||||
|
||||
Source de vérité minimale (POC) : variable d'environnement
|
||||
RPA_AGENT_LATEST_VERSION. Permet de piloter la fleet sans rebuild. Une
|
||||
évolution future pourra la lire d'un manifeste/DB (cf. design).
|
||||
⭐ SÉCURITÉ flotte ⭐ — la version servie est résolue PAR MACHINE via la
|
||||
politique canary (`update_policy.resolve_target_version_from_env`) : un
|
||||
poste canary (Émilie `lea-4zbgwxty`) reçoit la nouvelle version en premier ;
|
||||
tous les autres restent sur le floor stable. Piloté 100 % par env, sans
|
||||
rebuild :
|
||||
RPA_AGENT_STABLE_VERSION (défaut 1.0.1) — servi à toute la flotte.
|
||||
RPA_AGENT_CANARY_VERSION — servi AUX SEULS postes canary.
|
||||
RPA_AGENT_CANARY_MACHINES — allow-list CSV des machine_id canary.
|
||||
|
||||
Rétrocompat : si `RPA_AGENT_LATEST_VERSION` (ancienne var globale) est
|
||||
positionnée, elle prime — évite toute régression d'un déploiement existant.
|
||||
"""
|
||||
return os.environ.get("RPA_AGENT_LATEST_VERSION", "1.0.1")
|
||||
legacy = os.environ.get("RPA_AGENT_LATEST_VERSION")
|
||||
if legacy:
|
||||
return legacy
|
||||
return _resolve_target_version_from_env(machine_id)
|
||||
|
||||
|
||||
@app.get("/api/v1/agents/update/check")
|
||||
@@ -7877,6 +7891,10 @@ async def check_agent_update(
|
||||
|
||||
Réponse : {update_available, latest_version, update_type, url}.
|
||||
|
||||
La version cible est résolue PAR MACHINE (canary) : voir
|
||||
`_latest_agent_version`. Un poste hors canary ne se voit JAMAIS proposer la
|
||||
version canary (blast radius borné à la liste canary).
|
||||
|
||||
GATED : si RPA_AUTO_UPDATE_SERVER_ENABLED n'est pas positionné → 503
|
||||
(aucun effet sur le pipeline existant — anti-régression). Auth Bearer
|
||||
requise (dépendance globale `_verify_token`).
|
||||
@@ -7891,7 +7909,7 @@ async def check_agent_update(
|
||||
)
|
||||
return _decide_update(
|
||||
current_version=current_version,
|
||||
latest_version=_latest_agent_version(),
|
||||
latest_version=_latest_agent_version(machine_id),
|
||||
update_type=update_type,
|
||||
machine_id=machine_id,
|
||||
)
|
||||
|
||||
139
agent_v0/server_v1/update_policy.py
Normal file
139
agent_v0/server_v1/update_policy.py
Normal file
@@ -0,0 +1,139 @@
|
||||
# agent_v0/server_v1/update_policy.py
|
||||
"""Politique de déploiement CANARY de la MAJ silencieuse Léa (DETTE-022 v2).
|
||||
|
||||
⭐ Brique de SÉCURITÉ centrale ⭐ — 10+ postes cliniques live (Wallerstein).
|
||||
|
||||
Une MAJ ratée peut briquer toute la flotte. La règle non négociable : on ne
|
||||
pousse JAMAIS une nouvelle version sur tous les postes d'un coup. On la déploie
|
||||
d'abord sur UN poste (canary = Émilie `lea-4zbgwxty`), on vérifie, puis on
|
||||
élargit. Ce module résout, PAR MACHINE, la version cible :
|
||||
|
||||
- poste dans la liste canary → `canary_version` (la nouvelle) ;
|
||||
- tous les autres postes → `stable_version` (le floor, inchangé).
|
||||
|
||||
Piloté 100 % par variables d'environnement (config serveur, sans rebuild) :
|
||||
RPA_AGENT_STABLE_VERSION — version servie à toute la flotte (défaut floor).
|
||||
RPA_AGENT_CANARY_VERSION — version servie AUX SEULS postes canary (optionnel).
|
||||
RPA_AGENT_CANARY_MACHINES — allow-list CSV des machine_id canary.
|
||||
|
||||
Promotion = quand le canary est validé, on met RPA_AGENT_STABLE_VERSION à la
|
||||
version canary (toute la flotte suit) et on vide RPA_AGENT_CANARY_MACHINES.
|
||||
Rollback canary = on remet RPA_AGENT_CANARY_VERSION à l'ancienne / on vide la
|
||||
liste : le prochain check ne proposera plus la MAJ (le swap réel côté client
|
||||
reste réservé révision humaine — cf. updater.py).
|
||||
|
||||
Module PUR (aucun import FastAPI, aucune IO) → importable et testable seul
|
||||
(DETTE-013). Le branchement HTTP vit dans api_stream.py.
|
||||
|
||||
Branche feat/push-log-dgx.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Optional, Set
|
||||
|
||||
# Réutilise le comparateur semver de la décision (même module serveur, pas de
|
||||
# duplication) : "1.0.2" < "1.0.10" correctement, tolérant aux formats invalides.
|
||||
try: # import relatif quand chargé comme package
|
||||
from .update_check import is_newer
|
||||
except Exception: # chargé par chemin (tests importlib) : import du voisin
|
||||
import importlib.util as _ilu
|
||||
from pathlib import Path as _Path
|
||||
|
||||
_uc_path = _Path(__file__).resolve().parent / "update_check.py"
|
||||
_spec = _ilu.spec_from_file_location("_rpa_update_check_for_policy", _uc_path)
|
||||
_uc = _ilu.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(_uc)
|
||||
is_newer = _uc.is_newer
|
||||
|
||||
|
||||
# Séparateurs tolérés dans l'allow-list canary (CSV, espaces, point-virgule).
|
||||
_CANARY_SEPARATORS = (",", ";")
|
||||
|
||||
|
||||
def parse_canary_machines(raw: Optional[str]) -> Set[str]:
|
||||
"""Parse l'allow-list canary en un ensemble de machine_id.
|
||||
|
||||
Tolérant : virgule / point-virgule / espace comme séparateurs, entrées
|
||||
vides ignorées. `None` ou chaîne vide → ensemble vide (aucun canary).
|
||||
"""
|
||||
if not raw or not isinstance(raw, str):
|
||||
return set()
|
||||
normalized = raw
|
||||
for sep in _CANARY_SEPARATORS:
|
||||
normalized = normalized.replace(sep, " ")
|
||||
return {tok for tok in (t.strip() for t in normalized.split()) if tok}
|
||||
|
||||
|
||||
def resolve_target_version(
|
||||
machine_id: Optional[str],
|
||||
stable_version: str,
|
||||
canary_version: Optional[str],
|
||||
canary_machines: Set[str],
|
||||
) -> str:
|
||||
"""Résout la version cible POUR CE POSTE (cœur canary — sécurité).
|
||||
|
||||
Règles (toutes prudentes par défaut) :
|
||||
1. Poste HORS liste canary → `stable_version` (jamais la nouvelle).
|
||||
2. machine_id absent / liste vide / pas de canary_version → `stable_version`.
|
||||
3. Poste DANS la liste canary ET `canary_version` fournie ET STRICTEMENT
|
||||
plus récente que stable → `canary_version`.
|
||||
4. Garde-fou : si `canary_version` <= `stable_version` (config douteuse,
|
||||
ex. downgrade), on sert quand même `stable_version` (jamais de recul).
|
||||
|
||||
Ne lève jamais. Une version illisible retombe naturellement sur le stable
|
||||
via le comparateur semver tolérant.
|
||||
"""
|
||||
# Cas 1/2 : hors canary → stable.
|
||||
if not machine_id or machine_id not in canary_machines:
|
||||
return stable_version
|
||||
if not canary_version:
|
||||
return stable_version
|
||||
|
||||
# Cas 4 : garde-fou anti-recul — le canary doit être STRICTEMENT plus récent.
|
||||
if not is_newer(canary_version, stable_version):
|
||||
return stable_version
|
||||
|
||||
# Cas 3 : poste canary → nouvelle version.
|
||||
return canary_version
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lecture de la politique depuis l'environnement (pilotage sans rebuild).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Défaut historique aligné sur AGENT_VERSION client (config.py) et sur le
|
||||
# fallback de _latest_agent_version().
|
||||
_DEFAULT_STABLE_VERSION = "1.0.1"
|
||||
|
||||
|
||||
def stable_version_from_env() -> str:
|
||||
"""Version servie à toute la flotte (floor). Défaut = 1.0.1."""
|
||||
return os.environ.get("RPA_AGENT_STABLE_VERSION", _DEFAULT_STABLE_VERSION)
|
||||
|
||||
|
||||
def canary_version_from_env() -> Optional[str]:
|
||||
"""Version canary (nouvelle), servie aux seuls postes canary. Optionnel."""
|
||||
val = os.environ.get("RPA_AGENT_CANARY_VERSION", "").strip()
|
||||
return val or None
|
||||
|
||||
|
||||
def canary_machines_from_env() -> Set[str]:
|
||||
"""Allow-list canary (machine_id) depuis RPA_AGENT_CANARY_MACHINES."""
|
||||
return parse_canary_machines(os.environ.get("RPA_AGENT_CANARY_MACHINES", ""))
|
||||
|
||||
|
||||
def resolve_target_version_from_env(machine_id: Optional[str]) -> str:
|
||||
"""Raccourci : résout la version cible pour `machine_id` d'après l'env.
|
||||
|
||||
C'est le point d'entrée que l'endpoint serveur appelle. Il isole toute la
|
||||
lecture d'environnement ici (testable en injectant les paramètres via
|
||||
`resolve_target_version`).
|
||||
"""
|
||||
return resolve_target_version(
|
||||
machine_id=machine_id,
|
||||
stable_version=stable_version_from_env(),
|
||||
canary_version=canary_version_from_env(),
|
||||
canary_machines=canary_machines_from_env(),
|
||||
)
|
||||
Reference in New Issue
Block a user