diff --git a/src/backend/api/routers/vision.py b/src/backend/api/routers/vision.py
new file mode 100644
index 0000000..7e4901e
--- /dev/null
+++ b/src/backend/api/routers/vision.py
@@ -0,0 +1,66 @@
+"""The client asks for a measurement and gets an outcome.
+
+It does not know, and must not know, whether the server or the station computed
+the numbers: that is what lets a station's configuration change without touching
+the frontend or the recipes.
+"""
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.backend.api.middleware.api_key import require_measurement_tec
+from src.backend.database import get_db
+from src.backend.models.api.vision import VisionExecuteResponse, VisionMeasurementResult
+from src.backend.models.orm.task import RecipeTask
+from src.backend.models.orm.user import User
+from src.backend.services import vision_service
+
+router = APIRouter(prefix="/api/vision", tags=["vision"])
+
+
+async def _task_or_404(db: AsyncSession, task_id: int) -> RecipeTask:
+    """Shared by the endpoints of this router."""
+    task = (await db.execute(
+        select(RecipeTask).where(RecipeTask.id == task_id)
+    )).scalar_one_or_none()
+    if task is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
+        )
+    return task
+
+
+@router.post("/execute", response_model=VisionExecuteResponse)
+async def execute(
+    task_id: int = Form(...),
+    image: UploadFile = File(...),
+    lot_number: str | None = Form(None),
+    serial_number: str | None = Form(None),
+    production_run_id: int | None = Form(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_measurement_tec),
+) -> VisionExecuteResponse:
+    task = await _task_or_404(db, task_id)
+
+    result, saved = await vision_service.execute_task(
+        db,
+        task,
+        await image.read(),
+        current_user.id,
+        version_id=task.version_id,
+        lot_number=lot_number,
+        serial_number=serial_number,
+        production_run_id=production_run_id,
+    )
+    await db.commit()
+
+    return VisionExecuteResponse(
+        vision_result_id=result.id,
+        engine_version=result.engine_version,
+        measurements=[
+            VisionMeasurementResult(
+                subtask_id=m.subtask_id, value=float(m.value), pass_fail=m.pass_fail,
+            )
+            for m in saved
+        ],
+    )
diff --git a/src/backend/config.py b/src/backend/config.py
index e850149..7eff6d7 100644
--- a/src/backend/config.py
+++ b/src/backend/config.py
@@ -38,6 +38,9 @@ class Settings(BaseSettings):
     openrouter_api_key: str | None = None
     openrouter_model: str = "anthropic/claude-sonnet-4"
 
+    # Vision worker (internal network only, reachable as `vision` in Compose)
+    vision_worker_url: str = "http://vision:8100"
+
     @property
     def database_url(self) -> str:
         """Async MySQL connection string."""
diff --git a/src/backend/main.py b/src/backend/main.py
index a1dc9f7..69cb309 100644
--- a/src/backend/main.py
+++ b/src/backend/main.py
@@ -22,6 +22,7 @@ from src.backend.api.routers.statistics import router as statistics_router
 from src.backend.api.routers.setup import router as setup_router
 from src.backend.api.routers.stations import router as stations_router
 from src.backend.api.routers.production import router as production_router
+from src.backend.api.routers.vision import router as vision_router
 
 
 @asynccontextmanager
@@ -75,6 +76,7 @@ app.include_router(reports_router)
 app.include_router(setup_router)
 app.include_router(stations_router)
 app.include_router(production_router)
+app.include_router(vision_router)
 
 
 @app.get("/api/health")
