feat(5.20): US-004/005/006/009 ingestie+API+worker+import pe mediu RAR

US-004: rezolva_rar_env (cerere>default cont>ancora globala) + MediuIndisponibil
+ cod RAR_MEDIU_INDISPONIBIL.
US-005: camp rar_env pe POST /v1/prezentari + /valideaza (Literal), echo in
SubmissionResult/ValidareResult/GET, build_key + INSERT env-aware.
US-006: AccountSessions re-cheiat (account_id, rar_env); RarClient base_url per
env; creds din slotul env; purge + recover_orphans scoped pe env (E1/1a, 1b/E6);
claim_one propaga rar_env (1c/E8); keepalive pe ancora globala (M2).
US-009: selector mediu la import (>=2 medii), eticheta la 1, banner la 0; commit
seteaza rar_env pe submissions.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/api/v1/import_router.py

@@ -55,6 +55,7 @@ from ...mapping import (
     resolve_prestatii,
 )
 from ...validation import validate_prezentare
+from ...rar_env import MediuIndisponibil, rar_env_efectiv_cont, rezolva_rar_env
 
 router = APIRouter(prefix="/v1/import", tags=["import"])
 
@@ -260,10 +261,10 @@ def _resolve_row_for_preview(
     }
 
 
-def _build_idempotency_key(account_id: int | None, resolved: dict[str, Any]) -> str:
+def _build_idempotency_key(account_id: int | None, resolved: dict[str, Any], rar_env: str = "test") -> str:
     """Construieste cheia de idempotenta pentru un rand rezolvat."""
     canon = canonicalize_row(resolved)
-    return build_key(account_id, canon)
+    return build_key(account_id, canon, rar_env)
 
 
 # Campuri de continut editabile in preview. Operatia/codul RAR NU se editeaza
