diff --git a/src/backend/tests/test_migrations.py b/src/backend/tests/test_migrations.py
index 9b61d71..3938829 100644
--- a/src/backend/tests/test_migrations.py
+++ b/src/backend/tests/test_migrations.py
@@ -131,3 +131,21 @@ class TestApiKeyHashMigration:
 
         command.upgrade(cfg, "head")  # and up again
         assert "api_keys" in _inspect(db_path).get_table_names()
+
+
+class TestVisionRecipeDirMigration:
+    """016_vision_recipe_dir: recipe_tasks.vision_recipe_dir, su e giu (#46)."""
+
+    def test_016_upgrade_and_downgrade(self, sqlite_database_url):
+        db_path, _ = sqlite_database_url
+        cfg = _alembic_config()
+
+        def cols():
+            return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
+
+        command.upgrade(cfg, "head")
+        assert "vision_recipe_dir" in cols()
+        command.downgrade(cfg, "015_api_key_hash")
+        assert "vision_recipe_dir" not in cols()
+        command.upgrade(cfg, "head")  # e di nuovo su
+        assert "vision_recipe_dir" in cols()
diff --git a/src/backend/tests/test_vision_recipe_upload.py b/src/backend/tests/test_vision_recipe_upload.py
new file mode 100644
index 0000000..ed922fb
--- /dev/null
+++ b/src/backend/tests/test_vision_recipe_upload.py
@@ -0,0 +1,265 @@
+"""Ricetta VisionSuite v2 su un task, execute dxf_compare, copy-on-write (#46)."""
+import io
+import json
+
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from src.backend.config import settings
+from src.backend.models.orm.recipe import RecipeVersion
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.models.orm.vision import VisionResult
+from src.backend.tests.conftest import auth_headers, create_test_recipe
+
+RECIPE = {
+    "schema_version": 2, "name": "dxf", "tools": [],
+    "meta": {"dxf": {"d1": {"path": "dxf/part.dxf"}}, "models": {"m": {"path": "models/m.npz"}}},
+}
+OUTPUTS = [{"key": "cmp.points_in_pct", "quantity": "ratio", "unit": "%"}]
+GOOD_FILES = (("part.dxf", b"0 EOF"), ("m.npz", b"npz"))
+
+
+@pytest.fixture(autouse=True)
+def redirect_uploads(monkeypatch, tmp_path):
+    # upload_path = parents[2] / upload_dir: un assoluto sostituisce la base
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
+
+
+@pytest.fixture
+def validate_says(monkeypatch):
+    """Il worker /validate non gira nei test: risposta finta."""
+    def _install(ok=True, errors=None, outputs=None):
+        async def fake(graph, recipe_dir):
+            return {
+                "ok": ok, "errors": errors or [],
+                "outputs": outputs if outputs is not None else OUTPUTS,
+            }
+        monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
+    return _install
+
+
+def an_image() -> bytes:
+    buf = io.BytesIO()
+    Image.new("L", (32, 32)).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+async def make_task(db_session, user, task_type="dxf_compare"):
+    recipe = await create_test_recipe(db_session, user.id)
+    version = (await db_session.execute(
+        select(RecipeVersion).where(RecipeVersion.recipe_id == recipe.id)
+    )).scalars().first()
+    task = (await db_session.execute(
+        select(RecipeTask).where(RecipeTask.version_id == version.id)
+    )).scalars().first()
+    subtask = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id)
+    )).scalars().first()
+    task.task_type = task_type
+    await db_session.commit()
+    return task, subtask
+
+
+def upload(client, user, task, recipe=RECIPE, files=()):
+    return client.post(
+        f"/api/vision/tasks/{task.id}/recipe",
+        headers=auth_headers(user),
+        files=[("recipe", ("r.json", json.dumps(recipe).encode(), "application/json"))]
+        + [("files", (n, c, "application/octet-stream")) for n, c in files],
+    )
+
+
+@pytest.mark.asyncio
+async def test_upload_ok_stores_folder_and_task(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    resp = await upload(client, maker_user, task, files=GOOD_FILES)
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["outputs"] == OUTPUTS
+    assert sorted(body["files"]) == ["dxf/part.dxf", "models/m.npz"]
+    assert body["recipe_dir"].startswith(f"vision/recipes/{task.id}/")
+    folder = settings.upload_path / body["recipe_dir"]
+    assert (folder / "recipe.json").is_file() and (folder / "dxf" / "part.dxf").is_file()
+    await db_session.refresh(task)
+    assert task.vision_json == RECIPE and task.vision_recipe_dir == body["recipe_dir"]
+
+    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
+    assert got.status_code == 200
+    assert got.json()["recipe"] == RECIPE and got.json()["outputs"] == OUTPUTS
+
+
+@pytest.mark.asyncio
+async def test_get_without_recipe_is_empty(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
+    assert got.json() == {
+        "task_id": task.id, "recipe_dir": None, "recipe": None, "files": [], "outputs": [],
+    }
+
+
+@pytest.mark.asyncio
+async def test_unmatched_file_name_is_422_list(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    resp = await upload(client, maker_user, task, files=[("other.dxf", b"x")])
+    assert resp.status_code == 422
+    assert isinstance(resp.json()["detail"], list) and "other.dxf" in resp.json()["detail"][0]
+    await db_session.refresh(task)
+    assert task.vision_json is None and task.vision_recipe_dir is None
+
+
+@pytest.mark.asyncio
+async def test_traversal_in_meta_path_is_422(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    evil = {"schema_version": 2, "tools": [], "meta": {"dxf": {"d": {"path": "../../evil.dxf"}}}}
+    resp = await upload(client, maker_user, task, recipe=evil, files=[("evil.dxf", b"x")])
+    assert resp.status_code == 422
+    assert not (settings.upload_path.parent / "evil.dxf").exists()
+
+
+@pytest.mark.asyncio
+async def test_wrong_task_type_is_422(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user, task_type="measure")
+    resp = await upload(client, maker_user, task, files=GOOD_FILES)
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_invalid_json_is_422(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    resp = await client.post(
+        f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user),
+        files=[("recipe", ("r.json", b"{nope", "application/json"))],
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_worker_not_ok_is_422_and_task_unchanged(client, db_session, maker_user, validate_says):
+    validate_says(ok=False, errors=["meta file missing"])
+    task, _ = await make_task(db_session, maker_user)
+    task.vision_json = {"old": 1}
+    task.vision_recipe_dir = "vision/recipes/old"
+    await db_session.commit()
+    resp = await upload(client, maker_user, task, files=GOOD_FILES)
+    assert resp.status_code == 422
+    assert resp.json()["detail"] == ["meta file missing"]
+    await db_session.refresh(task)
+    assert task.vision_json == {"old": 1} and task.vision_recipe_dir == "vision/recipes/old"
+
+
+@pytest.mark.asyncio
+async def test_existing_folder_is_not_rewritten(client, db_session, maker_user, validate_says):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    first = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
+    marker = settings.upload_path / first["recipe_dir"] / "dxf" / "part.dxf"
+    marker.write_bytes(b"CHANGED")
+    second = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
+    assert second["recipe_dir"] == first["recipe_dir"]
+    assert marker.read_bytes() == b"CHANGED"
+
+
+@pytest.mark.asyncio
+async def test_upload_requires_maker(
+    client, db_session, maker_user, measurement_tec_user, validate_says,
+):
+    validate_says()
+    task, _ = await make_task(db_session, maker_user)
+    resp = await upload(client, measurement_tec_user, task, files=GOOD_FILES)
+    assert resp.status_code in (401, 403)
+
+
+@pytest.mark.asyncio
+async def test_execute_dxf_compare_saves_files_and_extended_response(
+    client, db_session, measurement_tec_user, monkeypatch,
+):
+    task, subtask = await make_task(db_session, measurement_tec_user)
+    task.vision_json = RECIPE
+    task.vision_recipe_dir = "vision/recipes/1/abc"
+    subtask.vision_output = "cmp.points_in_pct"
+    await db_session.commit()
+    seen = {}
+    overlay = [{"type": "contour", "role": "found", "rows": [1], "cols": [2], "tool_id": "cmp"}]
+
+    async def fake_call(image_bytes, graph, **kwargs):
+        seen.update(kwargs)
+        return {
+            "outputs": {"cmp.points_in_pct": float(subtask.nominal)},
+            "failures": [], "engine_version": "e" * 40, "duration_ms": 5,
+            "overlay": overlay, "outcome": "OK",
+        }
+    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
+
+    resp = await client.post(
+        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
+        files={"image": ("p.png", an_image(), "image/png")},
+        data={"task_id": str(task.id)},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert seen["recipe_dir"] == "vision/recipes/1/abc" and "calibration" not in seen
+    assert body["outcome"] == "OK" and body["overlay"] == overlay and body["failures"] == []
+    assert body["image_url"] == f"/api/vision/results/{body['vision_result_id']}/image"
+    assert body["measurements"][0]["pass_fail"] == "pass"
+
+    result = (await db_session.execute(select(VisionResult))).scalars().one()
+    assert (settings.upload_path / result.image_path).read_bytes() == an_image()
+    assert json.loads((settings.upload_path / result.overlay_path).read_text()) == overlay
+
+    img = await client.get(body["image_url"], headers=auth_headers(measurement_tec_user))
+    assert img.status_code == 200 and img.content == an_image()
+
+
+@pytest.mark.asyncio
+async def test_execute_dxf_compare_out_of_tolerance_fails(
+    client, db_session, measurement_tec_user, monkeypatch,
+):
+    task, subtask = await make_task(db_session, measurement_tec_user)
+    task.vision_json = RECIPE
+    subtask.vision_output = "cmp.points_in_pct"
+    await db_session.commit()
+
+    async def fake_call(image_bytes, graph, **kwargs):
+        return {"outputs": {"cmp.points_in_pct": 99.0}, "failures": [],
+                "engine_version": "e" * 40, "outcome": "OK"}
+    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
+    resp = await client.post(
+        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
+        files={"image": ("p.png", an_image(), "image/png")},
+        data={"task_id": str(task.id)},
+    )
+    assert resp.status_code == 200
+    # il verdetto e' delle tolleranze del subtask, non di outcome di VisionSuite
+    assert resp.json()["measurements"][0]["pass_fail"] == "fail"
+    assert resp.json()["overlay"] == []
+
+
+@pytest.mark.asyncio
+async def test_copy_on_write_carries_recipe_dir(client, db_session, maker_user):
+    from src.backend.tests.test_vision_copy_on_write import (
+        _current_task_and_subtask, _seed_measurement,
+    )
+    recipe = await create_test_recipe(db_session, maker_user.id)
+    _, task, _ = await _current_task_and_subtask(db_session, recipe.id)
+    task.task_type = "dxf_compare"
+    task.vision_json = RECIPE
+    task.vision_recipe_dir = "vision/recipes/1/abc"
+    await db_session.flush()
+    await _seed_measurement(db_session, recipe.id, maker_user.id)
+    db_session.expunge_all()
+    resp = await client.put(
+        f"/api/recipes/{recipe.id}", headers=auth_headers(maker_user),
+        json={"change_notes": "cow"},
+    )
+    assert resp.status_code == 200
+    db_session.expunge_all()
+    version, new_task, _ = await _current_task_and_subtask(db_session, recipe.id)
+    assert version.version_number == 2
+    assert new_task.vision_recipe_dir == "vision/recipes/1/abc" and new_task.vision_json == RECIPE