diff --git a/src/backend/migrations/versions/012_vision_results.py b/src/backend/migrations/versions/012_vision_results.py
new file mode 100644
index 0000000..acaed5b
--- /dev/null
+++ b/src/backend/migrations/versions/012_vision_results.py
@@ -0,0 +1,79 @@
+"""one row per vision execution, and camera as an input method
+
+The vision surroundings do not go on `measurements`: statistics and the export
+read that table on every pass. And it would be wrong modelling anyway - one
+acquisition produces N quotes, and image, overlay, device and engine version are
+the same for all of them.
+
+Revision ID: 012_vision_results
+Revises: 011_vision_graph
+Create Date: 2026-08-16
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '012_vision_results'
+down_revision: Union[str, None] = '011_vision_graph'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+OLD_METHODS = ('usb_caliper', 'manual')
+NEW_METHODS = ('usb_caliper', 'manual', 'camera')
+
+
+def upgrade() -> None:
+    op.create_table(
+        'vision_results',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('task_id', sa.Integer(),
+                  sa.ForeignKey('recipe_tasks.id'), nullable=False, index=True),
+        sa.Column('image_path', sa.String(length=500), nullable=True),
+        sa.Column('overlay_path', sa.String(length=500), nullable=True),
+        sa.Column('engine_version', sa.String(length=64), nullable=False),
+        sa.Column('executed_on',
+                  sa.Enum('server', 'station', name='vision_executed_on_enum'),
+                  nullable=False),
+        sa.Column('station_id', sa.Integer(),
+                  sa.ForeignKey('stations.id'), nullable=True),
+        sa.Column('device_code', sa.String(length=100), nullable=True),
+        sa.Column('calibration_snapshot', sa.JSON(), nullable=True),
+        sa.Column('graph_snapshot', sa.JSON(), nullable=False),
+        sa.Column('duration_ms', sa.Integer(), nullable=True),
+        sa.Column('executed_at', sa.DateTime(), nullable=False,
+                  server_default=sa.func.now()),
+        mysql_engine='InnoDB',
+        mysql_charset='utf8mb4',
+    )
+
+    with op.batch_alter_table('measurements') as batch:
+        batch.alter_column(
+            'input_method',
+            existing_type=sa.Enum(*OLD_METHODS, name='input_method_enum'),
+            type_=sa.Enum(*NEW_METHODS, name='input_method_enum'),
+            existing_nullable=False,
+        )
+        batch.add_column(
+            sa.Column('vision_result_id', sa.Integer(), nullable=True)
+        )
+        batch.create_foreign_key(
+            'fk_measurements_vision_result',
+            'vision_results', ['vision_result_id'], ['id'],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('measurements') as batch:
+        batch.drop_constraint('fk_measurements_vision_result',
+                              type_='foreignkey')
+        batch.drop_column('vision_result_id')
+        batch.alter_column(
+            'input_method',
+            existing_type=sa.Enum(*NEW_METHODS, name='input_method_enum'),
+            type_=sa.Enum(*OLD_METHODS, name='input_method_enum'),
+            existing_nullable=False,
+        )
+
+    op.drop_table('vision_results')
diff --git a/src/backend/models/api/vision.py b/src/backend/models/api/vision.py
new file mode 100644
index 0000000..6180b5e
--- /dev/null
+++ b/src/backend/models/api/vision.py
@@ -0,0 +1,16 @@
+"""Pydantic schemas for the vision execution response."""
+from pydantic import BaseModel
+
+
+class VisionMeasurementResult(BaseModel):
+    """One quote's outcome, as much of it as the client needs to show."""
+    subtask_id: int
+    value: float
+    pass_fail: str
+
+
+class VisionExecuteResponse(BaseModel):
+    """What `POST /api/vision/execute` returns: the run, and every quote it filled."""
+    vision_result_id: int
+    engine_version: str
+    measurements: list[VisionMeasurementResult]
diff --git a/src/backend/models/orm/__init__.py b/src/backend/models/orm/__init__.py
index 92bdc04..b121349 100644
--- a/src/backend/models/orm/__init__.py
+++ b/src/backend/models/orm/__init__.py
@@ -7,6 +7,7 @@ from src.backend.models.orm.access_log import AccessLog
 from src.backend.models.orm.setting import SystemSetting, RecipeVersionAudit
 from src.backend.models.orm.station import Station, StationRecipeAssignment
 from src.backend.models.orm.production import ProductionRun, ProductionEvent
