feat(5.19): auto-send toggle per cont + tinere manuala randuri (held)

Comutator accounts.auto_send_enabled per cont: Auto OFF (default) tine randurile
la ingestie (submissions.held=1), worker-ul (claim_one AND held=0) le sare pana la
eliberare umana (per rand/bulk/auto-release OFF->ON). Snapshot held prin chokepoint
unic held_for_account pe toate caile de ingestie (API, import, reresolve, reactivare).

- schema/migrare: coloana held + index partial idx_submissions_held; auto_send_enabled
- API: echo onest held+motiv (US-010), ruta /prezentari/{id}/trimite-acum
- web: toggle header, modal confirmare tipata, buton Trimite per rand + Trimite toate,
  banner coada tinuta imbatranita (L.142), contor "In asteptare (manual)"
- worker: expire_held (US-008, inchide gaura retentie PII), metrics held gauges
- ops: tools/carantina_held + runbook rollback (R4)

Nota review (/code-review high): re-snapshot held lipseste pe caile repune/corectie
(requeue_submission, post_corectie, bulk-fix) — de aliniat separat cu create_prezentari.

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

tests/conftest.py

@@ -28,6 +28,69 @@ os.environ.setdefault("AUTOPASS_EMBEDDINGS_ENABLED", "false")
 os.environ.setdefault("AUTOPASS_SEED_OPERATII_ENABLED", "false")
 
 
+# Referinta la init_db-ul REAL (nemodificat), capturata O SINGURA DATA la prima rulare a
+# fixturii de mai jos. Wrapper-ul o refoloseste ca sa nu se auto-imbrace recursiv.
+_REAL_INIT_DB = None
+
+
+@pytest.fixture(autouse=True)
+def _auto_send_on_default_account(request, monkeypatch):
+    """PRD 5.19 US-009: contul default id=1 (seed schema.sql, INSERT OR IGNORE) porneste pe
+    `auto_send_enabled=0` (Auto OFF) — corect in productie (nimic nu pleaca fara confirmare).
+
+    Dar suita exercita lantul `POST -> claim_one -> sent` pe contul id=1
+    (test_import_e2e, test_creds_delivery, test_live_rar): cu Auto OFF, ingestia ar marca
+    `held=1` (US-002) si worker-ul (`claim_one ... AND held=0`, US-003) nu ar mai lua randul
+    -> testele ar stagna tacit. E un fix de STARE DB, nu env var (coloana e per-rand in
+    `accounts`), deci il aplicam wraparind `init_db` (contul id=1 NU e creat de `create_account`
+    -> un patch pe factory NU l-ar acoperi — Eng HIGH).
+
+    Wrapam `init_db` astfel incat orice DB de test creat sa aiba id=1 pe Auto ON (held=0 la
+    ingestie), pastrand suita verde ca inainte de 5.19. Conturile create explicit de teste
+    (`create_account`) raman pe default OFF -> testele care VOR Auto OFF il primesc corect.
+
+    Testele care verifica DEFAULT-ul brut (id=1 == 0 imediat dupa init_db) se exclud cu
+    markerul `no_auto_send_seed`.
+    """
+    global _REAL_INIT_DB
+    import sys
+
+    import app.db as _db
+
+    if _REAL_INIT_DB is None:
+        _REAL_INIT_DB = _db.init_db
+
+    if request.node.get_closest_marker("no_auto_send_seed"):
+        yield
+        return
+
+    def _init_db_auto_on():
+        _REAL_INIT_DB()
+        conn = _db.get_connection()
+        try:
+            conn.execute("UPDATE accounts SET auto_send_enabled=1 WHERE id=1")
+            conn.commit()
+        finally:
+            conn.close()
+
+    # app.db.init_db: acopera fixturile care fac `from app.db import init_db` in corpul lor
+    # (importul ruleaza DUPA acest patch -> primeste wrapper-ul) + apelurile `app.db.init_db()`.
+    monkeypatch.setattr(_db, "init_db", _init_db_auto_on)
+    # app.main.init_db: lifespan-ul TestClient cheama numele importat la nivel de modul
+    # (`from .db import init_db`); daca app.main e deja importat, patch-uim si referinta lui.
+    # Daca NU e inca importat, primul `from app.main import app` din test il va lega la
+    # wrapper-ul de mai sus automat (citeste app.db.init_db la momentul importului).
+    _main = sys.modules.get("app.main")
+    if _main is not None:
+        monkeypatch.setattr(_main, "init_db", _init_db_auto_on, raising=False)
+    yield
+    # Anti-leak: daca app.main a fost importat IN timpul testului (legand wrapper-ul la
+    # import, nu prin monkeypatch), readu-l la init_db-ul real ca sa nu ramana wrapat global.
+    _main = sys.modules.get("app.main")
+    if _main is not None and getattr(_main, "init_db", None) is _init_db_auto_on:
+        _main.init_db = _REAL_INIT_DB
+
+
 @pytest.fixture(autouse=True)
 def _reset_embeddings_singleton():
     """Reseteaza singleton-ul global de embeddings intre teste (izolare de ordine).
@@ -69,3 +132,8 @@ def pytest_configure(config):
     config.addinivalue_line(
         "markers", "live: test live pe RAR test (necesita AUTOPASS_LIVE_RAR=1 + creds reale)"
     )
+    config.addinivalue_line(
+        "markers",
+        "no_auto_send_seed: dezactiveaza fixtura autouse care pune id=1 pe Auto ON "
+        "(teste care verifica default-ul brut auto_send_enabled=0 al contului id=1)",
+    )

tests/test_auto_send_schema.py

@@ -0,0 +1,150 @@
+"""US-001 (PRD 5.19): schema comutator Auto per cont + flag `held` pe submission.
+
+Acopera: migrarea aditiva idempotenta (coloane + index partial pe DB vechi), helperii
+`get_auto_send`/`set_auto_send`/`held_for_account` din app/accounts.py si default-ul 0.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+
+import pytest
+
+
+@pytest.fixture()
+def fresh_conn(monkeypatch):
+    """DB nou cu schema curenta (init_db)."""
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.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 _old_db(path: str) -> sqlite3.Connection:
+    """DB in forma PRE-5.19: accounts fara auto_send_enabled, submissions fara held."""
+    conn = sqlite3.connect(path, isolation_level=None)
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        "CREATE TABLE accounts (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, "
+        "cui TEXT)"
+    )
+    conn.execute(
+        "CREATE TABLE submissions (id INTEGER PRIMARY KEY AUTOINCREMENT, "
+        "idempotency_key TEXT NOT NULL UNIQUE, account_id INTEGER, status TEXT, "
+        "payload_json TEXT NOT NULL)"
+    )
+    return conn
+
+
+def _migrate_old(path: str, monkeypatch) -> sqlite3.Connection:
+    from app.config import get_settings
+    get_settings.cache_clear()
+    conn = sqlite3.connect(path, isolation_level=None)
+    conn.row_factory = sqlite3.Row
+    from app.db import _migrate
+    _migrate(conn)
+    return conn
+
+
+def test_migrare_adauga_coloane_si_index(tmp_path, monkeypatch):
+    """Pe DB vechi (fara coloane) migrarea adauga held, auto_send_enabled si indexul partial."""
+    path = str(tmp_path / "old.db")
+    _old_db(path).close()
+    conn = _migrate_old(path, monkeypatch)
+
+    sub = {r["name"] for r in conn.execute("PRAGMA table_info(submissions)").fetchall()}
+    assert "held" in sub
+    acc = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
+    assert "auto_send_enabled" in acc
+
+    idx = {r["name"] for r in conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='submissions'"
+    ).fetchall()}
+    assert "idx_submissions_held" in idx
+    conn.close()
+
+
+def test_migrare_idempotenta_a_doua_rulare(tmp_path, monkeypatch):
+    """A doua rulare _migrate pe DB deja migrat nu arunca si nu dubleaza nimic."""
+    path = str(tmp_path / "old.db")
+    _old_db(path).close()
+    conn = _migrate_old(path, monkeypatch)
+    from app.db import _migrate
+    _migrate(conn)  # nu arunca
+    acc = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
+    assert "auto_send_enabled" in acc
+    conn.close()
+
+
+def test_migrare_idempotenta_pe_db_fresh(fresh_conn):
+    """_migrate re-rulat pe schema curenta (init_db) nu strica nimic."""
+    from app.db import _migrate
+    _migrate(fresh_conn)
+    sub = {r["name"] for r in fresh_conn.execute("PRAGMA table_info(submissions)").fetchall()}
+    assert "held" in sub
+
+
+def test_held_for_account_auto_on(fresh_conn):
+    from app.accounts import create_account, held_for_account, set_auto_send
+    aid = create_account(fresh_conn, "Auto ON")
+    set_auto_send(fresh_conn, aid, True)
+    assert held_for_account(fresh_conn, aid) == 0
+
+
+def test_held_for_account_auto_off(fresh_conn):
+    from app.accounts import create_account, held_for_account
+    aid = create_account(fresh_conn, "Auto OFF")
+    assert held_for_account(fresh_conn, aid) == 1
+
+
+def test_held_for_account_cont_inexistent(fresh_conn):
+    from app.accounts import held_for_account
+    assert held_for_account(fresh_conn, 99999) == 1
+
+
+def test_get_set_auto_send_round_trip(fresh_conn):
+    from app.accounts import create_account, get_auto_send, set_auto_send
+    aid = create_account(fresh_conn, "Round trip")
+    assert get_auto_send(fresh_conn, aid) is False
+    set_auto_send(fresh_conn, aid, True)
+    assert get_auto_send(fresh_conn, aid) is True
+    # Idempotent: acelasi set nu arunca si pastreaza valoarea.
+    set_auto_send(fresh_conn, aid, True)
+    assert get_auto_send(fresh_conn, aid) is True
+    set_auto_send(fresh_conn, aid, False)
+    assert get_auto_send(fresh_conn, aid) is False
+
+
+def test_get_auto_send_cont_inexistent(fresh_conn):
+    from app.accounts import get_auto_send
+    assert get_auto_send(fresh_conn, 99999) is False
+
+
+@pytest.mark.no_auto_send_seed  # verifica default-ul BRUT id=1==0; nu wrapa init_db (conftest)
+def test_default_auto_send_zero(fresh_conn):
+    """Cont nou (create_account) si contul id=1 pornesc pe auto_send_enabled=0 (Auto OFF)."""
+    from app.accounts import create_account
+    aid = create_account(fresh_conn, "Nou")
+    row = fresh_conn.execute(
+        "SELECT auto_send_enabled FROM accounts WHERE id=?", (aid,)
+    ).fetchone()
+    assert row["auto_send_enabled"] == 0
+    row1 = fresh_conn.execute(
+        "SELECT auto_send_enabled FROM accounts WHERE id=1"
+    ).fetchone()
+    assert row1["auto_send_enabled"] == 0
+
+
+def test_config_held_zile():
+    from app.config import Settings
+    s = Settings()
+    assert s.held_warn_days == 7
+    assert s.held_retention_days == 90