@@ -767,6 +768,11 @@ def preview_import(
         valid_codes = load_nomenclator_codes(conn) or None
         text_rules = load_text_rules(conn, acct)
 
+        # Mediul RAR efectiv al contului — folosit la calculul cheii de idempotenta
+        # la preview (trebuie sa coincida cu ce va folosi commit-ul fara rar_env explicit).
+        from ...config import get_settings as _get_settings_env
+        preview_env = rar_env_efectiv_cont(conn, account_id) or _get_settings_env().rar_env or "test"
+
         # Recalculam coercion_flags din valorile stocate (nu sunt persistate separat):
         # detectie simpla de VIN numeric.
         coercion_flags_map: dict[int, list[str]] = {}
@@ -822,7 +828,7 @@ def preview_import(
             key = None
             if resolved_info["resolved_status"] in ("ok", "needs_review", "needs_data"):
                 try:
-                    key = _build_idempotency_key(account_id, resolved_info["resolved"])
+                    key = _build_idempotency_key(account_id, resolved_info["resolved"], preview_env)
                     keys_for_lookup.append(key)
                     if key not in key_to_index:
                         key_to_index[key] = []
@@ -930,6 +936,7 @@ class CommitIn(BaseModel):
         description="Indecsi de rand needs_review bifate explicit de utilizator",
     )
     confirmed_by: str | None = Field(None, description="Email/identifier utilizator (log atestare)")
+    rar_env: str | None = Field(None, description="Mediu RAR tinta ('test'|'prod'). None = default cont.")
 
 
 @router.post("/{import_id}/commit")
@@ -1024,6 +1031,18 @@ def commit_import(
         if n_total_ok == 0:
             raise HTTPException(status_code=422, detail="Niciun rand ok de confirmat.")
 
+        # Rezolva mediul RAR tinta al lotului (US-009): cerut > default cont > ancora globala.
+        try:
+            env = rezolva_rar_env(conn, account_id, req.rar_env)
+        except ValueError as e:
+            raise HTTPException(status_code=422, detail={"error": "mediu_invalid", "message": str(e)})
+        except MediuIndisponibil as e:
+            raise HTTPException(status_code=422, detail={
+                "error": "mediu_indisponibil",
+                "message": str(e),
+                "disponibile": e.disponibile,
+            })
+
         # T3 (PRD 5.17): enforce volum plan — INAINTE de enqueue (invariant idempotenta).
         # Decizie #21: respingere TOTALA a lotului (nu enqueue partial tacut).
         from ...config import get_settings as _get_settings
@@ -1172,8 +1191,8 @@ def commit_import(
                         "odometru_final": canon["odometru_final"],
                     })
 
-                # Cheia de idempotenta (identica cu cheia din preview — aceeasi ordine)
-                key = build_key(account_id, canon)
+                # Cheia de idempotenta (identica cu cheia din preview — aceeasi ordine + env)
+                key = build_key(account_id, canon, env)
 
                 # Hash row pentru atestare (valori rezolvate)
                 rows_for_hash.append(json.dumps({
@@ -1189,9 +1208,9 @@ def commit_import(
                 # INSERT ON CONFLICT DO NOTHING (TOCTOU)
                 cur = conn.execute(
                     "INSERT OR IGNORE INTO submissions "
-                    "(idempotency_key, account_id, status, payload_json, batch_id, row_index, purge_after) "
-                    "VALUES (?, ?, 'queued', ?, ?, ?, " + purge_after_sql + ")",
-                    (key, acct, payload_json, import_id, row_index),
+                    "(idempotency_key, account_id, status, payload_json, batch_id, row_index, purge_after, rar_env) "
+                    "VALUES (?, ?, 'queued', ?, ?, ?, " + purge_after_sql + ", ?)",
+                    (key, acct, payload_json, import_id, row_index, env),
                 )
 
                 if cur.rowcount == 0:

app/api/v1/router.py

@@ -24,6 +24,7 @@ from ...crypto import encrypt_creds
 from ...db import get_connection
 from ...errors import eroare as err_eroare
 from ...idempotency import build_key, canonicalize_row
+from ...rar_env import MediuIndisponibil, rezolva_rar_env
 from ...mapping import (
     _emite_text_rule_hits,
     account_or_default,
@@ -122,7 +123,7 @@ def _rezultat_enqueue(submission_id: int | None, cl: dict, **extra) -> Submissio
     )
 
 
-def _rezultat_respins(submission_id: int | None, cl: dict) -> SubmissionResult:
+def _rezultat_respins(submission_id: int | None, cl: dict, rar_env: str = "test") -> SubmissionResult:
     """Rezultat pentru on_unmapped_error=True: status='error', fara enqueue/reactivare.
 
     `erori` pastreaza COD_NEMAPAT (compat clienti vechi); `nemapate` + `motiv` adaugate.
@@ -131,6 +132,7 @@ def _rezultat_respins(submission_id: int | None, cl: dict) -> SubmissionResult:
     return SubmissionResult(
         submission_id=submission_id, status="error",
         erori=nem, nemapate=nem, motiv=_motiv_clasificare(cl),
+        rar_env=rar_env,
     )
 
 
@@ -168,6 +170,29 @@ def create_prezentari(
         text_rules = load_text_rules(conn, acct)
         error_mode = _effective_on_unmapped_error(conn, acct, req.on_unmapped_error)
 
+        # US-005: rezolva mediul RAR tinta (cerut > default cont > ancora globala).
+        # MediuIndisponibil -> 422 inainte de orice enqueue (respinge tot lotul).
+        try:
+            env = rezolva_rar_env(conn, acct, req.rar_env)
+        except MediuIndisponibil as e:
+            raise HTTPException(
+                status_code=422,
+                detail=err_eroare(
+                    "RAR_MEDIU_INDISPONIBIL",
+                    cauza=(
+                        f"mediu cerut: {e.env}; disponibile: "
+                        f"{', '.join(e.disponibile) or 'niciunul'}"
+                    ),
+                ),
+            )
+        except ValueError:
+            # Pydantic Literal prinde valorile invalide inainte sa ajunga aici;
+            # ramura e defensiva pentru apeluri directe fara model Pydantic.
+            raise HTTPException(
+                status_code=422,
+                detail=err_eroare("RAR_MEDIU_INDISPONIBIL", cauza="valoare invalida pentru rar_env"),
+            )
+
         # T3 (PRD 5.17): enforce volum plan — INAINTE de build_key/enqueue (invariant idempotenta).
         # Decizie #21: respingere TOTALA a lotului (nu enqueue partial tacut).
         from ...config import get_settings as _get_settings
@@ -213,7 +238,7 @@ def create_prezentari(
             # build_key aplica account_or_default(account_id) inainte de hash:
             # None si 1 colapseaza la aceeasi cheie (canal API + canal import).
             canon = canonicalize_row(content)
-            key = build_key(account_id, canon)
+            key = build_key(account_id, canon, env)
             # Aplica normalizarea si in content (odometru canonicalizat inainte de validare)
             content.update({
                 "vin": canon["vin"],
@@ -232,19 +257,20 @@ def create_prezentari(
                     cl = _classify_modal(content, mapping, mapping_meta, valid_codes, error_mode, text_rules)
                     if cl["blocked_error"]:
                         # on_unmapped_error=True: nu reactivam; randul ramane 'error'.
-                        results.append(_rezultat_respins(existing["id"], cl))
+                        results.append(_rezultat_respins(existing["id"], cl, rar_env=env))
                         continue
                     cur = conn.execute(
                         "UPDATE submissions SET status=?, payload_json=?, rar_error=?, "
                         "rar_creds_enc=COALESCE(?, rar_creds_enc), retry_count=0, "
                         "next_attempt_at=NULL, sending_since=NULL, purge_after=NULL, "
-                        "updated_at=datetime('now') WHERE id=? AND status='error'",
+                        "rar_env=?, updated_at=datetime('now') WHERE id=? AND status='error'",
                         (cl["status"], json.dumps(cl["content"], ensure_ascii=False),
-                         cl["rar_error"], creds_enc, existing["id"]),
+                         cl["rar_error"], creds_enc, env, existing["id"]),
                     )
                     if cur.rowcount == 1:
                         # Creds noi se propaga si in canalul durabil (accounts.rar_creds_enc)
                         # — ambele canale converg pe parola corectata.
+                        # US-013: muta pe slot env dupa login (write-back conservator).
                         if req.rar_credentials is not None:
                             conn.execute(
                                 "UPDATE accounts SET rar_creds_enc=? WHERE id=?",
@@ -253,7 +279,7 @@ def create_prezentari(
                         _emite_text_rule_hits(conn, acct, existing["id"], cl["resolved"])
                         # Raspuns onest si la reactivare: daca re-clasificarea cade pe
                         # needs_data/needs_mapping, expune motivul (nu doar status).
-                        results.append(_rezultat_enqueue(existing["id"], cl, reactivated=True))
+                        results.append(_rezultat_enqueue(existing["id"], cl, reactivated=True, rar_env=env))
                         continue
                     # Cursa: alt POST/requeue a schimbat starea intre SELECT si UPDATE
                     # (rowcount==0) -> raspuns dedup pe starea CURENTA.
@@ -267,6 +293,7 @@ def create_prezentari(
                         status=existing["status"],
                         id_prezentare=existing["id_prezentare"],
                         deduped=True,
+                        rar_env=env,
                     )
                 )
                 continue
@@ -276,17 +303,17 @@ def create_prezentari(
             cl = _classify_modal(content, mapping, mapping_meta, valid_codes, error_mode, text_rules)
             if cl["blocked_error"]:
                 # on_unmapped_error=True: respinge fara enqueue (cod necunoscut/nemapat).
-                results.append(_rezultat_respins(None, cl))
+                results.append(_rezultat_respins(None, cl, rar_env=env))
                 continue
             cur = conn.execute(
-                "INSERT INTO submissions (idempotency_key, account_id, status, payload_json, rar_error, rar_creds_enc) "
-                "VALUES (?, ?, ?, ?, ?, ?)",
-                (key, acct, cl["status"], json.dumps(cl["content"], ensure_ascii=False), cl["rar_error"], creds_enc),
+                "INSERT INTO submissions (idempotency_key, account_id, status, payload_json, rar_error, rar_creds_enc, rar_env) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (key, acct, cl["status"], json.dumps(cl["content"], ensure_ascii=False), cl["rar_error"], creds_enc, env),
             )
             sub_id = int(cur.lastrowid)
             _emite_text_rule_hits(conn, acct, sub_id, cl["resolved"])
             # Raspuns onest: pe needs_data/needs_mapping expune erori/nemapate/motiv.
-            results.append(_rezultat_enqueue(sub_id, cl))
+            results.append(_rezultat_enqueue(sub_id, cl, rar_env=env))
 
         # Audit cerere API per cont. Doar metadate (count + distributie status),
         # NICIUN camp de payload PII integral. Reuse conn (fara contentie WAL).
@@ -332,6 +359,27 @@ def valideaza_prezentari(
         # Acelasi seam ca trimiterea reala: dry-run trebuie sa vada aceleasi reguli text.
         text_rules = load_text_rules(conn, acct)
         error_mode = _effective_on_unmapped_error(conn, acct, req.on_unmapped_error)
+
+        # US-005 (DX F5): rezolva env identic ca trimiterea reala si ecou-ieste in raspuns.
+        try:
+            env = rezolva_rar_env(conn, acct, req.rar_env)
+        except MediuIndisponibil as e:
+            raise HTTPException(
+                status_code=422,
+                detail=err_eroare(
+                    "RAR_MEDIU_INDISPONIBIL",
+                    cauza=(
+                        f"mediu cerut: {e.env}; disponibile: "
+                        f"{', '.join(e.disponibile) or 'niciunul'}"
+                    ),
+                ),
+            )
+        except ValueError:
+            raise HTTPException(
+                status_code=422,
+                detail=err_eroare("RAR_MEDIU_INDISPONIBIL", cauza="valoare invalida pentru rar_env"),
+            )
+
         for i, prez in enumerate(req.prezentari):
             content = prez.model_dump()
             res = _classify_modal(content, mapping, mapping_meta, valid_codes, error_mode, text_rules)
@@ -346,6 +394,7 @@ def valideaza_prezentari(
                 index=i,
                 valid=(res["status"] == "queued"),
                 status_estimat=res["status"],
+                rar_env=env,
                 erori=res["errors"],
                 nemapate=nemapate,
                 prestatii_rezolvate=res["resolved"],
@@ -366,9 +415,10 @@ def list_prezentari(
         scope_sql, scope_params = account_scope_clause(account_id)
         # payload_json e plaintext (vezi submissions.payload_json); il citim doar ca
         # sa derivam campurile afisabile prin helper-ul partajat, nu il expunem.
+        # rar_env inclus (US-005): badge mediu in lista.
         cols = (
             "id, status, id_prezentare, rar_status_code, retry_count, "
-            "created_at, updated_at, payload_json"
+            "created_at, updated_at, payload_json, rar_env"
         )
         if status:
             rows = conn.execute(
@@ -403,6 +453,8 @@ _PREZENTARE_FIELDS = frozenset({
     # erori din catalog (niciodata creds, ex. RAR_CREDS_INVALIDE poarta doar cauza
     # "credentiale RAR invalide", fara parola). Face recovery-ul observabil prin API.
     "rar_error",
+    # US-005: mediul RAR tinta (Test/Productie) — necesar pentru badge + ecou API.
+    "rar_env",
 })
 
 

app/errors.py

@@ -194,6 +194,13 @@ CATALOG: dict[str, dict[str, str]] = {
             " Contacteaza-ne pentru a face upgrade la planul Pro."
         ),
     },
+    "RAR_MEDIU_INDISPONIBIL": {
+        "problema": "Mediul RAR cerut nu e disponibil",
+        "fix": (
+            "Activeaza mediul si introdu credentialele RAR in tab-ul Cont."
+            " Mediile disponibile acum sunt in campul cauza."
+        ),
+    },
 }
 
 

app/models.py

@@ -7,6 +7,8 @@ odometru) este in app.validation.
 
 from __future__ import annotations
 
+from typing import Literal
+
 from pydantic import BaseModel, Field, field_validator, model_validator
 
 
@@ -93,6 +95,8 @@ class PrezentareRequest(BaseModel):
     #   False -> submission 'needs_mapping' (intra in editorul de mapare);
     #   None  -> se foloseste accounts.on_unmapped_error_default (implicit False).
     on_unmapped_error: bool | None = None
+    # Mediul RAR tinta: 'test' | 'prod'. Absent -> default-ul contului (REQ-DEFAULT).
+    rar_env: Literal["test", "prod"] | None = None
 
 
 class SubmissionResult(BaseModel):
@@ -105,6 +109,8 @@ class SubmissionResult(BaseModel):
     # RE-ACTIVAT (re-clasificat + creds actualizate) la resubmit. `deduped` pastreaza
     # semantica actuala (clientii vechi care testeaza `deduped` nu se sparg).
     reactivated: bool = False
+    # Mediul RAR tinta efectiv (ecou din DB / rezolvat la ingestie).
+    rar_env: str = "test"
     # Raspuns ONEST pentru randuri blocate: orice status != 'queued' isi expune
     # motivul, ca integratorul sa nu trateze un needs_data/needs_mapping drept succes.
     #   erori    = validare de continut (needs_data), 3 niveluri [{field, cod, problema, cauza, fix, message}].
@@ -126,6 +132,8 @@ class ValidarePrezentariRequest(BaseModel):
     rar_credentials: RarCredentials | None = None
     prezentari: list[PrezentareIn] = Field(..., min_length=1)
     on_unmapped_error: bool | None = None
+    # Mediul RAR tinta: 'test' | 'prod'. Absent -> default-ul contului.
+    rar_env: Literal["test", "prod"] | None = None
 
 
 class ValidareResult(BaseModel):
@@ -134,6 +142,7 @@ class ValidareResult(BaseModel):
     index: int
     valid: bool
     status_estimat: str  # "queued" | "needs_data" | "needs_mapping"
+    rar_env: str = "test"  # mediul RAR tinta efectiv (ecou din rezolvare)
     erori: list[dict] = []
     nemapate: list[dict] = []
     prestatii_rezolvate: list[dict] = []

app/rar_client.py

@@ -44,6 +44,11 @@ class RarAuthError(RarError):
     """Login esuat (401 / credentiale invalide). NU se face retry."""
 
 
+def base_url_pentru_env(settings: "Settings", env: str) -> str:
+    """URL de baza al mediului RAR: 'prod' -> rar_base_url_prod, altfel rar_base_url_test."""
+    return settings.rar_base_url_prod if env == "prod" else settings.rar_base_url_test
+
+
 class RarClient:
     """Client sincron httpx. Folosit din worker (proces separat).
 
@@ -53,10 +58,10 @@ class RarClient:
             data = rar.post_prezentare(token, payload)
     """
 
-    def __init__(self, settings: Settings | None = None):
+    def __init__(self, settings: Settings | None = None, *, base_url: str | None = None):
         self.settings = settings or get_settings()
         self._client = httpx.Client(
-            base_url=self.settings.rar_base_url,
+            base_url=base_url if base_url is not None else self.settings.rar_base_url,
             timeout=self.settings.http_timeout_s,
             headers={"User-Agent": self.settings.http_user_agent},  # fix WAF 403
         )

app/rar_env.py

@@ -89,3 +89,68 @@ def medii_disponibile_cont(conn: sqlite3.Connection, account_id: int) -> list[st
 
 def rar_env_efectiv_cont(conn: sqlite3.Connection, account_id: int) -> str | None:
     return rar_env_efectiv(load_account_env(conn, account_id))
+
+
+# --------------------------------------------------------------------------- #
+# Exceptie si rezolvator de mediu tinta (US-004, dependent de US-002)          #
+# --------------------------------------------------------------------------- #
+
+class MediuIndisponibil(Exception):
+    """Mediul RAR cerut e valid dar nu e disponibil pentru contul dat.
+
+    Atribute
+    --------
+    env:        mediul cerut (ex. 'test')
+    disponibile: lista mediilor disponibile pentru cont in momentul erorii
+    """
+
+    def __init__(self, env: str, disponibile: list[str]) -> None:
+        self.env = env
+        self.disponibile = disponibile
+        super().__init__(
+            f"mediu indisponibil: {env!r} (disponibile: {disponibile!r})"
+        )
+
+
+def rezolva_rar_env(
+    conn: sqlite3.Connection,
+    account_id: int,
+    cerut: str | None = None,
+) -> str:
+    """Determina mediul RAR tinta pentru un submission la ingestie.
+
+    Precedenta stricta (de la cea mai mare la cea mai mica):
+      1. `cerut` explicit si disponibil -> intoarce `cerut`.
+      2. `cerut` explicit dar indisponibil -> ridica MediuIndisponibil.
+      3. `cerut` invalid (nu in VALID_ENVS) -> ridica ValueError (fara fallback silentios).
+      4. `cerut` None -> incearca rar_env_efectiv_cont (default-ul contului).
+      5. Daca contul nu are niciun mediu disponibil (rar_env_efectiv_cont == None)
+         -> cade pe ancora globala get_settings().rar_env, normalizata la VALID_ENVS.
+         Acest fallback e intentionat (PRD 5.20 §2 Non-Goals): AUTOPASS_RAR_ENV ramane
+         ancora de migrare si fallback pentru actiuni fara cont (keepalive, canal API cu
+         creds efemere pe conturi nou-create fara medii configurate).
+
+    Ridica
+    ------
+    ValueError        -- `cerut` nu e in VALID_ENVS
+    MediuIndisponibil -- `cerut` e valid dar nu e disponibil pentru cont
+    """
+    if cerut is not None:
+        if cerut not in VALID_ENVS:
+            raise ValueError(f"mediu invalid: {cerut!r}")
+        disp = medii_disponibile_cont(conn, account_id)
+        if cerut not in disp:
+            raise MediuIndisponibil(cerut, disp)
+        return cerut
+
+    # cerut e None: incearca default-ul contului
+    efectiv = rar_env_efectiv_cont(conn, account_id)
+    if efectiv is not None:
+        return efectiv
+
+    # Ancora globala: 0 medii disponibile pe cont -> fallback la AUTOPASS_RAR_ENV.
+    from .config import get_settings
+    global_env = get_settings().rar_env
+    if global_env in VALID_ENVS:
+        return global_env
+    return "test"  # rar_env invalid in config -> cel mai sigur default

app/web/routes.py

@@ -89,10 +89,24 @@ from ..mapping import (
     text_rules_overlap,
 )
 from ..shared_store import record_human_validation
+from ..rar_env import MediuIndisponibil, medii_disponibile_cont, rar_env_efectiv_cont, rezolva_rar_env
 
 # Campuri canonice cu eticheta umana pentru dropdown mapare coloane
 _CANONICAL_FIELDS = [(k, v[0]) for k, v in _CANONICAL_SYNONYMS.items()]
 
+
+def _import_env_ctx(conn, account_id: int) -> dict:
+    """Contextul de mediu RAR pentru paginile de import (US-009, PRD 5.20).
+
+    Intoarce {'medii': list[str], 'env_default': str} pentru template-ul _upload.html.
+    Un mediu e disponibil = activat SI are credentiale. La 0 medii template afiseaza
+    un banner non-blocant; la 1 eticheta statica; la >=2 selector.
+    """
+    medii = medii_disponibile_cont(conn, account_id)
+    env_default = rar_env_efectiv_cont(conn, account_id) or "prod"
+    return {"medii": medii, "env_default": env_default}
+
+
 router = APIRouter(tags=["web"])
 templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent / "templates"))
 # Expune parse_erori in toate template-urile
@@ -738,8 +752,13 @@ def fragment_acasa(request: Request) -> HTMLResponse:
 @router.get("/_fragments/import", response_class=HTMLResponse)
 def fragment_import(request: Request) -> HTMLResponse:
     """Fragment HTMX pentru tab-ul Import — include zona de upload."""
-    require_login(request)
-    return templates.TemplateResponse("_upload.html", _ctx(request))
+    account_id = require_login(request)
+    conn = get_connection()
+    try:
+        env_ctx = _import_env_ctx(conn, account_id)
+    finally:
+        conn.close()
+    return templates.TemplateResponse("_upload.html", _ctx(request, **env_ctx))
 
 
 @router.get("/_fragments/coada", response_class=HTMLResponse)
@@ -2684,13 +2703,20 @@ def _web_compute_preview(
     conn,
     import_id: int,
     account_id: int,
+    rar_env: str | None = None,
 ) -> dict[str, Any] | str:
     """Calculeaza preview pentru un batch; intoarce date sau str cu mesaj de eroare.
 
     Reutilizeaza _resolve_row_for_preview, _already_sent_lookup, _signature
     din import_router. Nu repeta logica de rezolvare — only orchestrare.
+
+    `rar_env`: mediul RAR ales de operator; None = default efectiv al contului
+    (sau ancora globala daca contul nu are medii configurate). Folosit la calculul
+    cheii de idempotenta la preview — trebuie sa coincida cu env-ul de la commit.
     """
     acct = account_or_default(account_id)
+    # Mediul folosit la calculul cheii de idempotenta (preview == commit).
+    preview_env = rar_env or rar_env_efectiv_cont(conn, account_id) or get_settings().rar_env or "test"
 
     batch = conn.execute(
         "SELECT id, account_id, filename FROM import_batches WHERE id=? AND account_id=?",
@@ -2796,7 +2822,7 @@ def _web_compute_preview(
         key: str | None = None
         if info["resolved_status"] in ("ok", "needs_review", "needs_data"):
             try:
-                key = _build_idempotency_key(account_id, info["resolved"])
+                key = _build_idempotency_key(account_id, info["resolved"], preview_env)
                 keys_for_lookup.append(key)
                 key_to_indices.setdefault(key, []).append(i)
             except Exception:
@@ -2901,6 +2927,7 @@ async def web_upload_import(
     file: UploadFile = File(...),
     sheet_name: str | None = Form(None),
     csrf_token: str | None = Form(None),
+    rar_env: str | None = Form(None),
 ) -> HTMLResponse:
     """Upload fisier xlsx/csv → staging; intoarce fragment HTML.
 
@@ -2919,31 +2946,64 @@ async def web_upload_import(
     try:
         parsed = parse_file(data, filename, sheet_name=sheet_name)
     except MultipleSheets as ms:
-        return templates.TemplateResponse("_upload.html", _ctx(request, sheets=ms.sheet_names))
+        conn = get_connection()
+        try:
+            env_ctx = _import_env_ctx(conn, account_id)
+        finally:
+            conn.close()
+        return templates.TemplateResponse("_upload.html", _ctx(request, sheets=ms.sheet_names, **env_ctx))
     except FileTooLarge as e:
         eroare_upload = _errors.eroare("IMPORT_FISIER_PREA_MARE", cauza=str(e))
+        conn = get_connection()
+        try:
+            env_ctx = _import_env_ctx(conn, account_id)
+        finally:
+            conn.close()
         return templates.TemplateResponse("_upload.html", _ctx(
-            request, error=str(e), eroare_upload=eroare_upload
+            request, error=str(e), eroare_upload=eroare_upload, **env_ctx
         ))
     except HeaderError as e:
         eroare_upload = _errors.eroare("IMPORT_ANTET_NECLAR", cauza=f"Antet neclar: {e}")
+        conn = get_connection()
+        try:
+            env_ctx = _import_env_ctx(conn, account_id)
+        finally:
+            conn.close()
         return templates.TemplateResponse("_upload.html", _ctx(
-            request, error=f"Antet neclar: {e}", eroare_upload=eroare_upload
+            request, error=f"Antet neclar: {e}", eroare_upload=eroare_upload, **env_ctx
         ))
     except UnicodeDecodeError as e:
         eroare_upload = _errors.eroare("IMPORT_ENCODING", cauza=f"Encoding nesuportat: {e.reason}")
+        conn = get_connection()
+        try:
+            env_ctx = _import_env_ctx(conn, account_id)
+        finally:
+            conn.close()
         return templates.TemplateResponse("_upload.html", _ctx(
-            request, error=f"Encoding nesuportat: {e.reason}", eroare_upload=eroare_upload
+            request, error=f"Encoding nesuportat: {e.reason}", eroare_upload=eroare_upload, **env_ctx
         ))
     except Exception as e:
         eroare_upload = _errors.eroare("IMPORT_FISIER_NERECUNOSCUT", cauza=f"Fisier nerecunoscut (xlsx/csv): {type(e).__name__}")
+        conn = get_connection()
+        try:
+            env_ctx = _import_env_ctx(conn, account_id)
+        finally:
+            conn.close()
         return templates.TemplateResponse("_upload.html", _ctx(
-            request, error=f"Fisier nerecunoscut (xlsx/csv): {type(e).__name__}", eroare_upload=eroare_upload
+            request, error=f"Fisier nerecunoscut (xlsx/csv): {type(e).__name__}", eroare_upload=eroare_upload, **env_ctx
         ))
 
     conn = get_connection()
     try:
         sig = _signature(parsed.columns)
+        env_ctx = _import_env_ctx(conn, account_id)
+
+        # Rezolva mediul RAR ales — cerut din form sau default cont (fallback ancora globala).
+        # La 0 medii: rezolva_rar_env cade pe ancora globala (rar_env config), non-blocant.
+        try:
+            upload_env = rezolva_rar_env(conn, account_id, rar_env or None)
+        except (ValueError, MediuIndisponibil):
+            upload_env = rar_env_efectiv_cont(conn, account_id) or get_settings().rar_env or "test"
 
         # Stagingul in DB (tranzactie explicita)
         conn.execute("BEGIN IMMEDIATE")
@@ -2978,18 +3038,17 @@ async def web_upload_import(
 
         if existing:
             # Mapare retinuta → computa preview imediat
-            result = _web_compute_preview(conn, batch_id_int, account_id)
+            result = _web_compute_preview(conn, batch_id_int, account_id, rar_env=upload_env)
             if isinstance(result, str):
-                return templates.TemplateResponse("_upload.html", {
-                    "request": request,
-                    "error": result,
-                    "csrf_token": get_csrf_token(request),
-                })
+                return templates.TemplateResponse("_upload.html", _ctx(
+                    request, error=result, **env_ctx
+                ))
             return templates.TemplateResponse("_preview_import.html", {
                 "request": request,
                 "import_id": batch_id_int,
                 "message": "Mapare retinuta aplicata automat.",
                 "csrf_token": get_csrf_token(request),
+                "rar_env": upload_env,
                 **result,
             })
 
@@ -3012,6 +3071,7 @@ async def web_upload_import(
             "canonical_fields": _CANONICAL_FIELDS,
             "format_data": None,
             "csrf_token": get_csrf_token(request),
+            "rar_env": upload_env,
         })
     finally:
         conn.close()
@@ -3129,8 +3189,9 @@ async def web_save_mapare_coloane(
             (import_id, acct),
         ).fetchone()
         if not batch:
+            env_ctx = _import_env_ctx(conn, account_id)
             return templates.TemplateResponse("_upload.html", _ctx(
-                request, error="Batch de import inexistent sau expirat."
+                request, error="Batch de import inexistent sau expirat.", **env_ctx
             ))
 
         # Semnatura = antetul COMPLET al fisierului (toate coloanele, inclusiv cele
@@ -3148,12 +3209,20 @@ async def web_save_mapare_coloane(
             (acct, sig, json.dumps(json_mapare, ensure_ascii=False), format_data_val),
         )
 
+        # Mediu RAR transmis din form-ul de mapare (daca exista) sau default cont
+        form_rar_env = str(form.get("rar_env") or "").strip() or None
+        try:
+            mapare_env = rezolva_rar_env(conn, account_id, form_rar_env)
+        except (ValueError, MediuIndisponibil):
+            mapare_env = rar_env_efectiv_cont(conn, account_id) or get_settings().rar_env or "test"
+
         # Computa preview
-        result = _web_compute_preview(conn, import_id, account_id)
+        result = _web_compute_preview(conn, import_id, account_id, rar_env=mapare_env)
         if isinstance(result, str):
-            return templates.TemplateResponse("_upload.html", _ctx(request, error=result))
+            env_ctx = _import_env_ctx(conn, account_id)
+            return templates.TemplateResponse("_upload.html", _ctx(request, error=result, **env_ctx))
         return templates.TemplateResponse("_preview_import.html", _ctx(
-            request, import_id=import_id, **result
+            request, import_id=import_id, rar_env=mapare_env, **result
         ))
     finally:
         conn.close()
@@ -3163,22 +3232,31 @@ async def web_save_mapare_coloane(
 def web_preview_import(
     request: Request,
     import_id: int,
+    rar_env: str | None = None,
 ) -> HTMLResponse:
     """Preview 6 stari per rand. Tinta HTMX dupa mapare retinuta sau navigare directa."""
     account_id = require_login(request)
     conn = get_connection()
     try:
-        result = _web_compute_preview(conn, import_id, account_id)
+        # Rezolva mediul pentru preview (din query param sau default cont)
+        try:
+            preview_env = rezolva_rar_env(conn, account_id, rar_env)
+        except (ValueError, MediuIndisponibil):
+            preview_env = rar_env_efectiv_cont(conn, account_id) or get_settings().rar_env or "test"
+        result = _web_compute_preview(conn, import_id, account_id, rar_env=preview_env)
         if isinstance(result, str):
+            env_ctx = _import_env_ctx(conn, account_id)
             return templates.TemplateResponse("_upload.html", {
                 "request": request,
                 "error": result,
                 "csrf_token": get_csrf_token(request),
+                **env_ctx,
             })
         return templates.TemplateResponse("_preview_import.html", {
             "request": request,
             "import_id": import_id,
             "csrf_token": get_csrf_token(request),
+            "rar_env": preview_env,
             **result,
         })
     finally:
@@ -3601,10 +3679,13 @@ async def web_mapare_operatii(
 @router.get("/_import/reset", response_class=HTMLResponse)
 def web_import_reset(request: Request) -> HTMLResponse:
     """Reseteaza sectiunea de import la starea initiala (drop zone gol)."""
-    return templates.TemplateResponse("_upload.html", {
-        "request": request,
-        "csrf_token": get_csrf_token(request),
-    })
+    account_id = require_login(request)
+    conn = get_connection()
+    try:
+        env_ctx = _import_env_ctx(conn, account_id)
+    finally:
+        conn.close()
+    return templates.TemplateResponse("_upload.html", _ctx(request, **env_ctx))
 
 
 @router.post("/_import/{import_id}/confirma", response_class=HTMLResponse)
@@ -3631,6 +3712,9 @@ async def web_confirma_import(
     except (ValueError, TypeError):
         n_confirmat = 0
 
+    # Mediu RAR din form (selectat in preview); None = default cont (fallback ancora globala)
+    rar_env_cerut = str(form.get("rar_env") or "").strip() or None
+
     # US-007: reviewed_rows (checkboxe vechi) NU mai este sursa de adevar pentru gate-ul
     # de commit pe canalul web. Gate-ul este derivat din DB import_rows.reviewed (D#8).
     # Randurile needs_review confirmate de operator via /confirma-review au resolved_status='ok'
@@ -3798,6 +3882,12 @@ async def web_confirma_import(
         valid_codes = load_nomenclator_codes(conn) or None
         text_rules = load_text_rules(conn, acct)
 
+        # Rezolva mediul RAR tinta al lotului (US-009): form > default cont > ancora globala.
+        try:
+            env = rezolva_rar_env(conn, account_id, rar_env_cerut)
+        except (ValueError, MediuIndisponibil):
+            env = rar_env_efectiv_cont(conn, account_id) or get_settings().rar_env or "test"
+
         # Enqueue in tranzactie explicita — INSERT ON CONFLICT DO NOTHING (TOCTOU)
         enqueued: list[dict] = []
         toctou: list[int] = []
@@ -3857,7 +3947,7 @@ async def web_confirma_import(
                         "odometru_final": canon["odometru_final"],
                     })
 
-                key = build_key(account_id, canon)
+                key = build_key(account_id, canon, env)
 
                 rows_for_hash.append(json.dumps({
                     "row_index": row_index,
@@ -3872,9 +3962,9 @@ async def web_confirma_import(
 
                 cur = conn.execute(
                     "INSERT OR IGNORE INTO submissions "
-                    "(idempotency_key, account_id, status, payload_json, batch_id, row_index, purge_after) "
-                    "VALUES (?, ?, 'queued', ?, ?, ?, datetime('now', '+90 days'))",
-                    (key, acct, json.dumps(mapped, ensure_ascii=False), import_id, row_index),
+                    "(idempotency_key, account_id, status, payload_json, batch_id, row_index, purge_after, rar_env) "
+                    "VALUES (?, ?, 'queued', ?, ?, ?, datetime('now', '+90 days'), ?)",
+                    (key, acct, json.dumps(mapped, ensure_ascii=False), import_id, row_index, env),
                 )
                 if cur.rowcount == 0:
                     toctou.append(row_index)
@@ -3922,8 +4012,9 @@ async def web_confirma_import(
         status_ctx = _build_status_ctx(request, conn, account_id, oob=True)
 
         # Randeaza imediat (conn inca deschis — query-urile s-au facut mai sus).
+        env_ctx_success = _import_env_ctx(conn, account_id)
         upload_html = templates.get_template("_upload.html").render(
-            _ctx(request, are_trimiteri=True, message=succes_msg)
+            _ctx(request, are_trimiteri=True, message=succes_msg, **env_ctx_success)
         )
         coada_html = templates.get_template("_coada.html").render(acasa_ctx)
         status_html = templates.get_template("_status.html").render(status_ctx)

app/web/templates/_preview_import.html

@@ -14,6 +14,14 @@
         Preview —
         <span class="muted" style="font-weight:400;">{{ filename or ("import #" ~ import_id) }}</span>
       </h2>
+      {# Badge mediu RAR (US-009): vizibil intotdeauna in preview (claritate tinta) #}
+      {% if rar_env %}
+      <span class="pill" style="font-size:var(--fs-xs);
+            {% if rar_env == 'prod' %}background:color-mix(in srgb,#B4452F 15%,var(--card)); border-color:#B4452F; color:#B4452F; font-weight:700;
+            {% else %}background:color-mix(in srgb,var(--accent) 12%,var(--card)); border-color:var(--accent); color:var(--accent);{% endif %}">
+        {{ "PRODUCTIE" if rar_env == "prod" else "Testare" }}
+      </span>
+      {% endif %}
       <span class="muted" style="margin-left:auto; font-size:var(--fs-sm);">{{ total }} randuri</span>
     </div>
 
@@ -178,6 +186,8 @@
           hx-target="#import-section"
           hx-swap="outerHTML">
       <input type="hidden" name="csrf_token" value="{{ csrf_token or '' }}">
+      {# Mediu RAR ales la upload — propagar la commit (US-009) #}
+      <input type="hidden" name="rar_env" value="{{ rar_env or '' }}">
       <div class="sticky-bar">
         <div style="flex:1; min-width:280px;">
           <!-- Banner declarant — direct deasupra input-ului N -->

app/web/templates/_upload.html

@@ -32,6 +32,38 @@
           hx-indicator="#upload-spinner">
       <input type="hidden" name="csrf_token" value="{{ csrf_token or '' }}">
 
+      {# Indicator mediu RAR (US-009, PRD 5.20): vizibil inainte de drop-zone #}
+      {% set medii_rar = medii | default([]) %}
+      {% if medii_rar | length == 0 %}
+      {# Banner avertisment — non-blocant (upload continua; commit foloseste ancora globala) #}
+      <div style="margin-bottom:10px; padding:8px 14px; border-radius:6px;
+                  background:color-mix(in srgb, var(--warn, #e6b34a) 12%, var(--card));
+                  border:1px solid var(--warn, #e6b34a); font-size:13px;" role="note">
+        <strong>Niciun mediu RAR configurat.</strong>
+        Trimiterea va folosi configuratia globala. Pentru a activa Testare sau Productie,
+        <a href="?tab=cont" style="color:var(--accent);">configureaza credentialele RAR</a>.
+      </div>
+      {% elif medii_rar | length == 1 %}
+      {# Eticheta statica (un singur mediu disponibil) #}
+      <input type="hidden" name="rar_env" value="{{ medii_rar[0] }}">
+      <div style="margin-bottom:10px; font-size:var(--fs-sm); color:var(--muted);">
+        Mediu RAR:
+        <strong>{{ "Testare" if medii_rar[0] == "test" else "Productie" }}</strong>
+      </div>
+      {% else %}
+      {# Selector (doua medii disponibile) #}
+      <div style="margin-bottom:10px; display:flex; align-items:center; gap:10px;">
+        <label for="rar-env-select"
+               style="font-size:var(--fs-sm); color:var(--muted); white-space:nowrap;">
+          Mediu RAR:
+        </label>
+        <select id="rar-env-select" name="rar_env">
+          <option value="test" {% if env_default == "test" %}selected{% endif %}>Testare</option>
+          <option value="prod" {% if env_default == "prod" %}selected{% endif %}>Productie</option>
+        </select>
+      </div>
+      {% endif %}
+
       {% if sheets %}
       <div style="margin-bottom:12px;">
         <label for="sheet-select"

app/worker/__main__.py

@@ -39,7 +39,7 @@ from ..observ import log_event, set_source
 from ..mapping import DEFAULT_ACCOUNT_ID, upsert_nomenclator
 from ..payload import build_rar_payload
 from ..reconcile import match_finalizata
-from ..rar_client import RarAuthError, RarClient, RarError
+from ..rar_client import RarAuthError, RarClient, RarError, base_url_pentru_env
 
 _running = True
 
@@ -167,7 +167,7 @@ def claim_one(conn) -> dict | None:
     conn.execute("BEGIN IMMEDIATE")
     try:
         row = conn.execute(
-            "SELECT s.id, s.account_id, s.payload_json, s.rar_creds_enc "
+            "SELECT s.id, s.account_id, s.payload_json, s.rar_creds_enc, s.rar_env "
             "FROM submissions s LEFT JOIN accounts a ON a.id = s.account_id "
             "WHERE s.status='queued' "
             "AND (s.next_attempt_at IS NULL OR s.next_attempt_at <= ?) "
@@ -189,6 +189,7 @@ def claim_one(conn) -> dict | None:
         return {
             "id": row["id"],
             "account_id": row["account_id"] if row["account_id"] is not None else DEFAULT_ACCOUNT_ID,
+            "rar_env": row["rar_env"],
             "creds_enc": row["rar_creds_enc"],
             "content": json.loads(row["payload_json"]),
         }
@@ -281,11 +282,13 @@ def _handle_transient(conn, settings: Settings, rar: RarClient, token: str, sid:
     return "requeued"
 
 
-def recover_orphans(conn, settings: Settings, rar: RarClient, token: str, account_id: int | None = None) -> int:
+def recover_orphans(conn, settings: Settings, rar: RarClient, token: str,
+                    account_id: int | None = None, rar_env: str | None = None) -> int:
     """Randuri 'sending' mai vechi de lease (worker mort mid-POST). Reconciliaza; altfel requeue.
 
-    `account_id` filtreaza la orfanii unui cont (login-ul e per-cont); None = toti
-    (compat teste / single-account).
+    `account_id` filtreaza la orfanii unui cont (login-ul e per-cont); None = toti.
+    `rar_env` filtreaza la orfanii unui mediu (1b/E6): orfanii prod contra endpoint prod,
+    NU contra test — altfel no-match -> re-POST prod = DUPLICAT real ireversibil.
     """
     # Cutoff calculat SQLite-side, in ACELASI format ca sending_since (scris cu
     # datetime('now') in claim_one -> 'YYYY-MM-DD HH:MM:SS', cu spatiu). Daca am
@@ -293,18 +296,18 @@ def recover_orphans(conn, settings: Settings, rar: RarClient, token: str, accoun
     # orice rand 'sending' sa para mereu <= cutoff -> lease-ul de 120s ignorat,
     # iar fiecare rand proaspat revendicat ar fi tratat instant ca orfan.
     lease = f"-{int(settings.worker_sending_lease_s)} seconds"
+    base_sql = (
+        "SELECT id, payload_json FROM submissions WHERE status='sending' "
+        "AND (sending_since IS NULL OR sending_since <= datetime('now', ?))"
+    )
+    params: list = [lease]
     if account_id is not None:
-        orphans = conn.execute(
-            "SELECT id, payload_json FROM submissions WHERE status='sending' "
-            "AND (sending_since IS NULL OR sending_since <= datetime('now', ?)) AND account_id=?",
-            (lease, account_id),
-        ).fetchall()
-    else:
-        orphans = conn.execute(
-            "SELECT id, payload_json FROM submissions WHERE status='sending' "
-            "AND (sending_since IS NULL OR sending_since <= datetime('now', ?))",
-            (lease,),
-        ).fetchall()
+        base_sql += " AND account_id=?"
+        params.append(account_id)
+    if rar_env is not None:
+        base_sql += " AND rar_env=?"
+        params.append(rar_env)
+    orphans = conn.execute(base_sql, params).fetchall()
     recovered = 0
     for row in orphans:
         sid = row["id"]
@@ -337,25 +340,26 @@ def _refresh_nomenclator(conn, rar: RarClient, token: str) -> None:
 
 
 class AccountSessions:
-    """Sesiuni RAR per cont: login lazy cu creds din submission + cache JWT (30h).
+    """Sesiuni RAR per (cont, env): login lazy cu creds din submission + cache JWT (30h).
 
-    La primul login reusit pentru un cont sterge creds-urile criptate ale contului
-    (token-ul in memorie acopera restul). Pe 401 mid-sesiune se invalideaza sesiunea
-    -> re-login la urmatorul submission cu creds.
+    Cheia = (account_id, rar_env): test si prod sunt sisteme RAR separate cu JWT separate.
+    La primul login reusit pentru (cont, env) sterge creds-urile efemere scoped pe acel env.
+    Pe 401 mid-sesiune se invalideaza sesiunea -> re-login la urmatorul submission cu creds.
     """
 
     def __init__(self, settings: Settings):
         self.settings = settings
-        self._sessions: dict[int, tuple[RarClient, str]] = {}
+        self._sessions: dict[tuple[int, str], tuple[RarClient, str]] = {}
 
-    def get_token(self, conn, account_id: int, creds: dict | None) -> str | None:
-        """Token valid pentru cont. Login daca lipseste din cache si avem creds; altfel None."""
-        sess = self._sessions.get(account_id)
+    def get_token(self, conn, account_id: int, creds: dict | None, rar_env: str = "test") -> str | None:
+        """Token valid pentru (cont, env). Login daca lipseste din cache si avem creds; altfel None."""
+        key = (account_id, rar_env)
+        sess = self._sessions.get(key)
         if sess is not None:
             return sess[1]
         if not creds or not creds.get("email") or not creds.get("password"):
             return None
-        rar = RarClient(self.settings)
+        rar = RarClient(self.settings, base_url=base_url_pentru_env(self.settings, rar_env))
         try:
             token = rar.login(creds["email"], creds["password"])
         except RarAuthError as exc:
@@ -363,40 +367,52 @@ class AccountSessions:
             # Login esuat (401) — FARA email/parola (doar codul HTTP + contul).
             log_event("rar_login", nivel="WARNING", account_id=account_id,
                       cod="RAR_CREDS_INVALIDE",
-                      mesaj=f"login RAR esuat (cont {account_id}): {exc.status_code or 401}",
+                      mesaj=f"login RAR esuat (cont {account_id}, env={rar_env}): {exc.status_code or 401}",
                       context={"rezultat": "esuat", "http": exc.status_code or 401},
                       conn=conn, sursa="worker")
             raise
         except Exception:
             rar.close()
             raise
-        self._sessions[account_id] = (rar, token)
-        write_heartbeat(conn, rar_login_ok=True, detail=f"login RAR ok (cont {account_id})")
+        self._sessions[key] = (rar, token)
+        write_heartbeat(conn, rar_login_ok=True, detail=f"login RAR ok (cont {account_id}, env={rar_env})")
         # Login reusit (fara email/parola in clar — context curat).
-        log_event("rar_login", account_id=account_id, mesaj=f"login RAR ok (cont {account_id})",
+        log_event("rar_login", account_id=account_id,
+                  mesaj=f"login RAR ok (cont {account_id}, env={rar_env})",
                   context={"rezultat": "ok", "http": 200}, conn=conn, sursa="worker")
         # Creds efemere pe submissions nu mai sunt necesare: JWT acopera retry-urile -> sterge.
-        # GATE PURJARE: sterge DOAR submissions.rar_creds_enc, NU accounts.rar_creds_enc.
-        # Canal web: fallback exista in accounts -> purjarea e inofensiva (re-login dupa restart).
-        # Canal API pur: purjarea e identica cu Treapta 1 (neatinsa).
+        # GATE PURJARE: sterge DOAR submissions.rar_creds_enc (WHERE account_id=? AND rar_env=?),
+        # NU accounts.rar_creds_{env}_enc. Scoped pe env: login test NU sterge creds prod (E1/1a).
         conn.execute(
-            "UPDATE submissions SET rar_creds_enc=NULL WHERE account_id=? AND rar_creds_enc IS NOT NULL",
-            (account_id,),
+            "UPDATE submissions SET rar_creds_enc=NULL "
+            "WHERE account_id=? AND rar_env=? AND rar_creds_enc IS NOT NULL",
+            (account_id, rar_env),
         )
         # Nomenclator live (autoritativ) la fiecare login proaspat.
+        # Nota (1e): nomenclatorul e presupus identic intre medii (aceleasi 18 coduri RAR);
+        # daca diverge in viitor, scoparea per-env a tabelei nomenclator_rar e out of scope acum.
         _refresh_nomenclator(conn, rar, token)
         return token
 
-    def rar(self, account_id: int) -> RarClient:
-        return self._sessions[account_id][0]
+    def rar(self, account_id: int, rar_env: str = "test") -> RarClient:
+        return self._sessions[(account_id, rar_env)][0]
 
-    def active(self) -> list[tuple[int, RarClient, str]]:
-        return [(acct, rar, tok) for acct, (rar, tok) in self._sessions.items()]
+    def active(self) -> list[tuple[int, str, RarClient, str]]:
+        """Sesiunile active: lista de (account_id, rar_env, RarClient, token)."""
+        return [(acct, env, rar, tok) for (acct, env), (rar, tok) in self._sessions.items()]
 
-    def invalidate(self, account_id: int) -> None:
-        sess = self._sessions.pop(account_id, None)
-        if sess is not None:
-            sess[0].close()
+    def invalidate(self, account_id: int, rar_env: str | None = None) -> None:
+        """Invalideaza sesiunea (cont, env). rar_env=None invalideaza TOATE sesiunile contului."""
+        if rar_env is not None:
+            sess = self._sessions.pop((account_id, rar_env), None)
+            if sess is not None:
+                sess[0].close()
+        else:
+            # Invalideaza toate mediile pentru acest cont.
+            keys_to_remove = [k for k in self._sessions if k[0] == account_id]
+            for k in keys_to_remove:
+                sess = self._sessions.pop(k)
+                sess[0].close()
 
     def close_all(self) -> None:
         for rar, _tok in self._sessions.values():
@@ -414,33 +430,39 @@ def _creds_for(claimed: dict, settings: Settings) -> dict | None:
     return None
 
 
-def _creds_from_account(conn, account_id: int) -> dict | None:
-    """Fallback: crede RAR durabile per-cont din accounts.rar_creds_enc.
+def _creds_from_account(conn, account_id: int, rar_env: str = "test") -> dict | None:
+    """Creds RAR durabile per-cont din slotul per-env, cu fallback la coloana legacy.
 
-    Canal web nu are re-pusher. Cand submission n-are creds (sterse dupa primul login
-    sau upload web fara creds), worker-ul re-citeste din cont si poate re-login oricand.
+    Canal web: creds in accounts.rar_creds_{rar_env}_enc (per-env). Fallback la
+    accounts.rar_creds_enc (legacy, back-compat inainte de US-013 care dropa coloana veche).
     """
+    env_slot = f"rar_creds_{rar_env}_enc"
     row = conn.execute(
-        "SELECT rar_creds_enc FROM accounts WHERE id=?", (account_id,)
+        f"SELECT {env_slot}, rar_creds_enc FROM accounts WHERE id=?", (account_id,)
     ).fetchone()
-    if row and row["rar_creds_enc"]:
-        return decrypt_creds(row["rar_creds_enc"])
-    return None
+    if not row:
+        return None
+    enc = row[env_slot] or row["rar_creds_enc"]  # per-env intai, legacy fallback
+    return decrypt_creds(enc) if enc else None
 
 
 def _keepalive_target(conn, settings: Settings) -> tuple[int | None, dict | None]:
     """Un cont cu creds durabile pentru login-ul de proba (sau creds <test> in dev).
 
+    Ancora M2: cauta in slotul per-env al mediului `settings.rar_env` (ancora globala).
+    Fallback la coloana legacy `rar_creds_enc` (back-compat inainte de US-013).
     Sare conturile ale caror creds NU se decripteaza sub cheia curenta — in dev
-    `start.sh both` genereaza o cheie efemera noua la fiecare pornire, deci creds-urile
-    durabile criptate sub cheia veche dau decrypt -> None. Fallback la creds <test>.
+    `start.sh both` genereaza o cheie efemera noua la fiecare pornire.
     """
+    env_slot = f"rar_creds_{settings.rar_env}_enc"
     rows = conn.execute(
-        "SELECT id, rar_creds_enc FROM accounts "
-        "WHERE rar_creds_enc IS NOT NULL ORDER BY id"
+        f"SELECT id, {env_slot}, rar_creds_enc FROM accounts ORDER BY id"
     ).fetchall()
     for row in rows:
-        creds = decrypt_creds(row["rar_creds_enc"])
+        enc = row[env_slot] or row["rar_creds_enc"]  # per-env intai, legacy fallback
+        if not enc:
+            continue
+        creds = decrypt_creds(enc)
         if creds and creds.get("email") and creds.get("password"):
             return row["id"], creds
     if settings.worker_use_test_creds:
@@ -478,9 +500,9 @@ def _maybe_keepalive(conn, settings: Settings, sessions: "AccountSessions", stat
     account_id, creds = _keepalive_target(conn, settings)
     if account_id is None or not creds:
         return  # niciun cont cu creds durabile — nimic de sondat
-    sessions.invalidate(account_id)  # forteaza login real, nu token din cache
+    sessions.invalidate(account_id, settings.rar_env)  # forteaza login real pt env-ul global
     try:
-        sessions.get_token(conn, account_id, creds)  # reimprospateaza last_rar_login_ok la succes
+        sessions.get_token(conn, account_id, creds, settings.rar_env)  # reimprospateaza last_rar_login_ok la succes
     except RarAuthError:
         pass  # creds invalide — deja logat in get_token (WARNING)
     except Exception as exc:
@@ -527,8 +549,8 @@ def run() -> int:
             claimed = claim_one(conn)
             if claimed is None:
                 # Nimic de trimis: recupereaza orfanii conturilor deja logate.
-                for acct, rar, tok in sessions.active():
-                    recover_orphans(conn, settings, rar, tok, account_id=acct)
+                for acct, env, rar, tok in sessions.active():
+                    recover_orphans(conn, settings, rar, tok, account_id=acct, rar_env=env)
                 # Login de proba periodic ca dashboard-ul sa nu afiseze fals
                 # "RAR inaccesibil" din lipsa de trafic (vezi _maybe_keepalive).
                 _maybe_keepalive(conn, settings, sessions, _keepalive_state)
@@ -537,18 +559,20 @@ def run() -> int:
 
             sid = claimed["id"]
             account_id = claimed["account_id"]
+            # Mediul tinta al trimiterii (per-submission, cu fallback la ancora globala).
+            rar_env = claimed.get("rar_env") or settings.rar_env
             # Randul poarta creds proaspete (rar_creds_enc != NULL) — fie prima trimitere
             # a contului, fie o REACTIVARE dupa creds gresite. Invalidam sesiunea RAR
-            # cache-uita ca un JWT vechi (30h) din parola GRESITA sa nu trimita cu ea,
-            # ignorand corectia. Re-login imediat cu creds-urile noi.
+            # cache-uita (per env) ca un JWT vechi (30h) din parola GRESITA sa nu trimita
+            # cu ea, ignorand corectia. Re-login imediat cu creds-urile noi.
             if claimed.get("creds_enc"):
-                sessions.invalidate(account_id)
+                sessions.invalidate(account_id, rar_env)
             # Incearca creds din submission (canal API efemer), cu fallback la
-            # accounts.rar_creds_enc (canal web durabil). Canal web n-are re-pusher.
-            creds = _creds_for(claimed, settings) or _creds_from_account(conn, account_id)
+            # accounts.rar_creds_{rar_env}_enc (canal web durabil, per-env).
+            creds = _creds_for(claimed, settings) or _creds_from_account(conn, account_id, rar_env)
 
             try:
-                token = sessions.get_token(conn, account_id, creds)
+                token = sessions.get_token(conn, account_id, creds, rar_env)
             except RarAuthError as exc:
                 # Creds gresite (login 401): NU se face retry.
                 mark(conn, sid, "error", rar_status_code=401,
@@ -565,9 +589,9 @@ def run() -> int:
                 requeue_with_backoff(conn, settings, sid, reason="creds RAR indisponibile (astept re-trimitere)")
                 continue
 
-            rar = sessions.rar(account_id)
-            # Recupereaza orfanii contului inainte de trimitere (acelasi token).
-            recover_orphans(conn, settings, rar, token, account_id=account_id)
+            rar = sessions.rar(account_id, rar_env)
+            # Recupereaza orfanii contului + env-ului inainte de trimitere (acelasi token).
+            recover_orphans(conn, settings, rar, token, account_id=account_id, rar_env=rar_env)
             # Guard: recover_orphans putea atinge chiar randul tocmai revendicat
             # (reconciliat 'sent' sau requeue 'queued'). Daca nu mai e 'sending',
             # NU mai face POST — altfel s-ar crea un duplicat la RAR.
@@ -579,9 +603,9 @@ def run() -> int:
             try:
                 process_one(conn, settings, rar, token, claimed)
             except RarAuthError as exc:
-                # Token expirat mid-sesiune: invalideaza sesiunea, re-pune randul.
-                print(f"[worker] cont {account_id} token expirat: {exc}; re-login data viitoare", flush=True)
-                sessions.invalidate(account_id)
+                # Token expirat mid-sesiune: invalideaza sesiunea (per env), re-pune randul.
+                print(f"[worker] cont {account_id} env={rar_env} token expirat: {exc}; re-login data viitoare", flush=True)
+                sessions.invalidate(account_id, rar_env)
                 requeue_with_backoff(conn, settings, sid, reason="token RAR expirat")
 
         except Exception as exc:  # noqa: BLE001 — loop top-level: o eroare punctuala nu opreste worker-ul