+from src.backend.models.orm.vision import VisionResult
 
 __all__ = [
     "User",
@@ -22,4 +23,5 @@ __all__ = [
     "StationRecipeAssignment",
     "ProductionRun",
     "ProductionEvent",
+    "VisionResult",
 ]
diff --git a/src/backend/models/orm/measurement.py b/src/backend/models/orm/measurement.py
index 6fba791..1020fd7 100644
--- a/src/backend/models/orm/measurement.py
+++ b/src/backend/models/orm/measurement.py
@@ -38,11 +38,17 @@ class Measurement(Base):
 
     # Input method
     input_method: Mapped[str] = mapped_column(
-        Enum("usb_caliper", "manual", name="input_method_enum"),
+        Enum("usb_caliper", "manual", "camera", name="input_method_enum"),
         nullable=False,
         default="manual",
     )
 
+    # The vision execution that produced this value, when there was one. Null for
+    # everything taken by caliper or typed by hand.
+    vision_result_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("vision_results.id"), nullable=True
+    )
+
     # Time spent by the operator entering this value (client-reported)
     input_duration_ms: Mapped[Optional[int]] = mapped_column(
         Integer, nullable=True
diff --git a/src/backend/models/orm/vision.py b/src/backend/models/orm/vision.py
new file mode 100644
index 0000000..1749cb2
--- /dev/null
+++ b/src/backend/models/orm/vision.py
@@ -0,0 +1,40 @@
+"""One row per vision execution - not per quote."""
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, JSON, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.backend.database import Base
+
+
+class VisionResult(Base):
+    __tablename__ = "vision_results"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    task_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("recipe_tasks.id"), nullable=False, index=True
+    )
+    image_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    overlay_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    engine_version: Mapped[str] = mapped_column(String(64), nullable=False)
+    executed_on: Mapped[str] = mapped_column(
+        Enum("server", "station", name="vision_executed_on_enum"), nullable=False
+    )
+    station_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("stations.id"), nullable=True
+    )
+    device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    # Copied, not referenced: a calibration redone six months from now must not
+    # rewrite what a measurement already taken means. Same for the graph.
+    calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    graph_snapshot: Mapped[dict] = mapped_column(JSON, nullable=False)
+    duration_ms: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    executed_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},)
+
+    def __repr__(self) -> str:
+        return f"<VisionResult task={self.task_id} engine={self.engine_version}>"
diff --git a/src/backend/services/measurement_service.py b/src/backend/services/measurement_service.py
index cacdb9a..37bab5c 100644
--- a/src/backend/services/measurement_service.py
+++ b/src/backend/services/measurement_service.py
@@ -224,6 +224,7 @@ async def save_measurement(
     input_method: str = "manual",
     input_duration_ms: int | None = None,
     production_run_id: int | None = None,
+    vision_result_id: int | None = None,
 ) -> Measurement:
     """Save a single measurement with auto-calculated pass/fail."""
     # Get subtask for tolerance values
@@ -266,6 +267,7 @@ async def save_measurement(
         input_method=input_method,
         input_duration_ms=input_duration_ms,
         production_run_id=production_run_id,
+        vision_result_id=vision_result_id,
     )
     db.add(measurement)
     await db.flush()