tests/test_carantina_held.py

@@ -0,0 +1,99 @@
+"""Teste PRD 5.19 R4 — tools/carantina_held.py (atenuare hazard de rollback).
+
+Carantineaza randurile `queued AND held=1` (-> `error`/`ROLLBACK_QUARANTINE`)
+ca sa nu fie trimise de un worker fara filtrul `AND held=0` dupa un revert.
+`--dry-run` doar numara, fara sa scrie.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+
+import pytest
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.db import get_connection, init_db
+    init_db()
+    conn = get_connection()
+    yield conn
+    conn.close()
+    get_settings.cache_clear()
+
+
+_CONTENT = {
+    "vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
+    "data_prestatie": "2026-06-15", "odometru_final": "123456",
+    "prestatii": [{"cod_prestatie": "OE-1"}],
+}
+
+
+def _insert(conn, status="queued", held=0):
+    cur = conn.execute(
+        "INSERT INTO submissions (idempotency_key, status, payload_json, account_id, held) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (f"key-{os.urandom(4).hex()}", status, json.dumps(_CONTENT), 1, held),
+    )
+    return int(cur.lastrowid)
+
+
+def _row(conn, sid):
+    return conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone()
+
+
+def _run(argv):
+    from tools.carantina_held import main
+    return main(argv)
+
+
+def test_carantineaza_rand_tinut(env, capsys):
+    conn = env
+    sid = _insert(conn, status="queued", held=1)
+    rc = _run([])
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "1 randuri carantinate" in out
+    r = _row(conn, sid)
+    assert r["status"] == "error"
+    assert r["rar_error"] == "ROLLBACK_QUARANTINE"
+
+
+def test_dry_run_nu_modifica(env, capsys):
+    conn = env
+    sid = _insert(conn, status="queued", held=1)
+    rc = _run(["--dry-run"])
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "dry-run" in out
+    assert "1 randuri" in out
+    r = _row(conn, sid)
+    assert r["status"] == "queued"  # neschimbat
+    assert r["rar_error"] is None
+
+
+def test_nu_atinge_randuri_ne_tinute_sau_ne_queued(env, capsys):
+    conn = env
+    sid_libera = _insert(conn, status="queued", held=0)
+    sid_sent = _insert(conn, status="sent", held=1)  # held dar deja plecat
+    sid_tinuta = _insert(conn, status="queued", held=1)
+    rc = _run([])
+    capsys.readouterr()
+    assert rc == 0
+    assert _row(conn, sid_libera)["status"] == "queued"
+    assert _row(conn, sid_sent)["status"] == "sent"
+    assert _row(conn, sid_tinuta)["status"] == "error"
+
+
+def test_fara_randuri_tinute(env, capsys):
+    _insert(env, status="queued", held=0)
+    rc = _run([])
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "Niciun rand tinut" in out

tests/test_held_api_echo.py

@@ -0,0 +1,146 @@
+"""Teste PRD 5.19 US-010 — onestitate + observabilitate `held` pe canalul API.
+
+- `held` in raspunsul enqueue (SubmissionResult) + motiv non-null.
+- `held` in proiectiile GET /v1/prezentari si /v1/prezentari/{id}.
+- endpoint de eliberare API POST /v1/prezentari/{id}/trimite-acum (paritate cu /repune).
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "echo.db"))
+    monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
+    monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.main import app
+    with TestClient(app) as c:
+        yield c
+    get_settings.cache_clear()
+
+
+def _set_auto_send(enabled: bool, account_id: int = 1) -> None:
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        conn.execute(
+            "UPDATE accounts SET auto_send_enabled=? WHERE id=?",
+            (1 if enabled else 0, account_id),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _held(sid: int) -> int:
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        return conn.execute("SELECT held FROM submissions WHERE id=?", (sid,)).fetchone()["held"]
+    finally:
+        conn.close()
+
+
+def _body(**over) -> dict:
+    prez = {
+        "vin": "WVWZZZ1KZAW000123",
+        "nr_inmatriculare": "B999TST",
+        "data_prestatie": "2026-06-15",
+        "odometru_final": "123456",
+        "prestatii": [{"cod_prestatie": "OE-1"}],
+    }
+    prez.update(over)
+    return {"prezentari": [prez]}
+
+
+def _enqueue_held(client) -> int:
+    _set_auto_send(False)
+    r = client.post("/v1/prezentari", json=_body())
+    return r.json()["results"][0]["submission_id"]
+
+
+# --------------------------------------------------------------------------- #
+# GET proiectii                                                               #
+# --------------------------------------------------------------------------- #
+
+def test_get_prezentare_expune_held(client):
+    sid = _enqueue_held(client)
+    r = client.get(f"/v1/prezentari/{sid}")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert "held" in body
+    assert body["held"] in (1, True)
+
+
+def test_list_prezentari_expune_held(client):
+    sid = _enqueue_held(client)
+    r = client.get("/v1/prezentari")
+    assert r.status_code == 200, r.text
+    rows = {row["id"]: row for row in r.json()["submissions"]}
+    assert sid in rows
+    assert "held" in rows[sid]
+    assert rows[sid]["held"] in (1, True)
+
+
+# --------------------------------------------------------------------------- #
+# POST /v1/prezentari/{id}/trimite-acum                                       #
+# --------------------------------------------------------------------------- #
+
+def test_trimite_acum_elibereaza(client):
+    """Rand tinut -> trimite-acum -> held=0."""
+    sid = _enqueue_held(client)
+    assert _held(sid) == 1
+    r = client.post(f"/v1/prezentari/{sid}/trimite-acum")
+    assert r.status_code == 200, r.text
+    assert r.json()["ok"] is True
+    assert _held(sid) == 0
+
+
+def test_trimite_acum_id_inexistent_404(client):
+    r = client.post("/v1/prezentari/999999/trimite-acum")
+    assert r.status_code == 404
+
+
+def test_trimite_acum_rand_nequeued_noop(client):
+    """Rand non-queued (sent) -> no-op sigur, ramane neschimbat, raspuns 200."""
+    _set_auto_send(True)
+    sid = client.post("/v1/prezentari", json=_body()).json()["results"][0]["submission_id"]
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        conn.execute("UPDATE submissions SET status='sent', held=1 WHERE id=?", (sid,))
+        conn.commit()
+    finally:
+        conn.close()
+    r = client.post(f"/v1/prezentari/{sid}/trimite-acum")
+    assert r.status_code == 200, r.text
+    # held nemodificat (nu era queued)
+    assert _held(sid) == 1
+
+
+def test_trimite_acum_cont_strain_404(client):
+    """Un rand al altui cont -> 404-before-leak (nu elibereaza cross-account)."""
+    sid = _enqueue_held(client)
+    from app.accounts import create_account
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        # Cont real strain (id>=2) + muta randul pe el (FK valid).
+        other = create_account(conn, "Alt cont")
+        conn.execute("UPDATE submissions SET account_id=? WHERE id=?", (other, sid))
+        conn.commit()
+    finally:
+        conn.close()
+    # Clientul (dev, fara cheie) e pe contul id=1 -> randul altui cont e 404-before-leak.
+    r = client.post(f"/v1/prezentari/{sid}/trimite-acum")
+    assert r.status_code == 404
+    assert _held(sid) == 1

tests/test_held_ingestie.py

@@ -0,0 +1,259 @@
+"""Teste PRD 5.19 US-002 — snapshot `held` la ingestie din comutatorul de cont.
+
+Acopera TOATE situ-rile care scriu `status='queued'` prin ingestie (chokepoint
+`held_for_account`): enqueue API, reactivare error->queued, dedup echo, commit import,
+reresolve `needs_mapping`. `held` NU intra in idempotency_key.
+
+Contul default id=1 e fortat pe Auto ON de fixtura autouse din conftest; testele care
+vor Auto OFF il pun explicit pe 0 (`_set_auto_send(client, False)`).
+"""
+
+from __future__ import annotations
+
+import io
+import os
+import tempfile
+
+import openpyxl
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "held.db"))
+    monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
+    monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.main import app
+    with TestClient(app) as c:
+        yield c
+    get_settings.cache_clear()
+
+
+def _set_auto_send(enabled: bool, account_id: int = 1) -> None:
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        conn.execute(
+            "UPDATE accounts SET auto_send_enabled=? WHERE id=?",
+            (1 if enabled else 0, account_id),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(sid: int) -> dict:
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        r = conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone()
+        return dict(r) if r else {}
+    finally:
+        conn.close()
+
+
+def _force_status(sid: int, status: str) -> None:
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        conn.execute("UPDATE submissions SET status=? WHERE id=?", (status, sid))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _body(**over) -> dict:
+    prez = {
+        "vin": "WVWZZZ1KZAW000123",
+        "nr_inmatriculare": "B999TST",
+        "data_prestatie": "2026-06-15",
+        "odometru_final": "123456",
+        "prestatii": [{"cod_prestatie": "OE-1"}],
+    }
+    prez.update(over)
+    return {"prezentari": [prez]}
+
+
+# --------------------------------------------------------------------------- #
+# Enqueue API                                                                 #
+# --------------------------------------------------------------------------- #
+
+def test_auto_off_ingestie_held(client):
+    """Cont Auto OFF -> POST valid -> rand queued, held=1; raspuns held=true + motiv non-null."""
+    _set_auto_send(False)
+    r = client.post("/v1/prezentari", json=_body())
+    assert r.status_code == 200, r.text
+    res = r.json()["results"][0]
+    assert res["status"] == "queued"
+    assert res["held"] is True
+    assert res["motiv"], "motiv trebuie sa fie non-null cand randul e tinut"
+    assert _row(res["submission_id"])["held"] == 1
+
+
+def test_auto_on_ingestie_not_held(client):
+    """Cont Auto ON -> POST valid -> rand queued, held=0; raspuns held=false, motiv None."""
+    _set_auto_send(True)
+    r = client.post("/v1/prezentari", json=_body())
+    assert r.status_code == 200, r.text
+    res = r.json()["results"][0]
+    assert res["status"] == "queued"
+    assert res["held"] is False
+    assert res["motiv"] is None
+    assert _row(res["submission_id"])["held"] == 0
+
+
+# --------------------------------------------------------------------------- #
+# Reactivare error->queued (bug de bypass Eng Finding A)                       #
+# --------------------------------------------------------------------------- #
+
+def test_reactivare_respecta_auto_off(client):
+    """Un rand `error` re-POST-at pe cont Auto OFF re-intra queued CU held=1 (nu pastreaza vechiul)."""
+    _set_auto_send(True)
+    r = client.post("/v1/prezentari", json=_body())
+    sid = r.json()["results"][0]["submission_id"]
+    assert _row(sid)["held"] == 0
+    # Contul trece pe OFF si randul cade in error; re-POST-ul trebuie sa-l tina.
+    _set_auto_send(False)
+    _force_status(sid, "error")
+    r2 = client.post("/v1/prezentari", json=_body())
+    res = r2.json()["results"][0]
+    assert res["submission_id"] == sid
+    assert res["reactivated"] is True
+    assert res["held"] is True
+    assert res["motiv"]
+    assert _row(sid)["held"] == 1
+
+
+# --------------------------------------------------------------------------- #
+# Dedup echo                                                                   #
+# --------------------------------------------------------------------------- #
+
+def test_dedup_echo_held(client):
+    """Re-POST acelasi continut (Auto OFF) -> deduped, dar held propagat corect (nu queued curat fals)."""
+    _set_auto_send(False)
+    r1 = client.post("/v1/prezentari", json=_body())
+    sid = r1.json()["results"][0]["submission_id"]
+    r2 = client.post("/v1/prezentari", json=_body())
+    res = r2.json()["results"][0]
+    assert res["submission_id"] == sid
+    assert res["deduped"] is True
+    assert res["held"] is True
+    assert res["motiv"]
+
+
+# --------------------------------------------------------------------------- #
+# Idempotenta: held nu schimba cheia                                           #
+# --------------------------------------------------------------------------- #
+
+def test_held_nu_schimba_idempotency_key(client):
+    """Acelasi continut Auto ON vs OFF -> aceeasi idempotency_key (held e pur coada)."""
+    _set_auto_send(True)
+    sid_on = client.post("/v1/prezentari", json=_body()).json()["results"][0]["submission_id"]
+    key_on = _row(sid_on)["idempotency_key"]
+    # Simulam un al doilea DB logic: acelasi continut, cont OFF -> dedup pe aceeasi cheie.
+    _set_auto_send(False)
+    r2 = client.post("/v1/prezentari", json=_body())
+    res2 = r2.json()["results"][0]
+    assert res2["deduped"] is True
+    assert res2["submission_id"] == sid_on
+    assert _row(res2["submission_id"])["idempotency_key"] == key_on
+
+
+# --------------------------------------------------------------------------- #
+# Reresolve needs_mapping -> queued                                           #
+# --------------------------------------------------------------------------- #
+
+def test_reresolve_held(client):
+    """Cont Auto OFF, submission needs_mapping -> reresolve dupa mapare -> queued cu held=1."""
+    _set_auto_send(False)
+    # Cod intern necunoscut -> needs_mapping (nu se trimite).
+    body = _body(prestatii=[{"cod_op_service": "REV-INTERN", "denumire": "Revizie interna"}])
+    r = client.post("/v1/prezentari", json=body)
+    res = r.json()["results"][0]
+    sid = res["submission_id"]
+    assert _row(sid)["status"] == "needs_mapping"
+    # Salvare mapare -> reresolve automat (POST /v1/mapari cheama reresolve_account).
+    rc = client.post("/v1/mapari", json={
+        "cod_op_service": "REV-INTERN", "cod_prestatie": "OE-1", "auto_send": True,
+    })
+    assert rc.status_code == 200, rc.text
+    row = _row(sid)
+    assert row["status"] == "queued"
+    assert row["held"] == 1
+
+
+def test_reresolve_held_auto_on(client):
+    """Cont Auto ON, reresolve -> queued cu held=0."""
+    _set_auto_send(True)
+    body = _body(prestatii=[{"cod_op_service": "REV-INTERN", "denumire": "Revizie interna"}])
+    sid = client.post("/v1/prezentari", json=body).json()["results"][0]["submission_id"]
+    assert _row(sid)["status"] == "needs_mapping"
+    client.post("/v1/mapari", json={
+        "cod_op_service": "REV-INTERN", "cod_prestatie": "OE-1", "auto_send": True,
+    })
+    row = _row(sid)
+    assert row["status"] == "queued"
+    assert row["held"] == 0
+
+
+# --------------------------------------------------------------------------- #
+# Commit import                                                                #
+# --------------------------------------------------------------------------- #
+
+_HEADER = ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"]
+_ROW = ["WVWZZZ1KZAW001111", "B100TST", "2026-06-15", "123456", "Revizie"]
+_COLMAP = {
+    "VIN": "vin", "Nr inmatriculare": "nr_inmatriculare",
+    "Data prestatie": "data_prestatie", "Odometru final": "odometru_final",
+    "Operatie": "operatie",
+}
+
+
+def _make_xlsx(rows: list[list]) -> bytes:
+    wb = openpyxl.Workbook()
+    ws = wb.active
+    ws.title = "Sheet1"
+    for row in rows:
+        ws.append(row)
+    buf = io.BytesIO()
+    wb.save(buf)
+    return buf.getvalue()
+
+
+def _commit_import(client) -> int:
+    data = _make_xlsx([_HEADER, _ROW])
+    import_id = client.post(
+        "/v1/import",
+        files={"file": ("t.xlsx", io.BytesIO(data), "application/octet-stream")},
+    ).json()["import_id"]
+    client.post(f"/v1/import/{import_id}/column-mapping", json={"json_mapare": _COLMAP})
+    client.post("/v1/mapari", json={"cod_op_service": "Revizie", "cod_prestatie": "OE-1", "auto_send": True})
+    rp = client.get(f"/v1/import/{import_id}/preview")
+    assert rp.status_code == 200 and rp.json()["summary"].get("ok", 0) == 1, rp.text
+    rc = client.post(f"/v1/import/{import_id}/commit", json={
+        "n_confirmat": 1, "reviewed_rows": [], "confirmed_by": "t@e2e.ro",
+    })
+    assert rc.status_code == 200, rc.text
+    return rc.json()["submissions"][0]["submission_id"]
+
+
+def test_import_commit_held_auto_off(client):
+    """Commit import pe cont Auto OFF -> rand queued cu held=1."""
+    _set_auto_send(False)
+    sid = _commit_import(client)
+    row = _row(sid)
+    assert row["status"] == "queued"
+    assert row["held"] == 1
+
+
+def test_import_commit_not_held_auto_on(client):
+    """Commit import pe cont Auto ON -> rand queued cu held=0."""
+    _set_auto_send(True)
+    sid = _commit_import(client)
+    row = _row(sid)
+    assert row["status"] == "queued"
+    assert row["held"] == 0

tests/test_labels_held.py

@@ -0,0 +1,88 @@
+"""
+Teste pentru eticheta randurilor tinute manual (held) — US-006, PRD 5.19.
+
+RED intai: scrise inainte de a extinde labels.py cu parametrul `held`.
+
+Regula de afisaj (D5, PRD 5.19):
+  - `status='queued' AND held=1` -> "In asteptare (manual)" + clasa CSS de
+    AVERTIZARE (amber, `--warn`) — asteptare benigna, NU eroare (rosu).
+  - `held=0` (sau lipsa) -> comportament neschimbat pentru TOATE starile.
+  - `held` NU e stare noua in masina de stari: CHECK-ul din schema ramane neatins.
+
+Fisiere atinse: app/web/labels.py, tests/test_labels_held.py.
+"""
+
+import pytest
+
+from app.web.labels import eticheta_scurta, eticheta_stare
+
+
+TEXT_HELD = "In asteptare (manual)"
+# Clasa CSS amber (--warn) reutilizata din base.html (.s-needs_review{color:var(--warn)}).
+# NU e o clasa needs_* rosie (s-needs_data / s-needs_mapping sunt --err).
+CLASA_WARN_HELD = "s-needs_review"
+
+# Toate starile non-queued: held nu trebuie sa le schimbe niciodata.
+STARI_NEAFECTATE = ["sending", "sent", "needs_mapping", "needs_data", "error"]
+
+
+# ---------------------------------------------------------------------------
+# eticheta_stare (text lung + subtext + clasa)
+# ---------------------------------------------------------------------------
+
+def test_queued_held_are_eticheta_manuala_si_clasa_warn():
+    text, subtext, css_class = eticheta_stare("queued", held=True)
+    assert text == TEXT_HELD, f"asteptam {TEXT_HELD!r}, got {text!r}"
+    assert css_class == CLASA_WARN_HELD, (
+        f"randul tinut trebuie amber (--warn) prin {CLASA_WARN_HELD!r}, got {css_class!r}"
+    )
+    # Nu trebuie sa fie o clasa de eroare rosie.
+    assert css_class not in ("s-error", "s-needs_data", "s-needs_mapping"), (
+        "held e asteptare benigna — nu trebuie sa foloseasca o clasa rosie de eroare"
+    )
+    assert isinstance(subtext, str)
+
+
+def test_queued_held_false_ramane_neschimbat():
+    """queued fara held == comportamentul dinainte de US-006 (PRD 5.19)."""
+    fara_held = eticheta_stare("queued")
+    held_false = eticheta_stare("queued", held=False)
+    assert fara_held == held_false, "held=False trebuie identic cu apelul fara held"
+    text, _subtext, css_class = held_false
+    assert "In asteptare" in text
+    assert text != TEXT_HELD, "queued normal NU trebuie sa afiseze varianta manuala"
+    assert css_class == "s-queued"
+
+
+@pytest.mark.parametrize("status", STARI_NEAFECTATE)
+def test_alte_stari_ignora_held(status):
+    """held nu are efect pe nicio stare in afara de queued."""
+    assert eticheta_stare(status, held=True) == eticheta_stare(status, held=False)
+    assert eticheta_stare(status, held=True) == eticheta_stare(status)
+
+
+def test_apel_pozitional_compat():
+    """Semnatura pastreaza compatibilitatea: apelul vechi cu un singur arg merge."""
+    assert eticheta_stare("sent") == eticheta_stare("sent", held=False)
+
+
+# ---------------------------------------------------------------------------
+# eticheta_scurta (pill) — pill-ul trebuie sa distinga held de queued normal (D5)
+# ---------------------------------------------------------------------------
+
+def test_pill_queued_held_diferit_de_queued_normal():
+    pill_held = eticheta_scurta("queued", held=True)
+    pill_normal = eticheta_scurta("queued", held=False)
+    assert pill_held != pill_normal, (
+        "pill-ul trebuie sa distinga randul tinut (D5: altfel randa 'In coada' identic)"
+    )
+    assert pill_normal == "In coada"
+
+
+def test_pill_alte_stari_ignora_held():
+    for status in STARI_NEAFECTATE:
+        assert eticheta_scurta(status, held=True) == eticheta_scurta(status, held=False)
+
+
+def test_pill_apel_pozitional_compat():
+    assert eticheta_scurta("queued") == eticheta_scurta("queued", held=False)

tests/test_metrics_held.py

@@ -0,0 +1,97 @@
+"""Teste PRD 5.19 US-007 — gauge-uri /metrics pentru coada tinuta.
+
+/metrics expune, pe langa `autopass_submissions{status=...}`, doua gauge-uri
+DERIVATE (zero stare noua), scoped global:
+  - autopass_held_submissions          = COUNT(queued AND held=1)
+  - autopass_held_oldest_age_seconds   = varsta (s) a celui mai vechi rand tinut
+
+TDD: fixtura izoleaza DB per test (pattern din test_worker_held.py).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.db import get_connection, init_db
+    init_db()
+    conn = get_connection()
+    yield conn
+    conn.close()
+    get_settings.cache_clear()
+
+
+def _client():
+    from app.main import app
+    return TestClient(app)
+
+
+_CONTENT = {
+    "vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
+    "data_prestatie": "2026-06-15", "odometru_final": "123456",
+    "prestatii": [{"cod_prestatie": "OE-1"}],
+}
+
+
+def _insert(conn, status="queued", held=0, created_at=None):
+    cur = conn.execute(
+        "INSERT INTO submissions (idempotency_key, status, payload_json, account_id, held) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (f"key-{os.urandom(4).hex()}", status, json.dumps(_CONTENT), 1, held),
+    )
+    sid = int(cur.lastrowid)
+    if created_at is not None:
+        conn.execute("UPDATE submissions SET created_at=? WHERE id=?", (created_at, sid))
+    return sid
+
+
+def _metric(text, name):
+    for line in text.splitlines():
+        if line.startswith(name + " "):
+            return line[len(name) + 1:].strip()
+    return None
+
+
+def test_held_gauges_numara_si_varsta(env):
+    conn = env
+    # doua randuri tinute (unul vechi de ~10 zile) + unul ne-tinut
+    _insert(conn, status="queued", held=1, created_at="2026-06-25 00:00:00")
+    _insert(conn, status="queued", held=1)
+    _insert(conn, status="queued", held=0)
+
+    body = _client().get("/metrics").text
+    assert _metric(body, "autopass_held_submissions") == "2"
+    oldest = int(_metric(body, "autopass_held_oldest_age_seconds"))
+    assert oldest > 0
+
+
+def test_held_gauge_zero_fara_randuri_tinute(env):
+    conn = env
+    _insert(conn, status="queued", held=0)
+    _insert(conn, status="sent", held=0)
+
+    body = _client().get("/metrics").text
+    assert _metric(body, "autopass_held_submissions") == "0"
+    # varsta = 0 cand nu exista randuri tinute (COALESCE)
+    assert _metric(body, "autopass_held_oldest_age_seconds") == "0"
+
+
+def test_held_gauge_ignora_tinut_ne_queued(env):
+    conn = env
+    # held=1 dar status != queued (ex. deja carantinat) NU se numara
+    _insert(conn, status="error", held=1)
+    _insert(conn, status="queued", held=1)
+
+    body = _client().get("/metrics").text
+    assert _metric(body, "autopass_held_submissions") == "1"

tests/test_web_auto_send.py

@@ -0,0 +1,347 @@
+"""Teste PRD 5.19 strat WEB: toggle "Trimite automat la RAR" (US-004), trimitere
+manuala per rand + bulk (US-005), afisaj held (US-006 UI), banner coada tinuta
+imbatranita + contor manual (US-007), audit app_events (US-009).
+
+Model: TestClient + login web (ca tests/test_web_lifecycle.py). NU atinge conftest.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import tempfile
+
+import pytest
+from starlette.testclient import TestClient
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "as.db"))
+    monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
+    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.web import ratelimit
+    ratelimit._hits.clear()
+    from app.main import app
+    with TestClient(app, follow_redirects=False) as c:
+        yield c
+    ratelimit._hits.clear()
+    get_settings.cache_clear()
+
+
+def _account_user(email, name="Service", password="parolasecreta10"):
+    from app.accounts import create_account
+    from app.users import create_user
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        aid = create_account(conn, name, active=True)
+        create_user(conn, aid, email, password)
+        return aid
+    finally:
+        conn.close()
+
+
+def _login(client, email, password="parolasecreta10"):
+    resp = client.get("/login")
+    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) or \
+        re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
+    resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": m.group(1)})
+    assert resp.status_code == 303, resp.text[:200]
+    return _csrf(client)
+
+
+def _csrf(client):
+    resp = client.get("/?tab=acasa")
+    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
+    assert m, "csrf_token negasit dupa login"
+    return m.group(1)
+
+
+def _ins(account_id, status="queued", held=0, created_at=None):
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        content = {"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
+                   "data_prestatie": "2026-06-15", "odometru_final": "123456",
+                   "prestatii": [{"cod_prestatie": "OE-1"}]}
+        if created_at is None:
+            cur = conn.execute(
+                "INSERT INTO submissions (idempotency_key, account_id, status, payload_json, held) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (f"k-{os.urandom(6).hex()}", account_id, status, json.dumps(content), held),
+            )
+        else:
+            cur = conn.execute(
+                "INSERT INTO submissions (idempotency_key, account_id, status, payload_json, held, created_at) "
+                "VALUES (?, ?, ?, ?, ?, ?)",
+                (f"k-{os.urandom(6).hex()}", account_id, status, json.dumps(content), held, created_at),
+            )
+        conn.commit()
+        return int(cur.lastrowid)
+    finally:
+        conn.close()
+
+
+def _get_auto_send(account_id):
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        r = conn.execute("SELECT auto_send_enabled FROM accounts WHERE id=?", (account_id,)).fetchone()
+        return int(r["auto_send_enabled"])
+    finally:
+        conn.close()
+
+
+def _held(sid):
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        return int(conn.execute("SELECT held FROM submissions WHERE id=?", (sid,)).fetchone()["held"])
+    finally:
+        conn.close()
+
+
+# --------------------------------------------------------------------------- #
+# US-004 — toggle Auto: persistenta + CSRF + scope                             #
+# --------------------------------------------------------------------------- #
+
+def test_auto_send_persista(client):
+    aid = _account_user("a1@test.com")
+    csrf = _login(client, "a1@test.com")
+    assert _get_auto_send(aid) == 0
+    r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 1
+    # Debifare
+    r = client.post("/auto-send", data={"csrf_token": csrf})  # enabled absent = OFF
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 0
+
+
+def test_auto_send_csrf(client):
+    aid = _account_user("a2@test.com")
+    _login(client, "a2@test.com")
+    r = client.post("/auto-send", data={"enabled": "1", "csrf_token": "gresit"})
+    assert r.status_code == 403
+    assert _get_auto_send(aid) == 0
+
+
+def test_auto_send_scoped_nu_atinge_alt_cont(client):
+    aid = _account_user("a3@test.com", name="A3")
+    other = _account_user("a3b@test.com", name="A3B")
+    csrf = _login(client, "a3@test.com")
+    r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 1
+    # contul celalalt ramane OFF (account_id vine din sesiune, nu din formular)
+    assert _get_auto_send(other) == 0
+
+
+def test_auto_send_scoped_ignora_account_id_din_formular(client):
+    aid = _account_user("a4@test.com", name="A4")
+    other = _account_user("a4b@test.com", name="A4B")
+    csrf = _login(client, "a4@test.com")
+    # incearca sa forteze alt cont prin formular -> trebuie ignorat
+    r = client.post("/auto-send", data={"enabled": "1", "account_id": str(other), "csrf_token": csrf})
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 1
+    assert _get_auto_send(other) == 0
+
+
+# --------------------------------------------------------------------------- #
+# US-004 — auto-release OFF->ON cu garda de confirmare                         #
+# --------------------------------------------------------------------------- #
+
+def test_off_to_on_fara_confirmare_nu_elibereaza(client):
+    aid = _account_user("g1@test.com")
+    csrf = _login(client, "g1@test.com")
+    sid = _ins(aid, "queued", held=1)
+    r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
+    assert r.status_code == 200
+    # fara confirmare: NU comita ON, NU elibereaza
+    assert _get_auto_send(aid) == 0
+    assert _held(sid) == 1
+    # raspunsul cere confirmare (modal)
+    assert "confirm" in r.text.lower() or "trimit" in r.text.lower()
+
+
+def test_off_to_on_cu_confirmare_elibereaza_scoped(client):
+    aid = _account_user("g2@test.com", name="G2")
+    other = _account_user("g2b@test.com", name="G2B")
+    csrf = _login(client, "g2@test.com")
+    s1 = _ins(aid, "queued", held=1)
+    s2 = _ins(aid, "queued", held=1)
+    s_other = _ins(other, "queued", held=1)
+    r = client.post("/auto-send", data={"enabled": "1", "confirma": "1", "csrf_token": csrf})
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 1
+    assert _held(s1) == 0
+    assert _held(s2) == 0
+    # cross-account izolare: randul altui cont ramane tinut
+    assert _held(s_other) == 1
+
+
+def test_off_to_on_fara_randuri_tinute_comite_direct(client):
+    aid = _account_user("g3@test.com")
+    csrf = _login(client, "g3@test.com")
+    # niciun rand tinut -> nu are nevoie de confirmare
+    r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
+    assert r.status_code == 200
+    assert _get_auto_send(aid) == 1
+
+
+# --------------------------------------------------------------------------- #
+# US-005 — trimitere manuala per rand + bulk                                   #
+# --------------------------------------------------------------------------- #
+
+def test_trimite_acum_rand_tinut(client):
+    aid = _account_user("t1@test.com")
+    csrf = _login(client, "t1@test.com")
+    sid = _ins(aid, "queued", held=1)
+    r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
+    assert r.status_code == 200
+    assert _held(sid) == 0
+
+
+def test_trimite_acum_cross_account_404(client):
+    aid = _account_user("t2@test.com", name="T2")
+    other = _account_user("t2b@test.com", name="T2B")
+    csrf = _login(client, "t2@test.com")
+    sid_other = _ins(other, "queued", held=1)
+    r = client.post(f"/trimitere/{sid_other}/trimite-acum", data={"csrf_token": csrf})
+    assert r.status_code == 404
+    assert _held(sid_other) == 1
+
+
+def test_trimite_acum_non_queued_noop(client):
+    aid = _account_user("t3@test.com")
+    csrf = _login(client, "t3@test.com")
+    sid = _ins(aid, "sent", held=0)
+    r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
+    assert r.status_code == 200
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        assert conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()["status"] == "sent"
+    finally:
+        conn.close()
+
+
+def test_trimite_acum_csrf(client):
+    aid = _account_user("t4@test.com")
+    _login(client, "t4@test.com")
+    sid = _ins(aid, "queued", held=1)
+    r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": "gresit"})
+    assert r.status_code == 403
+    assert _held(sid) == 1
+
+
+def test_trimite_toate_scoped(client):
+    aid = _account_user("t5@test.com", name="T5")
+    other = _account_user("t5b@test.com", name="T5B")
+    csrf = _login(client, "t5@test.com")
+    s1 = _ins(aid, "queued", held=1)
+    s2 = _ins(aid, "queued", held=1)
+    s_other = _ins(other, "queued", held=1)
+    r = client.post("/trimite-toate", data={"csrf_token": csrf})
+    assert r.status_code == 200
+    assert _held(s1) == 0
+    assert _held(s2) == 0
+    assert _held(s_other) == 1
+
+
+# --------------------------------------------------------------------------- #
+# US-006 (UI) — eticheta held + buton Trimite in lista                         #
+# --------------------------------------------------------------------------- #
+
+def test_rand_tinut_eticheta_si_buton(client):
+    aid = _account_user("u1@test.com")
+    _login(client, "u1@test.com")
+    sid_tinut = _ins(aid, "queued", held=1)
+    html = client.get("/_fragments/submissions").text
+    assert "In asteptare (manual)" in html or "Manual" in html
+    assert f"/trimitere/{sid_tinut}/trimite-acum" in html
+
+
+def test_rand_ne_tinut_fara_buton(client):
+    aid = _account_user("u2@test.com")
+    _login(client, "u2@test.com")
+    sid = _ins(aid, "queued", held=0)
+    html = client.get("/_fragments/submissions").text
+    assert f"/trimitere/{sid}/trimite-acum" not in html
+
+
+# --------------------------------------------------------------------------- #
+# US-007 — banner coada tinuta imbatranita + contor manual                     #
+# --------------------------------------------------------------------------- #
+
+def test_banner_aged_apare_pt_rand_vechi(client):
+    aid = _account_user("b1@test.com")
+    _login(client, "b1@test.com")
+    _ins(aid, "queued", held=1, created_at="2026-06-01 08:00:00")  # >7 zile fata de 2026-07-05
+    html = client.get("/_fragments/status").text
+    assert "declarare obligatorie" in html.lower() or "tinute de" in html.lower()
+
+
+def test_banner_aged_absent_pt_rand_recent(client):
+    aid = _account_user("b2@test.com")
+    _login(client, "b2@test.com")
+    from datetime import datetime, timezone
+    _ins(aid, "queued", held=1, created_at=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"))
+    html = client.get("/_fragments/status").text
+    assert "declarare obligatorie" not in html.lower()
+
+
+def test_contor_manual_apare(client):
+    aid = _account_user("b3@test.com")
+    _login(client, "b3@test.com")
+    _ins(aid, "queued", held=1)
+    _ins(aid, "queued", held=1)
+    html = client.get("/_fragments/status").text
+    assert "manual" in html.lower()
+
+
+# --------------------------------------------------------------------------- #
+# US-009 — audit app_events                                                    #
+# --------------------------------------------------------------------------- #
+
+def _events(account_id, tip):
+    from app.db import get_connection
+    conn = get_connection()
+    try:
+        rows = conn.execute(
+            "SELECT * FROM app_events WHERE tip=? AND account_id=?", (tip, account_id)
+        ).fetchall()
+        return rows
+    finally:
+        conn.close()
+
+
+def test_audit_auto_send_schimbat(client):
+    aid = _account_user("j1@test.com")
+    csrf = _login(client, "j1@test.com")
+    client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
+    assert len(_events(aid, "auto_send_schimbat")) >= 1
+
+
+def test_audit_held_eliberat_rand(client):
+    aid = _account_user("j2@test.com")
+    csrf = _login(client, "j2@test.com")
+    sid = _ins(aid, "queued", held=1)
+    client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
+    assert len(_events(aid, "held_eliberat")) >= 1
+
+
+def test_audit_held_eliberat_bulk(client):
+    aid = _account_user("j3@test.com")
+    csrf = _login(client, "j3@test.com")
+    _ins(aid, "queued", held=1)
+    _ins(aid, "queued", held=1)
+    client.post("/trimite-toate", data={"csrf_token": csrf})
+    evs = _events(aid, "held_eliberat")
+    assert len(evs) >= 1

tests/test_worker_held.py

@@ -0,0 +1,204 @@
+"""Teste PRD 5.19 US-003 + US-008 — worker respecta flag `held`.
+
+US-003: claim_one sare peste randurile `queued AND held=1` (tinute manual).
+US-008: expire_held expira randurile tinute mai vechi de `held_retention_days`
+        (default 90) -> status 'error', rar_error='TINUT_EXPIRAT', purge_after
+        DIRECT la momentul expirarii (NU +30z retentie blocati -> nu 120z total),
+        apoi purge_expired le sterge (PII purjat).
+
+TDD: testele se scriu INAINTE de modificarea worker-ului.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+
+import pytest
+
+
+# --- Fixture DB (pattern din test_worker_active_gate.py) ---
+
+@pytest.fixture()
+def env(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.db import get_connection, init_db
+    init_db()
+    conn = get_connection()
+    yield conn, get_settings()
+    conn.close()
+    get_settings.cache_clear()
+
+
+# --- 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(conn, account_id=None, status="queued", held=0, content=None, created_at=None):
+    content = content or _CONTENT
+    cur = conn.execute(
+        "INSERT INTO submissions (idempotency_key, status, payload_json, account_id, held) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (f"key-{os.urandom(4).hex()}", status, json.dumps(content), account_id, held),
+    )
+    sid = int(cur.lastrowid)
+    if created_at is not None:
+        conn.execute("UPDATE submissions SET created_at=? WHERE id=?", (created_at, sid))
+    return sid
+
+
+def _row(conn, sid):
+    return conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone()
+
+
+def _row_status(conn, sid):
+    r = _row(conn, sid)
+    return r["status"] if r else None
+
+
+# --- US-003: claim_one sare peste randurile tinute ---
+
+def test_claim_sare_rand_tinut(env):
+    """queued, held=1, cont activ, send on -> claim_one intoarce None; ramane queued."""
+    from app.accounts import create_account
+    from app.worker.__main__ import claim_one
+
+    conn, _ = env
+    acct_id = create_account(conn, "Service Tinut", active=True)
+    sid = _insert(conn, account_id=acct_id, held=1)
+
+    result = claim_one(conn)
+
+    assert result is None, "claim_one trebuia sa sara peste randul tinut (held=1)"
+    assert _row_status(conn, sid) == "queued", "randul tinut trebuia sa ramana queued"
+
+
+def test_claim_ia_rand_neted(env):
+    """Acelasi rand cu held=0 -> claim_one il ia si il marcheaza sending."""
+    from app.accounts import create_account
+    from app.worker.__main__ import claim_one
+
+    conn, _ = env
+    acct_id = create_account(conn, "Service Liber", active=True)
+    sid = _insert(conn, account_id=acct_id, held=0)
+
+    result = claim_one(conn)
+
+    assert result is not None, "claim_one trebuia sa ridice randul held=0"
+    assert result["id"] == sid
+    assert _row_status(conn, sid) == "sending"
+
+
+def test_eliberarea_deblocheaza_claim(env):
+    """held=1 -> claim_one None; dupa held=0 -> claim_one il ridica."""
+    from app.accounts import create_account
+    from app.worker.__main__ import claim_one
+
+    conn, _ = env
+    acct_id = create_account(conn, "Service Eliberat", active=True)
+    sid = _insert(conn, account_id=acct_id, held=1)
+
+    assert claim_one(conn) is None
+    assert _row_status(conn, sid) == "queued"
+
+    conn.execute("UPDATE submissions SET held=0 WHERE id=?", (sid,))
+    result = claim_one(conn)
+    assert result is not None and result["id"] == sid
+    assert _row_status(conn, sid) == "sending"
+
+
+# --- US-008: expirare randuri tinute ---
+
+def test_expire_held_expira_rand_vechi(env):
+    """queued, held=1, created_at vechi (>90z) -> error + TINUT_EXPIRAT + purge_after ~now."""
+    from app.accounts import create_account
+    from app.worker.__main__ import expire_held
+
+    conn, settings = env
+    acct_id = create_account(conn, "Service Vechi", active=True)
+    sid = _insert(conn, account_id=acct_id, held=1,
+                  created_at="datetime('now','-100 days')")
+    # created_at trebuie sa fie o valoare literala, nu expresie SQL text -> setam explicit.
+    conn.execute("UPDATE submissions SET created_at=datetime('now','-100 days') WHERE id=?", (sid,))
+
+    n = expire_held(conn, settings)
+
+    assert n == 1, "expire_held trebuia sa expire exact 1 rand"
+    row = _row(conn, sid)
+    assert row["status"] == "error"
+    assert row["rar_error"] == "TINUT_EXPIRAT"
+    assert row["purge_after"] is not None
+    # purge_after ~ now (NU +30z): verificam ca e <= acum + o mica marja (sub 1 zi).
+    aproape_acum = conn.execute(
+        "SELECT purge_after <= datetime('now', '+1 day') AS ok FROM submissions WHERE id=?",
+        (sid,),
+    ).fetchone()["ok"]
+    assert aproape_acum == 1, "purge_after trebuia setat ~now, nu +30z"
+    assert row["sending_since"] is None
+
+
+def test_expire_held_ignora_rand_recent(env):
+    """queued, held=1 recent (created_at now) -> neatins de expire_held."""
+    from app.accounts import create_account
+    from app.worker.__main__ import expire_held
+
+    conn, settings = env
+    acct_id = create_account(conn, "Service Recent", active=True)
+    sid = _insert(conn, account_id=acct_id, held=1)
+
+    n = expire_held(conn, settings)
+
+    assert n == 0
+    assert _row_status(conn, sid) == "queued"
+
+
+def test_expire_held_ignora_rand_neted_vechi(env):
+    """queued, held=0, vechi -> NU e tinut -> neatins de expire_held."""
+    from app.accounts import create_account
+    from app.worker.__main__ import expire_held
+
+    conn, settings = env
+    acct_id = create_account(conn, "Service NetedVechi", active=True)
+    sid = _insert(conn, account_id=acct_id, held=0)
+    conn.execute("UPDATE submissions SET created_at=datetime('now','-100 days') WHERE id=?", (sid,))
+
+    n = expire_held(conn, settings)
+
+    assert n == 0
+    assert _row_status(conn, sid) == "queued"
+
+
+def test_expire_held_apoi_purge_sterge_randul(env):
+    """Lant complet: expire_held (purge_after ~now) -> purge_expired sterge randul (PII purjat)."""
+    from app.accounts import create_account
+    from app.worker.__main__ import expire_held, purge_expired
+
+    conn, settings = env
+    acct_id = create_account(conn, "Service Purjat", active=True)
+    sid = _insert(conn, account_id=acct_id, held=1)
+    conn.execute("UPDATE submissions SET created_at=datetime('now','-100 days') WHERE id=?", (sid,))
+
+    assert expire_held(conn, settings) == 1
+    # purge_after=datetime('now') (moment expirarii). La URMATORUL ciclu de purjare
+    # (o data pe ora) purge_after < now -> se sterge. Simulam trecerea timpului
+    # (>= 1s) verificand ca purge_after e deja in trecutul acelui ciclu ulterior.
+    assert conn.execute(
+        "SELECT purge_after < datetime('now','+1 second') AS ok FROM submissions WHERE id=?",
+        (sid,),
+    ).fetchone()["ok"] == 1
+    conn.execute(
+        "UPDATE submissions SET purge_after=datetime('now','-1 second') WHERE id=?", (sid,)
+    )
+    stats = purge_expired(conn)
+
+    assert stats["submissions_purged"] >= 1
+    assert _row(conn, sid) is None, "randul expirat trebuia sters de purge_expired la ciclul ulterior"