From 19d8aaa7aa6757886fba12f051e83693b2033b61 Mon Sep 17 00:00:00 2001
From: Claude Agent <claude-agent@romfast.ro>
Date: Mon, 29 Jun 2026 20:30:11 +0000
Subject: [PATCH] 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            |  35 ++-
 app/api/v1/router.py                   |  76 +++++-
 app/errors.py                          |   7 +
 app/models.py                          |   9 +
 app/rar_client.py                      |   9 +-
 app/rar_env.py                         |  65 +++++
 app/web/routes.py                      | 147 ++++++++--
 app/web/templates/_preview_import.html |  10 +
 app/web/templates/_upload.html         |  32 +++
 app/worker/__main__.py                 | 166 +++++++-----
 tests/test_api_rar_target.py           | 226 ++++++++++++++++
 tests/test_creds_delivery.py           |   4 +-
 tests/test_import_e2e.py               |   4 +-
 tests/test_import_rar_env.py           | 359 +++++++++++++++++++++++++
 tests/test_rar_env_resolve.py          |  96 +++++++
 tests/test_t1_creds_durabile.py        |   2 +-
 tests/test_worker_keepalive_rar.py     |   4 +-
 tests/test_worker_observ.py            |   4 +-
 tests/test_worker_rar_env.py           | 326 ++++++++++++++++++++++
 19 files changed, 1451 insertions(+), 130 deletions(-)
 create mode 100644 tests/test_api_rar_target.py
 create mode 100644 tests/test_import_rar_env.py
 create mode 100644 tests/test_rar_env_resolve.py
 create mode 100644 tests/test_worker_rar_env.py

diff --git a/app/api/v1/import_router.py b/app/api/v1/import_router.py
index b55a810..61c566f 100644
--- a/app/api/v1/import_router.py
+++ b/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:
diff --git a/app/api/v1/router.py b/app/api/v1/router.py
index a2f722b..dd1576e 100644
--- a/app/api/v1/router.py
+++ b/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",
 })
 
 
diff --git a/app/errors.py b/app/errors.py
index a58d251..dabb64d 100644
--- a/app/errors.py
+++ b/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."
+        ),
+    },
 }
 
 
diff --git a/app/models.py b/app/models.py
index e6820fc..4045912 100644
--- a/app/models.py
+++ b/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] = []
diff --git a/app/rar_client.py b/app/rar_client.py
index 05053c5..a6bcf2a 100644
--- a/app/rar_client.py
+++ b/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
         )