diff --git a/src/backend/services/vision_service.py b/src/backend/services/vision_service.py
new file mode 100644
index 0000000..6f636fa
--- /dev/null
+++ b/src/backend/services/vision_service.py
@@ -0,0 +1,107 @@
+"""Route the image to the worker, map the outputs onto the quotes, save.
+
+This module never imports VisionSuite - see test_vision_boundary.py. It speaks
+to the worker over HTTP and stays ignorant of how a number was produced.
+"""
+from __future__ import annotations
+
+import json
+
+import httpx
+from fastapi import HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.backend.config import settings
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.models.orm.vision import VisionResult
+from src.backend.services import measurement_service
+
+
+async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
+    """The one place that talks to the worker; the tests replace it."""
+    async with httpx.AsyncClient(timeout=120.0) as client:
+        response = await client.post(
+            f"{settings.vision_worker_url}/run",
+            files={"image": ("frame.png", image_bytes, "image/png")},
+            data={"graph": json.dumps(graph)},
+        )
+    if response.status_code != 200:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"the vision worker refused the job: {response.text}",
+        )
+    return response.json()
+
+
+async def execute_task(
+    db: AsyncSession,
+    task: RecipeTask,
+    image_bytes: bytes,
+    measured_by: int,
+    *,
+    version_id: int,
+    lot_number: str | None = None,
+    serial_number: str | None = None,
+    production_run_id: int | None = None,
+) -> tuple[VisionResult, list[Measurement]]:
+    if not task.vision_json:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="this task carries no vision graph",
+        )
+
+    quotes = (await db.execute(
+        select(RecipeSubtask)
+        .where(RecipeSubtask.task_id == task.id)
+        .where(RecipeSubtask.vision_output.is_not(None))
+    )).scalars().all()
+    if not quotes:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="no quote on this task names a vision output",
+        )
+
+    report = await _call_worker(image_bytes, task.vision_json)
+    outputs = report["outputs"]
+
+    # Every quote or none. A half-measured piece read as a whole one is worse
+    # than a piece not measured at all.
+    missing = [q.vision_output for q in quotes if q.vision_output not in outputs]
+    if missing:
+        failed = ", ".join(f["tool_name"] for f in report["failures"])
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=(
+                f"the graph did not produce {', '.join(missing)}"
+                + (f" (tools that failed: {failed})" if failed else "")
+            ),
+        )
+
+    result = VisionResult(
+        task_id=task.id,
+        engine_version=report["engine_version"],
+        executed_on="server",
+        graph_snapshot=task.vision_json,
+        duration_ms=report.get("duration_ms"),
+    )
+    db.add(result)
+    await db.flush()
+
+    saved: list[Measurement] = []
+    for quote in quotes:
+        saved.append(await measurement_service.save_measurement(
+            db,
+            subtask_id=quote.id,
+            version_id=version_id,
+            measured_by=measured_by,
+            value=float(outputs[quote.vision_output]),
+            lot_number=lot_number,
+            serial_number=serial_number,
+            input_method="camera",
+            production_run_id=production_run_id,
+            vision_result_id=result.id,
+        ))
+
+    return result, saved
diff --git a/src/backend/tests/test_vision_execute.py b/src/backend/tests/test_vision_execute.py
new file mode 100644
index 0000000..34800ca
--- /dev/null
+++ b/src/backend/tests/test_vision_execute.py
@@ -0,0 +1,153 @@
+"""A quote from a camera is a measurement like any other: same verdict, same gate."""
+import io
+
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from src.backend.models.orm.measurement import Measurement
+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
+
+
+def an_image() -> bytes:
+    buffer = io.BytesIO()
+    Image.new("L", (64, 64), color=0).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+async def recipe_parts(db_session, recipe_id: int):
+    """`create_test_recipe` returns only the recipe: here it is broken apart."""
+    version = (await db_session.execute(
+        select(RecipeVersion).where(
+            RecipeVersion.recipe_id == recipe_id,
+            RecipeVersion.is_current == True,  # noqa: E712
+        )
+    )).scalar_one()
+    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()
+    return version, task, subtask
+
+
+@pytest.fixture
+def worker_says(monkeypatch):
+    """The worker does not run in tests: its response is faked here."""
+    def _install(outputs: dict, failures: list | None = None):
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
+            return {
+                "outputs": outputs,
+                "failures": failures or [],
+                "engine_version": "a" * 40,
+                "duration_ms": 12,
+            }
+        monkeypatch.setattr(
+            "src.backend.services.vision_service._call_worker", fake_call,
+        )
+    return _install
+
+
+@pytest.mark.asyncio
+async def test_a_camera_quote_becomes_a_measurement(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    # The fixture creates the quote with nominal 10 and tight limits around it.
+    await db_session.commit()
+    worker_says({"t1.diameter": float(subtask.nominal)})
+
+    response = 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), "lot_number": "L1"},
+    )
+
+    assert response.status_code == 200
+    saved = (await db_session.execute(
+        select(Measurement).where(Measurement.subtask_id == subtask.id)
+    )).scalars().all()
+    assert len(saved) == 1
+    assert saved[0].pass_fail == "pass"
+    assert saved[0].input_method == "camera"
+    assert saved[0].vision_result_id is not None
+
+
+@pytest.mark.asyncio
+async def test_the_execution_records_which_engine_produced_it(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+    worker_says({"t1.diameter": float(subtask.nominal)})
+
+    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), "lot_number": "L1"},
+    )
+
+    result = (await db_session.execute(select(VisionResult))).scalars().one()
+    assert result.engine_version == "a" * 40
+    assert result.executed_on == "server"
+    assert result.graph_snapshot == task.vision_json
+
+
+@pytest.mark.asyncio
+async def test_a_missing_output_names_itself(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    """No half measurement: if an output is missing, the error names it."""
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+    worker_says({"t1.radius": 5.0})
+
+    response = 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), "lot_number": "L1"},
+    )
+
+    assert response.status_code == 422
+    assert "t1.diameter" in response.json()["detail"]
+    assert (await db_session.execute(select(Measurement))).scalars().all() == []
+
+
+@pytest.mark.asyncio
+async def test_a_task_without_a_graph_is_refused(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    await db_session.commit()
+    worker_says({})
+
+    response = 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 response.status_code == 422
+    assert "graph" in response.json()["detail"].lower()