diff --git a/app/rar_env.py b/app/rar_env.py
index df20c52..43ba425 100644
--- a/app/rar_env.py
+++ b/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
diff --git a/app/web/routes.py b/app/web/routes.py
index 53dcf5c..7bacf7f 100644
--- a/app/web/routes.py
+++ b/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)
diff --git a/app/web/templates/_preview_import.html b/app/web/templates/_preview_import.html
index 492d351..1327757 100644
--- a/app/web/templates/_preview_import.html
+++ b/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 -->
diff --git a/app/web/templates/_upload.html b/app/web/templates/_upload.html
index 06322da..c4c2632 100644
--- a/app/web/templates/_upload.html
+++ b/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"
diff --git a/app/worker/__main__.py b/app/worker/__main__.py
index 2c3ff30..5a53cd6 100644
--- a/app/worker/__main__.py
+++ b/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
diff --git a/tests/test_api_rar_target.py b/tests/test_api_rar_target.py
new file mode 100644
index 0000000..fc01ecc
--- /dev/null
+++ b/tests/test_api_rar_target.py
@@ -0,0 +1,226 @@
+"""Teste US-005 — camp rar_env pe POST /v1/prezentari si /valideaza.
+
+Acopera: default din cont, tinta explicita, respingere tinta indisponibila,
+echo GET, valoare invalida (422 Pydantic), echo dry-run valideaza.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    """DB temporara izolata per test + settings reincarcate."""
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
+    from app.config import get_settings
+    get_settings.cache_clear()
+    yield monkeypatch
+    get_settings.cache_clear()
+
+
+def _client():
+    from app.main import app
+    return TestClient(app)
+
+
+def _body(rar_env=None, **over):
+    prez = {
+        "vin": "WVWZZZ1KZAW000123",
+        "nr_inmatriculare": "B999TST",
+        "data_prestatie": "2026-06-15",
+        "odometru_final": "123456",
+        "prestatii": [{"cod_prestatie": "OE-1"}],
+    }
+    prez.update(over)
+    body = {"rar_credentials": {"email": "x@y.ro", "password": "s"}, "prezentari": [prez]}
+    if rar_env is not None:
+        body["rar_env"] = rar_env
+    return body
+
+
+def _setup_prod_only(conn):
+    """Configureaza contul 1 ca prod-only (rar_prod_enabled=1, creds prod, default prod)."""
+    from app.crypto import encrypt_creds
+    enc = encrypt_creds({"email": "prod@rar.ro", "password": "paraprod"})
+    conn.execute(
+        "UPDATE accounts SET rar_prod_enabled=1, rar_creds_prod_enc=?, "
+        "rar_test_enabled=0, rar_creds_test_enc=NULL, rar_env_default='prod' WHERE id=1",
+        (enc,),
+    )
+    conn.commit()
+
+
+def _setup_dual_env(conn):
+    """Configureaza contul 1 cu ambele medii disponibile, default test."""
+    from app.crypto import encrypt_creds
+    enc_test = encrypt_creds({"email": "test@rar.ro", "password": "paratest"})
+    enc_prod = encrypt_creds({"email": "prod@rar.ro", "password": "paraprod"})
+    conn.execute(
+        "UPDATE accounts SET rar_prod_enabled=1, rar_creds_prod_enc=?, "
+        "rar_test_enabled=1, rar_creds_test_enc=?, rar_env_default='test' WHERE id=1",
+        (enc_prod, enc_test),
+    )
+    conn.commit()
+
+
+# --------------------------------------------------------------------------- #
+# test_default_din_cont_cand_lipseste                                          #
+# --------------------------------------------------------------------------- #
+
+def test_default_din_cont_cand_lipseste(env):
+    """Cont prod-only, POST fara rar_env -> submission rar_env='prod'."""
+    with _client() as c:
+        from app.db import get_connection
+        conn = get_connection()
+        try:
+            _setup_prod_only(conn)
+        finally:
+            conn.close()
+
+        r = c.post("/v1/prezentari", json=_body())
+        assert r.status_code == 200, r.text
+        res = r.json()["results"][0]
+        assert res["status"] == "queued"
+        assert res["rar_env"] == "prod"
+
+
+# --------------------------------------------------------------------------- #
+# test_target_explicit                                                          #
+# --------------------------------------------------------------------------- #
+
+def test_target_explicit(env):
+    """Cont cu ambele medii, POST cu rar_env='test' -> submission rar_env='test'."""
+    with _client() as c:
+        from app.db import get_connection
+        conn = get_connection()
+        try:
+            _setup_dual_env(conn)
+        finally:
+            conn.close()
+
+        r = c.post("/v1/prezentari", json=_body(rar_env="test"))
+        assert r.status_code == 200, r.text
+        res = r.json()["results"][0]
+        assert res["status"] == "queued"
+        assert res["rar_env"] == "test"
+
+        # Aceeasi prezentare cu rar_env='prod' -> cheie diferita (env-aware) -> alt submission
+        r2 = c.post("/v1/prezentari", json=_body(rar_env="prod"))
+        assert r2.status_code == 200, r2.text
+        res2 = r2.json()["results"][0]
+        assert res2["rar_env"] == "prod"
+        # Nu e dedup (env diferit -> cheie diferita)
+        assert res2["submission_id"] != res["submission_id"]
+
+
+# --------------------------------------------------------------------------- #
+# test_target_indisponibil_respins                                              #
+# --------------------------------------------------------------------------- #
+
+def test_target_indisponibil_respins(env):
+    """Cont prod-only, POST cu rar_env='test' -> 422 RAR_MEDIU_INDISPONIBIL, fara enqueue."""
+    with _client() as c:
+        from app.db import get_connection
+        conn = get_connection()
+        try:
+            _setup_prod_only(conn)
+        finally:
+            conn.close()
+
+        r = c.post("/v1/prezentari", json=_body(rar_env="test"))
+        assert r.status_code == 422, r.text
+        detail = r.json()["detail"]
+        assert detail["cod"] == "RAR_MEDIU_INDISPONIBIL"
+        # Cauza contine mediul cerut si lista disponibilelor
+        assert "test" in detail["cauza"]
+        assert "prod" in detail["cauza"]
+
+        # Verifica ca nu s-a facut enqueue
+        lista = c.get("/v1/prezentari").json()["submissions"]
+        assert lista == []
+
+
+# --------------------------------------------------------------------------- #
+# test_get_ecou_rar_env                                                        #
+# --------------------------------------------------------------------------- #
+
+def test_get_ecou_rar_env(env):
+    """Dupa enqueue, GET /v1/prezentari/{id} si lista contin rar_env."""
+    with _client() as c:
+        from app.db import get_connection
+        conn = get_connection()
+        try:
+            _setup_prod_only(conn)
+        finally:
+            conn.close()
+
+        # Enqueue pe prod (default contului prod-only)
+        r = c.post("/v1/prezentari", json=_body())
+        assert r.status_code == 200, r.text
+        sub_id = r.json()["results"][0]["submission_id"]
+        assert sub_id is not None
+
+        # GET detaliu
+        r_det = c.get(f"/v1/prezentari/{sub_id}")
+        assert r_det.status_code == 200, r_det.text
+        assert r_det.json()["rar_env"] == "prod"
+
+        # GET lista
+        r_lst = c.get("/v1/prezentari")
+        assert r_lst.status_code == 200, r_lst.text
+        sub_in_lista = next(s for s in r_lst.json()["submissions"] if s["id"] == sub_id)
+        assert sub_in_lista["rar_env"] == "prod"
+
+
+# --------------------------------------------------------------------------- #
+# test_valoare_invalida_422                                                     #
+# --------------------------------------------------------------------------- #
+
+def test_valoare_invalida_422(env):
+    """POST cu rar_env='staging' -> 422 din Pydantic Literal, fara echo de input."""
+    with _client() as c:
+        body = _body(rar_env="staging")
+        r = c.post("/v1/prezentari", json=body)
+        assert r.status_code == 422, r.text
+        # Handler-ul global sterge 'input'/'ctx' — valoarea invalida nu se ecou-ieste.
+        assert "staging" not in r.text
+        # Fara enqueue
+        lista = c.get("/v1/prezentari").json()["submissions"]
+        assert lista == []
+
+
+# --------------------------------------------------------------------------- #
+# test_valideaza_ecou_rar_env                                                  #
+# --------------------------------------------------------------------------- #
+
+def test_valideaza_ecou_rar_env(env):
+    """POST /valideaza (dry-run) ecou-ieste rar_env rezolvat in ValidareResult."""
+    with _client() as c:
+        from app.db import get_connection
+        conn = get_connection()
+        try:
+            _setup_prod_only(conn)
+        finally:
+            conn.close()
+
+        # Dry-run fara rar_env -> default cont = prod
+        r = c.post("/v1/prezentari/valideaza", json=_body())
+        assert r.status_code == 200, r.text
+        res = r.json()["results"][0]
+        assert res["rar_env"] == "prod"
+
+        # Dry-run cu rar_env='prod' explicit
+        r2 = c.post("/v1/prezentari/valideaza", json=_body(rar_env="prod"))
+        assert r2.status_code == 200, r2.text
+        assert r2.json()["results"][0]["rar_env"] == "prod"
+
+        # Dry-run tinta indisponibila -> 422, fara echo sensibil
+        r3 = c.post("/v1/prezentari/valideaza", json=_body(rar_env="test"))
+        assert r3.status_code == 422, r3.text
+        assert r3.json()["detail"]["cod"] == "RAR_MEDIU_INDISPONIBIL"
diff --git a/tests/test_creds_delivery.py b/tests/test_creds_delivery.py
index 562d408..504b355 100644
--- a/tests/test_creds_delivery.py
+++ b/tests/test_creds_delivery.py
@@ -157,7 +157,7 @@ class FakeRarClient:
 
     made: list = []
 
-    def __init__(self, settings=None, login_exc=None):
+    def __init__(self, settings=None, *, base_url=None, login_exc=None):
         self.closed = False
         self.login_calls = 0
         self._login_exc = login_exc
@@ -234,7 +234,7 @@ def test_get_token_bad_creds_raises(env, monkeypatch):
     import app.worker.__main__ as w
     from app.db import get_connection
 
-    def _factory(settings=None):
+    def _factory(settings=None, **kwargs):
         return FakeRarClient(settings, login_exc=RarAuthError("Credentiale RAR invalide", status_code=401))
 
     monkeypatch.setattr(w, "RarClient", _factory)
diff --git a/tests/test_import_e2e.py b/tests/test_import_e2e.py
index 66c460e..f68309e 100644
--- a/tests/test_import_e2e.py
+++ b/tests/test_import_e2e.py
@@ -511,7 +511,7 @@ class TestE2EMixedQueue:
 
         # 4. Worker cu MockRar injectat prin AccountSessions (simulam bucla worker)
         mock_rar = MockRar(id_prezentare=66001, login_token="tok-mock")
-        monkeypatch.setattr(w, "RarClient", lambda settings=None: mock_rar)
+        monkeypatch.setattr(w, "RarClient", lambda settings=None, **kw: mock_rar)
 
         sessions = w.AccountSessions(settings)
         conn = get_connection()
@@ -595,7 +595,7 @@ class TestE2EMixedQueue:
             conn.close()
 
         mock_rar = MockRar(id_prezentare=77777)
-        monkeypatch.setattr(w, "RarClient", lambda settings=None: mock_rar)
+        monkeypatch.setattr(w, "RarClient", lambda settings=None, **kw: mock_rar)
 
         sessions = w.AccountSessions(settings)
         conn = get_connection()
diff --git a/tests/test_import_rar_env.py b/tests/test_import_rar_env.py
new file mode 100644
index 0000000..81155dd
--- /dev/null
+++ b/tests/test_import_rar_env.py
@@ -0,0 +1,359 @@
+"""Teste US-009 (PRD 5.20) — Import web: selector mediu RAR conditionat de disponibilitate.
+
+Verifica:
+  - La 0 medii: banner avertisment non-blocant (upload functioneaza, commit foloseste ancora globala).
+  - La 1 mediu: eticheta statica, fara selector; submissions primesc acel mediu.
+  - La 2 medii: selector vizibil pre-bifat pe default-ul contului.
+  - La commit: toate submission-urile lotului primesc rar_env ales (sau fallback ancora globala).
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import os
+import re
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# --------------------------------------------------------------------------- #
+# Fixture client cu DB izolat                                                   #
+# --------------------------------------------------------------------------- #
+
+@pytest.fixture()
+def client(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "rar_env_test.db"))
+    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false")
+    monkeypatch.setenv("AUTOPASS_RAR_ENV", "test")  # ancora globala = test
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.crypto import reset_cache
+    reset_cache()
+    from app.main import app
+    with TestClient(app) as c:
+        yield c
+    get_settings.cache_clear()
+    reset_cache()
+
+
+# --------------------------------------------------------------------------- #
+# Utilitare                                                                     #
+# --------------------------------------------------------------------------- #
+
+def _csv_bytes(rows: list[dict], sep: str = ";") -> bytes:
+    buf = io.StringIO()
+    writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=sep)
+    writer.writeheader()
+    writer.writerows(rows)
+    return buf.getvalue().encode("utf-8")
+
+
+def _seed_nomenclator_si_mapare(client: TestClient) -> None:
+    """Semeaza nomenclatorul si o mapare pentru randuri ok."""
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        conn.execute(
+            "INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?,?)",
+            ("R-FRANE", "Reparatie frane"),
+        )
+        conn.execute(
+            "INSERT OR IGNORE INTO operations_mapping "
+            "(account_id, cod_op_service, cod_prestatie, auto_send) VALUES (1,?,?,1)",
+            ("OP-FRANE", "R-FRANE"),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _configureaza_un_mediu(client: TestClient, env: str = "test") -> None:
+    """Activeaza un singur mediu RAR pe contul 1 (simulate creds disponibile)."""
+    from app.db import get_connection
+    from app.crypto import encrypt_creds
+    conn = get_connection()
+    try:
+        fake_creds = encrypt_creds({"email": "test@rar.ro", "password": "pass"})
+        if env == "test":
+            conn.execute(
+                "UPDATE accounts SET rar_test_enabled=1, rar_creds_test_enc=?, "
+                "rar_prod_enabled=0, rar_creds_prod_enc=NULL, rar_env_default='test' WHERE id=1",
+                (fake_creds,),
+            )
+        else:
+            conn.execute(
+                "UPDATE accounts SET rar_prod_enabled=1, rar_creds_prod_enc=?, "
+                "rar_test_enabled=0, rar_creds_test_enc=NULL, rar_env_default='prod' WHERE id=1",
+                (fake_creds,),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _configureaza_doua_medii(client: TestClient, default_env: str = "test") -> None:
+    """Activeaza ambele medii RAR pe contul 1."""
+    from app.db import get_connection
+    from app.crypto import encrypt_creds
+    conn = get_connection()
+    try:
+        fake_test = encrypt_creds({"email": "test@rar.ro", "password": "pass_test"})
+        fake_prod = encrypt_creds({"email": "prod@rar.ro", "password": "pass_prod"})
+        conn.execute(
+            "UPDATE accounts SET "
+            "rar_test_enabled=1, rar_creds_test_enc=?, "
+            "rar_prod_enabled=1, rar_creds_prod_enc=?, "
+            "rar_env_default=? WHERE id=1",
+            (fake_test, fake_prod, default_env),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+_ROWS_OK = [
+    {
+        "VIN": "WVWZZZ1KZAW009001",
+        "Nr": "B009TST",
+        "Data": "2026-06-15",
+        "KM": "77000",
+        "Operatie": "OP-FRANE",
+    },
+]
+
+
+def _upload_si_mapare(client: TestClient, rows: list[dict]) -> int:
+    """Upload CSV si seteaza mapare coloane. Intoarce import_id."""
+    data = _csv_bytes(rows)
+    r = client.post(
+        "/_import/upload",
+        files={"file": ("test.csv", io.BytesIO(data), "text/csv")},
+    )
+    assert r.status_code == 200, r.text
+    m = re.search(r"/_import/(\d+)/", r.text)
+    assert m, f"import_id negasit in raspunsul de upload: {r.text[:400]}"
+    iid = int(m.group(1))
+
+    # Seteaza maparea daca nu e deja
+    if f"/_import/{iid}/mapare-coloane" in r.text or "mapare-coloane" in r.text.lower():
+        r2 = client.post(
+            f"/_import/{iid}/mapare-coloane",
+            data={
+                "colname": ["VIN", "Nr", "Data", "KM", "Operatie"],
+                "canon":   ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
+                "format_data": "YYYY-MM-DD",
+            },
+        )
+        assert r2.status_code == 200, r2.text
+
+    return iid
+
+
+def _get_preview(client: TestClient, iid: int) -> str:
+    rp = client.get(f"/_import/{iid}/preview")
+    assert rp.status_code == 200, rp.text
+    return rp.text
+
+
+def _commit(client: TestClient, iid: int, n_ok: int, rar_env: str | None = None) -> object:
+    data = {
+        "csrf_token": "",
+        "n_confirmat": str(n_ok),
+        "confirmed_by": "test@us009.ro",
+    }
+    if rar_env:
+        data["rar_env"] = rar_env
+    return client.post(f"/_import/{iid}/confirma", data=data)
+
+
+# --------------------------------------------------------------------------- #
+# Tests                                                                         #
+# --------------------------------------------------------------------------- #
+
+def test_selector_ascuns_la_un_mediu(client):
+    """La 1 mediu disponibil: nu apare selector; apare eticheta statica cu mediul."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_un_mediu(client, env="test")
+
+    # GET fragment/import: verifica ca nu exista selector si apare eticheta
+    r = client.get("/_fragments/import")
+    assert r.status_code == 200, r.text
+    html = r.text
+
+    # Eticheta statica "Testare" trebuie sa fie prezenta
+    assert "Testare" in html, "Eticheta mediu 'Testare' lipseste la 1 mediu disponibil"
+
+    # Selectorul nu trebuie sa apara (input cu name=rar_env hidden, dar fara <select>)
+    assert "<select" not in html or 'name="rar_env"' not in html or "rar-env-select" not in html, (
+        "Selector mediu RAR nu trebuie sa apara la 1 mediu disponibil"
+    )
+
+
+def test_selector_prezent_si_prebifat_la_doua(client):
+    """La 2 medii disponibile: selectorul apare si e pre-bifat pe default-ul contului."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_doua_medii(client, default_env="test")
+
+    r = client.get("/_fragments/import")
+    assert r.status_code == 200, r.text
+    html = r.text
+
+    # Selectorul trebuie sa apara
+    assert "rar-env-select" in html, "Selectorul mediu RAR lipseste la 2 medii disponibile"
+    assert 'name="rar_env"' in html, 'Atribut name="rar_env" lipsa din selector'
+
+    # Default pre-selectat = "test" (default contului)
+    # Optiunea Testare trebuie sa fie selectata
+    assert 'value="test"' in html and "selected" in html, (
+        "Optiunea Testare nu e pre-selectata (default cont = test)"
+    )
+
+
+def test_banner_avertisment_la_zero_medii(client):
+    """La 0 medii configurate: apare un banner de avertisment (non-blocant)."""
+    # Contul 1 implicit nu are medii configurate
+    r = client.get("/_fragments/import")
+    assert r.status_code == 200, r.text
+    html = r.text
+
+    # Banner avertisment sau link catre configurare credentiale
+    assert "mediu" in html.lower() or "configureaza" in html.lower() or "credentiale" in html.lower(), (
+        "Bannerul de avertisment pentru 0 medii lipseste din pagina de upload"
+    )
+
+    # Upload-ul NU e blocat: formularul de upload trebuie sa fie prezent
+    assert "upload-form" in html, (
+        "Formularul de upload lipseste — la 0 medii upload-ul nu trebuie blocat"
+    )
+
+
+def test_commit_seteaza_env_pe_submissions(client):
+    """La commit: submissions primesc rar_env ales (fallback la ancora globala pt 0 medii)."""
+    _seed_nomenclator_si_mapare(client)
+    # Contul 1 fara medii configurate -> ancora globala = "test"
+
+    iid = _upload_si_mapare(client, _ROWS_OK)
+    preview_html = _get_preview(client, iid)
+
+    m_ok = re.search(r'id="n-confirmat"[^>]*?value="(\d+)"', preview_html)
+    n_ok = int(m_ok.group(1)) if m_ok else 1
+
+    r = _commit(client, iid, n_ok)
+    assert r.status_code == 200, r.text
+    assert any(kw in r.text.lower() for kw in ("coada", "prezenta", "trimiter")), (
+        "Mesajul de succes lipseste din raspunsul de commit"
+    )
+
+    # Verifica ca submission-ul are rar_env setat (fallback "test" via ancora globala)
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        sub = conn.execute(
+            "SELECT rar_env FROM submissions WHERE account_id=1 ORDER BY id DESC LIMIT 1"
+        ).fetchone()
+    finally:
+        conn.close()
+
+    assert sub is not None, "Niciun submission gasit dupa commit"
+    assert sub["rar_env"] in ("test", "prod"), f"rar_env invalid: {sub['rar_env']!r}"
+    # Cu AUTOPASS_RAR_ENV=test si 0 medii configurate, expect "test"
+    assert sub["rar_env"] == "test", (
+        f"Expected rar_env='test' (ancora globala) dar primit {sub['rar_env']!r}"
+    )
+
+
+def test_commit_cu_un_mediu_seteaza_acel_mediu(client):
+    """La commit cu 1 mediu configurat: submission primeste mediul respectiv."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_un_mediu(client, env="test")
+
+    iid = _upload_si_mapare(client, _ROWS_OK)
+    preview_html = _get_preview(client, iid)
+
+    m_ok = re.search(r'id="n-confirmat"[^>]*?value="(\d+)"', preview_html)
+    n_ok = int(m_ok.group(1)) if m_ok else 1
+
+    r = _commit(client, iid, n_ok)
+    assert r.status_code == 200, r.text
+
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        sub = conn.execute(
+            "SELECT rar_env FROM submissions WHERE account_id=1 ORDER BY id DESC LIMIT 1"
+        ).fetchone()
+    finally:
+        conn.close()
+
+    assert sub is not None
+    assert sub["rar_env"] == "test", (
+        f"Expected rar_env='test' (singurul mediu disponibil) dar primit {sub['rar_env']!r}"
+    )
+
+
+def test_commit_cu_doua_medii_respecta_alegerea(client):
+    """La 2 medii: commit cu rar_env explicit seteaza mediul ales pe submissions."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_doua_medii(client, default_env="test")
+
+    iid = _upload_si_mapare(client, _ROWS_OK)
+    preview_html = _get_preview(client, iid)
+
+    m_ok = re.search(r'id="n-confirmat"[^>]*?value="(\d+)"', preview_html)
+    n_ok = int(m_ok.group(1)) if m_ok else 1
+
+    # Commit explicit pe "prod"
+    r = _commit(client, iid, n_ok, rar_env="prod")
+    assert r.status_code == 200, r.text
+
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        sub = conn.execute(
+            "SELECT rar_env FROM submissions WHERE account_id=1 ORDER BY id DESC LIMIT 1"
+        ).fetchone()
+    finally:
+        conn.close()
+
+    assert sub is not None
+    assert sub["rar_env"] == "prod", (
+        f"Expected rar_env='prod' (ales explicit) dar primit {sub['rar_env']!r}"
+    )
+
+
+def test_badge_mediu_in_preview(client):
+    """Preview-ul afiseaza badge-ul cu mediul tinta (US-009, F9/F10)."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_un_mediu(client, env="test")
+
+    iid = _upload_si_mapare(client, _ROWS_OK)
+
+    r = client.get(f"/_import/{iid}/preview")
+    assert r.status_code == 200, r.text
+    html = r.text
+
+    # Badge cu mediul trebuie sa fie prezent in HTML
+    assert "Testare" in html or "PRODUCTIE" in html or "rar_env" in html, (
+        "Badge-ul de mediu RAR lipseste din preview"
+    )
+
+
+def test_rar_env_in_confirm_form(client):
+    """Preview-ul contine un field hidden rar_env in formularul de confirmare."""
+    _seed_nomenclator_si_mapare(client)
+    _configureaza_un_mediu(client, env="test")
+
+    iid = _upload_si_mapare(client, _ROWS_OK)
+
+    r = client.get(f"/_import/{iid}/preview")
+    assert r.status_code == 200, r.text
+    html = r.text
+
+    # Formularul de confirmare trebuie sa contina rar_env ca hidden field
+    assert 'name="rar_env"' in html, (
+        "Campul hidden 'rar_env' lipseste din formularul de confirmare preview"
+    )
diff --git a/tests/test_rar_env_resolve.py b/tests/test_rar_env_resolve.py
new file mode 100644
index 0000000..354a5ae
--- /dev/null
+++ b/tests/test_rar_env_resolve.py
@@ -0,0 +1,96 @@
+"""Teste US-004 (PRD 5.20): rezolvare mediu tinta la ingestie + respingere tinte indisponibile.
+
+Fixtura `conn` urmareste acelasi pattern ca tests/test_accounts.py:
+monkeypatch AUTOPASS_DB_PATH pe tempdir, cache_clear, init_db, get_connection.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import pytest
+
+
+@pytest.fixture()
+def conn(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_rar_env_resolve.db"))
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.db import get_connection, init_db
+    init_db()
+    c = get_connection()
+    yield c
+    c.close()
+    get_settings.cache_clear()
+
+
+def _seteaza_cont_ambele(conn) -> None:
+    """Configureaza contul id=1 cu ambele medii disponibile, default = prod."""
+    conn.execute(
+        """UPDATE accounts
+           SET rar_test_enabled=1, rar_creds_test_enc='T',
+               rar_prod_enabled=1, rar_creds_prod_enc='P',
+               rar_env_default='prod'
+           WHERE id=1"""
+    )
+    conn.commit()
+
+
+def _seteaza_cont_doar_prod(conn) -> None:
+    """Configureaza contul id=1 cu doar prod disponibil, test off."""
+    conn.execute(
+        """UPDATE accounts
+           SET rar_test_enabled=0, rar_creds_test_enc=NULL,
+               rar_prod_enabled=1, rar_creds_prod_enc='P',
+               rar_env_default='prod'
+           WHERE id=1"""
+    )
+    conn.commit()
+
+
+def test_cerere_castiga(conn):
+    """Cererea explicita bate default-ul contului (prod)."""
+    _seteaza_cont_ambele(conn)
+    from app.rar_env import rezolva_rar_env
+    rezultat = rezolva_rar_env(conn, 1, "test")
+    assert rezultat == "test"
+
+
+def test_fallback_default_cont(conn):
+    """Fara cerere explicita -> default-ul contului (prod)."""
+    _seteaza_cont_ambele(conn)
+    from app.rar_env import rezolva_rar_env
+    rezultat = rezolva_rar_env(conn, 1, None)
+    assert rezultat == "prod"
+
+
+def test_tinta_indisponibila_respinsa(conn):
+    """Cerere pentru 'test' pe un cont doar-prod -> MediuIndisponibil cu .disponibile=['prod']."""
+    _seteaza_cont_doar_prod(conn)
+    from app.rar_env import MediuIndisponibil, rezolva_rar_env
+    with pytest.raises(MediuIndisponibil) as exc_info:
+        rezolva_rar_env(conn, 1, "test")
+    err = exc_info.value
+    assert err.env == "test"
+    assert err.disponibile == ["prod"]
+
+
+def test_valoare_invalida(conn):
+    """Cerere cu valoare in afara VALID_ENVS -> ValueError, fara fallback silentios."""
+    from app.rar_env import rezolva_rar_env
+    with pytest.raises(ValueError, match="mediu invalid"):
+        rezolva_rar_env(conn, 1, "staging")
+
+
+def test_zero_medii_cade_pe_ancora(conn, monkeypatch):
+    """Cont fara niciun mediu disponibil -> ancora globala AUTOPASS_RAR_ENV."""
+    # id=1 din fresh DB: rar_prod_enabled=1 dar rar_creds_prod_enc=NULL -> 0 disponibile
+    # (valoarea implicita a schemei: prod enabled fara creds -> nedisponibil)
+    monkeypatch.setenv("AUTOPASS_RAR_ENV", "test")
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.rar_env import rezolva_rar_env
+    rezultat = rezolva_rar_env(conn, 1, None)
+    assert rezultat == "test"
diff --git a/tests/test_t1_creds_durabile.py b/tests/test_t1_creds_durabile.py
index 505ccc5..d99b5ac 100644
--- a/tests/test_t1_creds_durabile.py
+++ b/tests/test_t1_creds_durabile.py
@@ -39,7 +39,7 @@ def env(monkeypatch):
 class FakeRar:
     """Stub RarClient pentru teste."""
 
-    def __init__(self, settings=None):
+    def __init__(self, settings=None, *, base_url=None):
         self.login_calls = 0
         self.closed = False
 
diff --git a/tests/test_worker_keepalive_rar.py b/tests/test_worker_keepalive_rar.py
index 5df1909..d87e69f 100644
--- a/tests/test_worker_keepalive_rar.py
+++ b/tests/test_worker_keepalive_rar.py
@@ -42,10 +42,10 @@ class _FakeSessions:
         self.invalidated: list[int] = []
         self.tokens: list[int] = []
 
-    def invalidate(self, account_id: int) -> None:
+    def invalidate(self, account_id: int, rar_env=None) -> None:
         self.invalidated.append(account_id)
 
-    def get_token(self, conn, account_id: int, creds) -> str | None:
+    def get_token(self, conn, account_id: int, creds, rar_env="test") -> str | None:
         self.tokens.append(account_id)
         if self._fail:
             raise RuntimeError("RAR jos")
diff --git a/tests/test_worker_observ.py b/tests/test_worker_observ.py
index 45cb3b4..8ba0ded 100644
--- a/tests/test_worker_observ.py
+++ b/tests/test_worker_observ.py
@@ -63,7 +63,7 @@ class FakeRar:
 def test_login_reusit_logat(env, monkeypatch):
     conn, settings = env
     import app.worker.__main__ as w
-    monkeypatch.setattr(w, "RarClient", lambda s: FakeRar())
+    monkeypatch.setattr(w, "RarClient", lambda s, **kw: FakeRar())
     sessions = w.AccountSessions(settings)
     tok = sessions.get_token(conn, 2, {"email": "a@b.ro", "password": "secretaXY"})
     assert tok == "JWT-TEST"
@@ -80,7 +80,7 @@ def test_login_reusit_logat(env, monkeypatch):
 def test_login_401_logat_fara_parola(env, monkeypatch):
     conn, settings = env
     import app.worker.__main__ as w
-    monkeypatch.setattr(w, "RarClient", lambda s: FakeRar(login_exc=RarAuthError("401", status_code=401)))
+    monkeypatch.setattr(w, "RarClient", lambda s, **kw: FakeRar(login_exc=RarAuthError("401", status_code=401)))
     sessions = w.AccountSessions(settings)
     with pytest.raises(RarAuthError):
         sessions.get_token(conn, 3, {"email": "a@b.ro", "password": "parolaGRESITA"})
diff --git a/tests/test_worker_rar_env.py b/tests/test_worker_rar_env.py
new file mode 100644
index 0000000..4c17f5a
--- /dev/null
+++ b/tests/test_worker_rar_env.py
@@ -0,0 +1,326 @@
+"""Teste US-006 (PRD 5.20) — sesiuni si trimitere worker per (cont, env).
+
+Verifica:
+- AccountSessions re-cheiat pe (account_id, rar_env): doua env ale aceluiasi cont
+  au sesiuni distincte.
+- RarClient creat cu base_url-ul mediului (test -> rar_base_url_test,
+  prod -> rar_base_url_prod), nu ancora globala.
+- Creds extrase din slotul accounts.rar_creds_{env}_enc corect per env.
+- Purjarea creds efemere scoped pe (account_id, rar_env): login pe test NU sterge
+  creds efemere ale submission-urilor PROD ale aceluiasi cont (auto-fix E1/1a).
+- recover_orphans per (cont, env): orfanii prod reconciliati contra endpoint prod,
+  nu contra test (auto-fix 1b/E6).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+
+import pytest
+from cryptography.fernet import Fernet
+
+
+# ---------------------------------------------------------------------------
+# Fixture DB
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def env(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
+    monkeypatch.setenv("AUTOPASS_CREDS_KEY", Fernet.generate_key().decode())
+    monkeypatch.setenv("AUTOPASS_WORKER_USE_TEST_CREDS", "false")
+    monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
+    from app.config import get_settings
+    from app import crypto
+    get_settings.cache_clear()
+    crypto.reset_cache()
+    from app.db import get_connection, init_db
+    init_db()
+    conn = get_connection()
+    settings = get_settings()
+    yield conn, settings
+    conn.close()
+    get_settings.cache_clear()
+    crypto.reset_cache()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_CONTENT = {
+    "vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
+    "data_prestatie": "2026-06-15", "odometru_final": "123456",
+    "prestatii": [{"cod_prestatie": "OE-1"}], "sistem_reparat": "null",
+}
+
+
+def _insert_sub(conn, account_id=1, rar_env="test", creds_enc=None, status="queued"):
+    """Insereaza un submission cu env si creds explicite."""
+    content = _CONTENT.copy()
+    cur = conn.execute(
+        "INSERT INTO submissions "
+        "(idempotency_key, account_id, status, payload_json, rar_env, rar_creds_enc) "
+        "VALUES (?, ?, ?, ?, ?, ?)",
+        (f"k-{os.urandom(4).hex()}", account_id, status, json.dumps(content), rar_env, creds_enc),
+    )
+    return int(cur.lastrowid)
+
+
+def _row(conn, sid):
+    return conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone()
+
+
+# Captura base_url-urilor clientilor creati de AccountSessions
+_created_clients: list = []
+
+
+class FakeRarClient:
+    """RarClient stub care captura base_url-ul pentru assertii."""
+
+    def __init__(self, settings=None, *, base_url=None, login_exc=None):
+        self.base_url = base_url
+        self._login_exc = login_exc
+        self.login_calls = 0
+        self.closed = False
+        _created_clients.append(self)
+
+    def login(self, email, password):
+        self.login_calls += 1
+        if self._login_exc:
+            raise self._login_exc
+        return f"TOK-{email}-{self.base_url}"
+
+    def get_nomenclator(self, token):
+        return []
+
+    def close(self):
+        self.closed = True
+
+
+# ---------------------------------------------------------------------------
+# test_sesiune_separata_per_env
+# ---------------------------------------------------------------------------
+
+def test_sesiune_separata_per_env(env, monkeypatch):
+    """Doua submission-uri ale aceluiasi cont, env test + prod -> doua login-uri distincte.
+
+    Cheia sesiunii e (account_id, rar_env): sesiunile test si prod sunt independente.
+    """
+    import app.worker.__main__ as w
+
+    _created_clients.clear()
+    monkeypatch.setattr(w, "RarClient", FakeRarClient)
+
+    conn, settings = env
+    # Cont secundar (contul 1 e default din schema)
+    conn.execute("INSERT INTO accounts (id, name) VALUES (2, 'Cont2')")
+    conn.commit()
+
+    sessions = w.AccountSessions(settings)
+
+    creds_test = {"email": "test@example.ro", "password": "ptest"}
+    creds_prod = {"email": "prod@example.ro", "password": "pprod"}
+
+    tok_test = sessions.get_token(conn, 2, creds_test, "test")
+    tok_prod = sessions.get_token(conn, 2, creds_prod, "prod")
+
+    # Doua login-uri distincte
+    assert len(_created_clients) == 2
+    assert _created_clients[0].login_calls == 1
+    assert _created_clients[1].login_calls == 1
+
+    # Tokenuri distincte (de la email-uri diferite)
+    assert tok_test != tok_prod
+
+    # Sesiunile active: doua intrari, ambele pt cont 2, env diferite
+    active = sessions.active()
+    assert len(active) == 2
+    envs_active = {env for _, env, _, _ in active}
+    assert envs_active == {"test", "prod"}
+
+    # Al doilea apel cu acelasi (cont, env) -> cache, NU re-login
+    tok_test2 = sessions.get_token(conn, 2, creds_test, "test")
+    assert tok_test2 == tok_test
+    assert len(_created_clients) == 2  # niciun client nou creat
+
+
+# ---------------------------------------------------------------------------
+# test_base_url_dupa_submission
+# ---------------------------------------------------------------------------
+
+def test_base_url_dupa_submission(env, monkeypatch):
+    """Un submission prod foloseste rar_base_url_prod; un submission test foloseste rar_base_url_test."""
+    import app.worker.__main__ as w
+
+    _created_clients.clear()
+    monkeypatch.setattr(w, "RarClient", FakeRarClient)
+
+    conn, settings = env
+    sessions = w.AccountSessions(settings)
+
+    creds = {"email": "x@example.ro", "password": "pw"}
+    sessions.get_token(conn, 1, creds, "test")
+    sessions.get_token(conn, 1, creds, "prod")
+
+    urls = {c.base_url for c in _created_clients}
+    assert settings.rar_base_url_test in urls, f"URL test asteptat in {urls}"
+    assert settings.rar_base_url_prod in urls, f"URL prod asteptat in {urls}"
+    # Cele doua URL-uri trebuie sa fie diferite (sisteme RAR separate)
+    assert settings.rar_base_url_test != settings.rar_base_url_prod
+
+
+# ---------------------------------------------------------------------------
+# test_creds_din_slotul_env
+# ---------------------------------------------------------------------------
+
+def test_creds_din_slotul_env(env, monkeypatch):
+    """Cand submissions.rar_creds_enc lipseste, worker ia din accounts.rar_creds_{env}_enc.
+
+    Prod ia din rar_creds_prod_enc, nu din slotul test (auto-fix 1c/E8 + fallback per-env).
+    """
+    import app.worker.__main__ as w
+    from app.crypto import encrypt_creds
+
+    conn, settings = env
+
+    enc_test = encrypt_creds({"email": "test@rar.ro", "password": "ptest"})
+    enc_prod = encrypt_creds({"email": "prod@rar.ro", "password": "pprod"})
+
+    # Salveaza creds in ambele sloturi per-env
+    conn.execute(
+        "UPDATE accounts SET rar_creds_test_enc=?, rar_creds_prod_enc=? WHERE id=1",
+        (enc_test, enc_prod),
+    )
+    conn.commit()
+
+    # Fara creds efemere pe submission -> fallback la slotul per-env
+    creds_test = w._creds_from_account(conn, 1, "test")
+    creds_prod = w._creds_from_account(conn, 1, "prod")
+
+    assert creds_test is not None, "slotul test trebuia sa aiba creds"
+    assert creds_test["email"] == "test@rar.ro"
+
+    assert creds_prod is not None, "slotul prod trebuia sa aiba creds"
+    assert creds_prod["email"] == "prod@rar.ro"
+
+    # Crucialmente: prod NU ia creds din slotul test
+    assert creds_prod["email"] != creds_test["email"]
+
+
+# ---------------------------------------------------------------------------
+# test_purge_creds_doar_pe_env (auto-fix E1/1a)
+# ---------------------------------------------------------------------------
+
+def test_purge_creds_doar_pe_env(env, monkeypatch):
+    """Dupa login pe env=test, creds efemere ale submission-urilor PROD raman neatinse.
+
+    Scopul purjarii: WHERE account_id=? AND rar_env=?. Altfel un login TEST sterge
+    creds ale submission-urilor PROD -> prod blocat (auto-fix E1/1a).
+    """
+    import app.worker.__main__ as w
+    from app.crypto import encrypt_creds
+
+    _created_clients.clear()
+    monkeypatch.setattr(w, "RarClient", FakeRarClient)
+
+    conn, settings = env
+
+    enc = encrypt_creds({"email": "u@rar.ro", "password": "pw"})
+
+    # Doua submission-uri ale aceluiasi cont: unul test, unul prod (ambele cu creds efemere)
+    sid_test = _insert_sub(conn, account_id=1, rar_env="test", creds_enc=enc)
+    sid_prod = _insert_sub(conn, account_id=1, rar_env="prod", creds_enc=enc)
+
+    sessions = w.AccountSessions(settings)
+
+    # Login pe env=test
+    sessions.get_token(conn, 1, {"email": "u@rar.ro", "password": "pw"}, "test")
+
+    # Creds efemere ale submission-ului TEST trebuie sterse (purjare normala)
+    row_test = _row(conn, sid_test)
+    assert row_test["rar_creds_enc"] is None, "creds test trebuiau sterse dupa login test"
+
+    # Creds efemere ale submission-ului PROD trebuie PASTRATE (nu sunt pentru env=test)
+    row_prod = _row(conn, sid_prod)
+    assert row_prod["rar_creds_enc"] is not None, \
+        "creds prod NU trebuiau sterse la login test (auto-fix E1/1a)"
+
+
+# ---------------------------------------------------------------------------
+# test_reconcile_pe_env_corect (auto-fix 1b/E6)
+# ---------------------------------------------------------------------------
+
+def test_reconcile_pe_env_corect(env, monkeypatch):
+    """Un orfan env=prod e reconciliat contra endpoint PROD, nu contra test.
+
+    auto-fix 1b/E6: recover_orphans filtreaza pe rar_env si foloseste clientul/token-ul
+    env-ului corect. Orfanii prod contra endpoint test -> no-match -> re-POST prod =
+    DUPLICAT real ireversibil.
+    """
+    import app.worker.__main__ as w
+
+    conn, settings = env
+
+    # Submission prod orfan (sending de mult timp)
+    sid_prod = _insert_sub(conn, account_id=1, rar_env="prod", status="sending")
+    conn.execute(
+        "UPDATE submissions SET sending_since=datetime('now', '-1 hour') WHERE id=?", (sid_prod,)
+    )
+    conn.commit()
+
+    # Submission test orfan (de verificat ca NU e atins de recover_orphans(rar_env='prod'))
+    sid_test = _insert_sub(conn, account_id=1, rar_env="test", status="sending")
+    conn.execute(
+        "UPDATE submissions SET sending_since=datetime('now', '-1 hour') WHERE id=?", (sid_test,)
+    )
+    conn.commit()
+
+    # Clientul prod fake — "gaseste" prezentarea prod la RAR
+    class FakeProdRar:
+        def __init__(self):
+            self.get_finalizate_calls = 0
+
+        def get_finalizate(self, token):
+            self.get_finalizate_calls += 1
+            return [{"id": 9999, "vin": "WVWZZZ1KZAW000123",
+                     "dataPrestatie": "2026-06-15", "odometruFinal": 123456}]
+
+        def post_prezentare(self, token, payload):
+            return {"id": 9999}
+
+    # Clientul test fake — nu gaseste nimic (sistemul test nu are prezentarea)
+    class FakeTestRar:
+        def __init__(self):
+            self.get_finalizate_calls = 0
+
+        def get_finalizate(self, token):
+            self.get_finalizate_calls += 1
+            return []  # nu e la RAR test
+
+        def post_prezentare(self, token, payload):
+            return {"id": 1111}
+
+    rar_prod = FakeProdRar()
+    rar_test = FakeTestRar()
+
+    # Apelam recover_orphans cu clientul PROD si env='prod' -> trebuie sa gaseasca orfanul prod
+    n = w.recover_orphans(conn, settings, rar_prod, "tok-prod", account_id=1, rar_env="prod")
+    assert n == 1, f"trebuia sa reconcilieze 1 orfan prod, a gasit {n}"
+
+    row_prod = _row(conn, sid_prod)
+    assert row_prod["status"] == "sent", f"orfanul prod trebuia marcat sent, e {row_prod['status']}"
+    assert row_prod["id_prezentare"] == 9999
+
+    # Submission-ul TEST nu trebuia atins de recover_orphans cu rar_env='prod'
+    row_test = _row(conn, sid_test)
+    assert row_test["status"] == "sending", \
+        f"orfanul test NU trebuia atins de recover cu env=prod, e {row_test['status']}"
+
+    # Confirmare ca clientul prod a interogat finalizate (reconciliere pe endpoint corect)
+    assert rar_prod.get_finalizate_calls == 1
+    # Clientul test NU trebuia folosit (recover_orphans cu env=prod NU atinge endpoint test)
+    assert rar_test.get_finalizate_calls == 0