Test copy-on-write dei task: rilettura della collezione in sessione condivisa (#53)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

src/backend/api/middleware/rate_limit.py

@@ -35,6 +35,8 @@ class RateLimitMiddleware(BaseHTTPMiddleware):
     """
 
     LOGIN_PATH = "/api/auth/login"
+    # #59: il login con PIN e' lo stesso attacco (4 cifre: ancora piu' facile da indovinare)
+    LOGIN_PIN_PATH = "/api/auth/login-pin"
     WINDOW_SECONDS = 60
 
     # Paths that carry a supervisor's (capoturno) credentials in the request
@@ -97,7 +99,7 @@ class RateLimitMiddleware(BaseHTTPMiddleware):
             return True
         if method != "POST":
             return False
-        if path == cls.LOGIN_PATH:
+        if path in (cls.LOGIN_PATH, cls.LOGIN_PIN_PATH):
             return True
         return any(p.match(path) for p in cls._SUPERVISOR_CREDENTIAL_PATTERNS)
 

src/backend/api/routers/auth.py

@@ -1,5 +1,6 @@
 """Authentication router - login, logout, profile."""
 from fastapi import APIRouter, Depends, HTTPException, Request, status
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.backend.database import get_db
@@ -8,10 +9,17 @@ from src.backend.models.orm.user import User
 from src.backend.models.api.user import (
     LoginRequest,
     LoginResponse,
+    PinLoginRequest,
+    PinUserItem,
     UserProfileUpdate,
     UserResponse,
 )
-from src.backend.services.auth_service import authenticate_user, login_user, logout_user
+from src.backend.services.auth_service import (
+    authenticate_pin,
+    authenticate_user,
+    login_user,
+    logout_user,
+)
 
 router = APIRouter(prefix="/api/auth", tags=["auth"])
 
@@ -32,6 +40,35 @@ async def login(data: LoginRequest, db: AsyncSession = Depends(get_db)):
     )
 
 
+@router.get("/pin-users", response_model=list[PinUserItem])
+async def pin_users(db: AsyncSession = Depends(get_db)):
+    """Utenti che possono entrare con PIN (#59): attivi, non admin, con PIN.
+
+    Pubblico per costruzione (serve prima del login): espone solo id e nome.
+    """
+    result = await db.execute(
+        select(User).where(
+            User.active == True,  # noqa: E712
+            User.is_admin == False,  # noqa: E712
+            User.pin_hash.is_not(None),
+        )
+    )
+    items = [
+        PinUserItem(id=u.id, display_name=u.display_name or u.username)
+        for u in result.scalars().all()
+    ]
+    items.sort(key=lambda i: i.display_name.casefold())
+    return items
+
+
+@router.post("/login-pin", response_model=LoginResponse)
+async def login_pin(data: PinLoginRequest, db: AsyncSession = Depends(get_db)):
+    """Login con user_id e PIN a 4 cifre (#59); stessa risposta del login."""
+    user = await authenticate_pin(db, data.user_id, data.pin)
+    api_key = await login_user(db, user)
+    return LoginResponse(user=UserResponse.model_validate(user), api_key=api_key)
+
+
 @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
 async def logout(
     request: Request,

src/backend/api/routers/files.py

@@ -61,7 +61,8 @@ def detect_file_type(content: bytes) -> tuple[str, str] | None:
 
 # Top-level folders nothing may DELETE through this router: statistics/reports are
 # audit records, the logo is only ever replaced via POST /api/settings/logo.
-_UNDELETABLE_PREFIXES = ("statistics/", "reports/", "logos/")
+# task_results/: allegati degli esiti dei task (#55), record di audit.
+_UNDELETABLE_PREFIXES = ("statistics/", "reports/", "logos/", "task_results/")
 
 # GET under these needs a role that is meant to see production data, not just
 # any authenticated user (a Maker or MeasurementTec has no business reading

src/backend/api/routers/measurements.py

@@ -147,6 +147,7 @@ async def authorise_measurement(
 
     supervisor = await auth_service.authorise_supervisor(
         db, action.supervisor_username, action.supervisor_password,
+        user_id=action.supervisor_user_id, pin=action.supervisor_pin,  # #59
     )
     measurement = await measurement_service.authorise_measurement(
         db, measurement, supervisor,

src/backend/api/routers/production.py

@@ -27,6 +27,7 @@ router = APIRouter(prefix="/api/production-runs", tags=["production"])
 
 _DERIVED = {
     "seconds_to_next_measurement", "overdue", "server_time", "measurement_task_ids",
+    "close_task_ids",
 }
 
 
@@ -53,6 +54,8 @@ async def _as_response(db: AsyncSession, run: ProductionRun) -> ProductionRunRes
         measurement_task_ids=await production_service.measurement_task_ids(
             db, run.version_id,
         ),
+        # #53: task da compilare alla fine produzione
+        close_task_ids=await production_service.close_task_ids(db, run.version_id),
     )
 
 
@@ -163,6 +166,7 @@ async def pause_production_run(
     run = await production_service.get_run(db, run_id)
     supervisor = await auth_service.authorise_supervisor(
         db, action.supervisor_username, action.supervisor_password,
+        user_id=action.supervisor_user_id, pin=action.supervisor_pin,  # #59
     )
     run = await production_service.pause_run(db, run, user, supervisor, note=action.note)
     return await _as_response(db, run)
@@ -179,6 +183,7 @@ async def resume_production_run(
     run = await production_service.get_run(db, run_id)
     supervisor = await auth_service.authorise_supervisor(
         db, action.supervisor_username, action.supervisor_password,
+        user_id=action.supervisor_user_id, pin=action.supervisor_pin,  # #59
     )
     run = await production_service.resume_run(db, run, user, supervisor, note=action.note)
     return await _as_response(db, run)
@@ -195,6 +200,7 @@ async def close_production_run(
     run = await production_service.get_run(db, run_id)
     supervisor = await auth_service.authorise_supervisor(
         db, action.supervisor_username, action.supervisor_password,
+        user_id=action.supervisor_user_id, pin=action.supervisor_pin,  # #59
     )
     run = await production_service.close_run(db, run, user, supervisor, note=action.note)
     return await _as_response(db, run)

src/backend/api/routers/task_results.py

@@ -0,0 +1,192 @@
+"""Task results router: l'esito di un task 'check' (OK/NOK) o 'note' (#54 #55).
+
+Un 'check' registra ok o nok (nok vuole la nota); un 'note' registra 'done' con i
+valori dei campi extra e gli allegati. I task di misura non passano di qui: le
+loro letture sono Measurement.
+"""
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+from fastapi import APIRouter, Depends, File, HTTPException, Query, 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.api.routers.files import (
+    _UPLOAD_CHUNK_SIZE, detect_file_type, sanitize_filename,
+)
+from src.backend.config import settings
+from src.backend.database import get_db
+from src.backend.models.api.task_result import TaskResultCreate, TaskResultResponse
+from src.backend.models.orm.production import ProductionRun
+from src.backend.models.orm.task import RecipeTask
+from src.backend.models.orm.task_result import TaskResult
+from src.backend.models.orm.user import User
+
+router = APIRouter(prefix="/api/task-results", tags=["task-results"])
+
+
+def _unprocessable(detail: str) -> HTTPException:
+    # 422 con detail testuale: l'api_client del frontend legge solo `detail`
+    return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=detail)
+
+
+# Radice degli allegati degli esiti: sotto upload_path, mai cancellabile (audit).
+ATTACHMENT_DIR = "task_results"
+
+
+def _safe_relative_path(path: str) -> bool:
+    """Allegato: relativo, dentro task_results/, senza '..', unita' o backslash."""
+    if path.startswith("/") or "\\" in path or ":" in path:
+        return False
+    parts = path.split("/")
+    return ".." not in parts and parts[0] == ATTACHMENT_DIR and len(parts) > 1
+
+
+@router.post("/attachments", status_code=status.HTTP_201_CREATED)
+async def upload_attachment(
+    file: UploadFile = File(...),
+    _user: User = Depends(require_measurement_tec),
+):
+    """Carica l'allegato di un task 'note' (l'operatore non ha accesso a /api/files/upload).
+
+    Stessi controlli di upload_file: tipo dai magic bytes, limite di dimensione
+    con lettura a blocchi, nome ripulito. Salva sotto task_results/<AAAA>/<MM>/.
+    """
+    max_bytes = settings.max_upload_size_mb * 1024 * 1024
+    buffer = bytearray()
+    while True:
+        chunk = await file.read(_UPLOAD_CHUNK_SIZE)
+        if not chunk:
+            break
+        buffer.extend(chunk)
+        if len(buffer) > max_bytes:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"File size exceeds maximum {settings.max_upload_size_mb}MB",
+            )
+    content = bytes(buffer)
+
+    detected = detect_file_type(content)
+    if detected is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="File type not allowed. Must be a recognised image or PDF.",
+        )
+    _mime, ext = detected
+
+    # nome originale ripulito (per mostrarlo); sul disco solo uuid + estensione rilevata
+    name = sanitize_filename(file.filename or "")
+    now = datetime.now(timezone.utc)
+    storage_dir = settings.upload_path / ATTACHMENT_DIR / f"{now:%Y}" / f"{now:%m}"
+    storage_dir.mkdir(parents=True, exist_ok=True)
+    target = storage_dir / f"{uuid.uuid4().hex}{ext}"
+    target.write_bytes(content)
+
+    return {
+        "file_path": str(target.relative_to(settings.upload_path)).replace("\\", "/"),
+        "name": name,
+    }
+
+
+@router.post("", response_model=TaskResultResponse, status_code=status.HTTP_201_CREATED)
+async def create_task_result(
+    data: TaskResultCreate,
+    user: User = Depends(require_measurement_tec),
+    db: AsyncSession = Depends(get_db),
+):
+    """Registra l'esito di un task 'check' o 'note'."""
+    task = (await db.execute(
+        select(RecipeTask).where(RecipeTask.id == data.task_id)
+    )).scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
+
+    note = data.note.strip() if data.note else None
+    note = note or None
+    values: Optional[dict] = None
+    attachments: Optional[list] = None
+
+    if task.task_type == "check":
+        if data.outcome not in ("ok", "nok"):
+            raise _unprocessable("Un controllo si registra come ok o nok")
+        if data.outcome == "nok" and not note:
+            raise _unprocessable("Per un NOK serve la nota")
+    elif task.task_type == "note":
+        if data.outcome != "done":
+            raise _unprocessable("Una nota si registra come done")
+        fields = {f["key"]: f for f in (task.fields or [])}
+        values = {}
+        for key, value in (data.values or {}).items():
+            if key not in fields:
+                raise _unprocessable(f"Campo '{key}' non previsto dal task")
+            if fields[key]["type"] == "number" and value.strip() != "":
+                try:
+                    float(value.replace(",", "."))
+                except ValueError:
+                    raise _unprocessable(f"Il campo '{fields[key]['label']}' deve essere un numero")
+            values[key] = value
+        attachments = []
+        for att in data.attachments or []:
+            if not _safe_relative_path(att.path):
+                raise _unprocessable(
+                    f"Percorso allegato non valido (deve stare in {ATTACHMENT_DIR}/): {att.path}"
+                )
+            if att.key is not None and att.key not in fields:
+                raise _unprocessable(f"Campo allegato '{att.key}' non previsto dal task")
+            attachments.append(att.model_dump())
+        values = values or None
+        attachments = attachments or None
+    else:
+        raise _unprocessable(
+            f"I task di tipo '{task.task_type}' non registrano un esito"
+        )
+
+    if data.production_run_id is not None:
+        run = (await db.execute(
+            select(ProductionRun).where(ProductionRun.id == data.production_run_id)
+        )).scalar_one_or_none()
+        if run is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail="Production run not found"
+            )
+        if run.status == "closed":
+            raise _unprocessable("La produzione e' gia' chiusa")
+        if run.version_id != task.version_id:
+            raise _unprocessable("Il task non appartiene alla versione di questa produzione")
+
+    result = TaskResult(
+        task_id=task.id,
+        version_id=task.version_id,  # presa dal task, non dal client
+        production_run_id=data.production_run_id,
+        user_id=user.id,
+        outcome=data.outcome,
+        note=note,
+        values=values,
+        attachments=attachments,
+    )
+    db.add(result)
+    await db.flush()
+    await db.refresh(result)  # created_at e' un server_default
+    return TaskResultResponse.model_validate(result)
+
+
+@router.get("", response_model=list[TaskResultResponse])
+async def list_task_results(
+    production_run_id: Optional[int] = Query(None, gt=0),
+    task_id: Optional[int] = Query(None, gt=0),
+    _user: User = Depends(require_measurement_tec),
+    db: AsyncSession = Depends(get_db),
+):
+    """Esiti registrati, filtrabili per produzione e/o task; il piu' recente prima."""
+    stmt = select(TaskResult)
+    if production_run_id is not None:
+        stmt = stmt.where(TaskResult.production_run_id == production_run_id)
+    if task_id is not None:
+        stmt = stmt.where(TaskResult.task_id == task_id)
+    rows = (await db.execute(
+        stmt.order_by(TaskResult.created_at.desc(), TaskResult.id.desc())
+    )).scalars().all()
+    return [TaskResultResponse.model_validate(r) for r in rows]

src/backend/api/routers/tasks.py

@@ -17,6 +17,9 @@ from src.backend.models.api.task import (
     TaskReorderRequest,
     TaskResponse,
     TaskUpdate,
+    check_task_loop_options,
+    default_in_loop,
+    normalize_note_fields,
     validate_tolerance_order,
 )
 from src.backend.models.orm.vision import VisionReferenceImage
@@ -202,6 +205,13 @@ async def create_task(
     This creates a NEW version via copy-on-write, then appends the task to
     the new version. Requires Maker role.
     """
+    # #53 #55: validazione di in_loop/at_close/fields prima di toccare il db
+    try:
+        fields = normalize_note_fields(data.fields)
+        check_task_loop_options(data.task_type, data.at_close, fields)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+
     # Ensure recipe exists and is active
     result = await db.execute(select(Recipe).where(Recipe.id == recipe_id))
     recipe = result.scalar_one_or_none()
@@ -240,6 +250,10 @@ async def create_task(
         file_type=data.file_type,
         annotations_json=data.annotations_json,
         vision_json=data.vision_json,
+        # #53 #55: ciclo (default dal tipo), chiusura, campi extra
+        in_loop=data.in_loop if data.in_loop is not None else default_in_loop(data.task_type),
+        at_close=data.at_close,
+        fields=fields,
     )
     db.add(new_task)
     await db.flush()
@@ -341,6 +355,26 @@ async def update_task(
     if not update_data:
         return TaskResponse.model_validate(task)
 
+    # #53 #55: i tre campi nuovi si validano sulla combinazione finale (valori
+    # nuovi, o quelli gia' sul task per cio' che il client non manda)
+    if "fields" in update_data:
+        try:
+            update_data["fields"] = normalize_note_fields(update_data["fields"])
+        except ValueError as exc:
+            raise HTTPException(status_code=422, detail=str(exc))
+    for flag in ("in_loop", "at_close"):
+        if flag in update_data and update_data[flag] is None:
+            del update_data[flag]  # i booleani non sono nullable: null = non toccare
+    final_type = update_data.get("task_type") or task.task_type
+    try:
+        check_task_loop_options(
+            final_type,
+            update_data.get("at_close", task.at_close),
+            update_data.get("fields", task.fields),
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+
     task = await _task_for_write(db, task, user)
 
     for field, value in update_data.items():

src/backend/api/routers/users.py

@@ -6,8 +6,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from src.backend.database import get_db
 from src.backend.api.middleware.api_key import require_admin_user
 from src.backend.models.orm.user import User
-from src.backend.models.api.user import UserCreate, UserPasswordChange, UserResponse, UserUpdate
-from src.backend.services.auth_service import create_user, hash_password, regenerate_api_key, revoke_all_keys
+from src.backend.models.api.user import PinSet, UserCreate, UserPasswordChange, UserResponse, UserUpdate
+from src.backend.services.auth_service import (
+    create_user,
+    hash_password,
+    regenerate_api_key,
+    revoke_all_keys,
+    set_user_pin,
+)
 
 router = APIRouter(prefix="/api/users", tags=["users"])
 
@@ -151,6 +157,30 @@ async def change_user_password(
     return {"message": f"Password changed for user {user.username}"}
 
 
+@router.put("/{user_id}/pin", status_code=status.HTTP_200_OK)
+async def change_user_pin(
+    user_id: int,
+    data: PinSet,
+    admin: User = Depends(require_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Imposta o toglie (pin null) il PIN di un utente (admin, #59/#51).
+
+    422 se l'utente e' admin. Il PIN non si restituisce mai.
+    """
+    result = await db.execute(select(User).where(User.id == user_id))
+    user = result.scalar_one_or_none()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    if user.is_admin:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Admin users cannot have a PIN",
+        )
+    await set_user_pin(db, user, data.pin)
+    return {"message": f"PIN {'set' if data.pin else 'removed'} for user {user.username}"}
+
+
 @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def deactivate_user(
     user_id: int,

src/backend/main.py

@@ -23,6 +23,7 @@ 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
+from src.backend.api.routers.task_results import router as task_results_router
 
 
 @asynccontextmanager
@@ -77,6 +78,7 @@ app.include_router(setup_router)
 app.include_router(stations_router)
 app.include_router(production_router)
 app.include_router(vision_router)
+app.include_router(task_results_router)
 
 
 @app.get("/api/health")

src/backend/migrations/versions/017_user_pin.py

@@ -0,0 +1,34 @@
+"""users: PIN a 4 cifre per login rapido e capoturno (#59)
+
+pin_hash (bcrypt, mai in chiaro), contatore errori consecutivi e fine blocco.
+
+Revision ID: 017_user_pin
+Revises: 016_vision_recipe_dir
+Create Date: 2026-10-07
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '017_user_pin'
+down_revision: Union[str, None] = '016_vision_recipe_dir'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # batch: SQLite non sa fare DROP COLUMN nativo su tutte le versioni
+    with op.batch_alter_table('users') as batch:
+        batch.add_column(sa.Column('pin_hash', sa.String(255), nullable=True))
+        batch.add_column(sa.Column(
+            'pin_failed_count', sa.Integer(), nullable=False, server_default='0'))
+        batch.add_column(sa.Column('pin_locked_until', sa.DateTime(), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('users') as batch:
+        batch.drop_column('pin_locked_until')
+        batch.drop_column('pin_failed_count')
+        batch.drop_column('pin_hash')

src/backend/migrations/versions/018_task_loop_results.py

@@ -0,0 +1,108 @@
+"""task: in_loop / at_close / fields, tipo 'check', tabella task_results (#53 #54 #55)
+
+in_loop: il task fa parte del ciclo di produzione (prima: implicito dal tipo).
+at_close: il task 'note' si compila alla fine produzione.
+fields: campi extra del task 'note' (JSON).
+task_results: esito registrato da un task 'check' (OK/NOK) o 'note' (done).
+
+Revision ID: 018_task_loop_results
+Revises: 017_user_pin
+Create Date: 2026-10-07
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '018_task_loop_results'
+down_revision: Union[str, None] = '017_user_pin'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+OLD_TYPES = ('note', 'measure', 'drawing', 'dxf_compare', 'camera_measure')
+NEW_TYPES = OLD_TYPES + ('check',)
+# stessi tipi di MEASURING_TASK_TYPES: fino a oggi erano l'unico ciclo possibile
+LOOP_TYPES = ('measure', 'camera_measure', 'dxf_compare')
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('recipe_tasks') as batch:
+        batch.add_column(sa.Column(
+            'in_loop', sa.Boolean(), nullable=False, server_default=sa.false()))
+        batch.add_column(sa.Column(
+            'at_close', sa.Boolean(), nullable=False, server_default=sa.false()))
+        batch.add_column(sa.Column('fields', sa.JSON(), nullable=True))
+        # enum allargato come in 011: 'check' = Controllo OK/NOK
+        batch.alter_column(
+            'task_type',
+            existing_type=sa.Enum(*OLD_TYPES, name='task_type_enum'),
+            type_=sa.Enum(*NEW_TYPES, name='task_type_enum'),
+            existing_nullable=False,
+            existing_server_default='note',
+        )
+
+    # i task di misura erano gia' nel ciclo: lo restano
+    recipe_tasks = sa.table(
+        'recipe_tasks', sa.column('task_type', sa.String), sa.column('in_loop', sa.Boolean))
+    op.execute(
+        recipe_tasks.update()
+        .where(recipe_tasks.c.task_type.in_(LOOP_TYPES))
+        .values(in_loop=True)
+    )
+
+    op.create_table(
+        'task_results',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('task_id', sa.Integer(),
+                  sa.ForeignKey('recipe_tasks.id', ondelete='CASCADE'), nullable=False),
+        sa.Column('version_id', sa.Integer(),
+                  sa.ForeignKey('recipe_versions.id'), nullable=False),
+        sa.Column('production_run_id', sa.Integer(),
+                  sa.ForeignKey('production_runs.id', ondelete='SET NULL'), nullable=True),
+        sa.Column('user_id', sa.Integer(), sa.ForeignKey('users.id'), nullable=False),
+        sa.Column('outcome', sa.Enum('ok', 'nok', 'done', name='task_outcome_enum'),
+                  nullable=False),
+        sa.Column('note', sa.Text(), nullable=True),
+        sa.Column('values', sa.JSON(), nullable=True),
+        sa.Column('attachments', sa.JSON(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), nullable=False,
+                  server_default=sa.func.now()),
+        mysql_engine='InnoDB',
+        mysql_charset='utf8mb4',
+    )
+    op.create_index('ix_task_results_task_id', 'task_results', ['task_id'])
+    op.create_index('ix_task_results_version_id', 'task_results', ['version_id'])
+    op.create_index(
+        'ix_task_results_production_run_id', 'task_results', ['production_run_id'])
+
+
+def downgrade() -> None:
+    # il tipo 'check' non esiste prima di 018: se un task lo usa, tornare indietro
+    # lo lascerebbe con un valore fuori enum
+    bind = op.get_bind()
+    used = bind.execute(
+        sa.text("SELECT COUNT(*) FROM recipe_tasks WHERE task_type = 'check'")
+    ).scalar()
+    if used:
+        raise RuntimeError(
+            f"Downgrade 018 impossibile: {used} task usano il tipo 'check'. "
+            "Cambiare il tipo o eliminare quei task prima di tornare indietro."
+        )
+
+    op.drop_index('ix_task_results_production_run_id', table_name='task_results')
+    op.drop_index('ix_task_results_version_id', table_name='task_results')
+    op.drop_index('ix_task_results_task_id', table_name='task_results')
+    op.drop_table('task_results')
+
+    with op.batch_alter_table('recipe_tasks') as batch:
+        batch.alter_column(
+            'task_type',
+            existing_type=sa.Enum(*NEW_TYPES, name='task_type_enum'),
+            type_=sa.Enum(*OLD_TYPES, name='task_type_enum'),
+            existing_nullable=False,
+            existing_server_default='note',
+        )
+        batch.drop_column('fields')
+        batch.drop_column('at_close')
+        batch.drop_column('in_loop')

src/backend/models/api/measurement.py

@@ -2,7 +2,7 @@
 from src.backend.models.api.tipi import UtcDateTime, UtcNaiveInput
 from typing import Optional
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 
 class MeasurementCreate(BaseModel):
@@ -71,8 +71,21 @@ class MeasurementListResponse(BaseModel):
 class MeasurementAuthorisation(BaseModel):
     """Credentials of the supervisor allowing an out-of-tolerance value to stand."""
 
-    supervisor_username: str = Field(..., min_length=1)
-    supervisor_password: str = Field(..., min_length=1)
+    # #59: o username+password, oppure supervisor_user_id+supervisor_pin
+    supervisor_username: Optional[str] = Field(None, min_length=1)
+    supervisor_password: Optional[str] = Field(None, min_length=1)
+    supervisor_user_id: Optional[int] = Field(None, gt=0)
+    supervisor_pin: Optional[str] = Field(None, pattern=r"^[0-9]{4}$")
+
+    @model_validator(mode="after")
+    def _one_credential_pair(self):
+        by_password = self.supervisor_username and self.supervisor_password
+        by_pin = self.supervisor_user_id and self.supervisor_pin
+        if not (by_password or by_pin):
+            raise ValueError(
+                "supervisor credentials required: username+password or user_id+pin"
+            )
+        return self
 
 
 class TaskProgressResponse(BaseModel):

src/backend/models/api/production.py

@@ -2,7 +2,7 @@
 from src.backend.models.api.tipi import UtcDateTime
 from typing import Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 
 class ProductionRunCreate(BaseModel):
@@ -55,6 +55,8 @@ class ProductionRunResponse(BaseModel):
     # needs them to bring the operator back to the measurement when the interval
     # expires, whichever screen they are on, and to know which task closes a cycle.
     measurement_task_ids: list[int] = Field(default_factory=list)
+    # #53: task 'note' con at_close, da compilare prima della modale del capoturno.
+    close_task_ids: list[int] = Field(default_factory=list)
 
 
 class ProductionRunWithEventsResponse(ProductionRunResponse):
@@ -64,8 +66,21 @@ class ProductionRunWithEventsResponse(ProductionRunResponse):
 class SupervisorAction(BaseModel):
     """Credentials of the supervisor authorising a line stop or a close."""
 
-    supervisor_username: str = Field(..., min_length=1)
-    supervisor_password: str = Field(..., min_length=1)
+    # #59: o username+password, oppure supervisor_user_id+supervisor_pin
+    supervisor_username: Optional[str] = Field(None, min_length=1)
+    supervisor_password: Optional[str] = Field(None, min_length=1)
+    supervisor_user_id: Optional[int] = Field(None, gt=0)
+    supervisor_pin: Optional[str] = Field(None, pattern=r"^[0-9]{4}$")
+
+    @model_validator(mode="after")
+    def _one_credential_pair(self):
+        by_password = self.supervisor_username and self.supervisor_password
+        by_pin = self.supervisor_user_id and self.supervisor_pin
+        if not (by_password or by_pin):
+            raise ValueError(
+                "supervisor credentials required: username+password or user_id+pin"
+            )
+        return self
     note: Optional[str] = None
 
 

src/backend/models/api/task.py

@@ -1,9 +1,10 @@
 """Pydantic schemas for RecipeTask and RecipeSubtask operations."""
+import re
 from typing import Any, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
-from src.backend.models.orm.task import TASK_TYPES
+from src.backend.models.orm.task import DEFAULT_LOOP_TASK_TYPES, TASK_TYPES
 
 # Kept in step with the ORM enum so the two cannot drift apart.
 TaskType = Literal[TASK_TYPES]
@@ -34,6 +35,76 @@ def validate_tolerance_order(
             )
 
 
+# #55: campi extra di un task 'note'
+MAX_NOTE_FIELDS = 20
+_FIELD_KEY_RE = re.compile(r"^[a-z0-9_]{1,40}$")
+NOTE_FIELD_TYPES = ("text", "number")
+
+
+def normalize_note_fields(fields: Optional[list]) -> Optional[list[dict]]:
+    """Valida e normalizza `fields` di un task 'note' (#55); lista vuota -> None.
+
+    Solleva ValueError con un messaggio leggibile (il router lo trasforma in 422).
+    """
+    if not fields:
+        return None
+    if len(fields) > MAX_NOTE_FIELDS:
+        raise ValueError(f"Al massimo {MAX_NOTE_FIELDS} campi per task")
+    out: list[dict] = []
+    seen: set[str] = set()
+    for i, raw in enumerate(fields, start=1):
+        if not isinstance(raw, dict):
+            raise ValueError(f"Campo {i}: formato non valido")
+        key = raw.get("key")
+        if not isinstance(key, str) or not _FIELD_KEY_RE.match(key):
+            raise ValueError(
+                f"Campo {i}: chiave non valida (solo a-z, 0-9, _, da 1 a 40 caratteri)"
+            )
+        if key in seen:
+            raise ValueError(f"Campo {i}: chiave '{key}' duplicata")
+        seen.add(key)
+        label = raw.get("label")
+        label = label.strip() if isinstance(label, str) else ""
+        if not 1 <= len(label) <= 100:
+            raise ValueError(f"Campo '{key}': etichetta da 1 a 100 caratteri")
+        ftype = raw.get("type")
+        if ftype not in NOTE_FIELD_TYPES:
+            raise ValueError(f"Campo '{key}': tipo deve essere text o number")
+        default = raw.get("default")
+        if default is not None:
+            if not isinstance(default, str):
+                raise ValueError(f"Campo '{key}': il valore predefinito deve essere testo")
+            if default == "":
+                default = None
+        if default is not None and ftype == "number":
+            try:
+                float(default.replace(",", "."))
+            except ValueError:
+                raise ValueError(f"Campo '{key}': il valore predefinito non e' un numero")
+        attachment = raw.get("attachment", False)
+        if not isinstance(attachment, bool):
+            raise ValueError(f"Campo '{key}': attachment deve essere true o false")
+        out.append({
+            "key": key, "label": label, "type": ftype,
+            "default": default, "attachment": attachment,
+        })
+    return out
+
+
+def check_task_loop_options(task_type: str, at_close: bool, fields: Optional[list]) -> None:
+    """at_close e fields sono ammessi solo per i task 'note' (#53 #55)."""
+    if task_type != "note":
+        if at_close:
+            raise ValueError("'Alla chiusura' e' ammesso solo per i task di tipo note")
+        if fields:
+            raise ValueError("I campi extra sono ammessi solo per i task di tipo note")
+
+
+def default_in_loop(task_type: str) -> bool:
+    """in_loop quando non dichiarato: misure e check si', gli altri no (#53)."""
+    return task_type in DEFAULT_LOOP_TASK_TYPES
+
+
 class SubtaskCreate(BaseModel):
     """Schema for creating a subtask."""
     marker_number: int = Field(..., ge=1)
@@ -98,6 +169,10 @@ class TaskCreate(BaseModel):
     file_type: Optional[str] = Field(None, pattern="^(image|pdf)$")
     annotations_json: Optional[dict[str, Any]] = None
     vision_json: Optional[dict] = None
+    # #53: assente = default dal tipo (misure e check nel ciclo)
+    in_loop: Optional[bool] = None
+    at_close: bool = False
+    fields: Optional[list[dict[str, Any]]] = None
     subtasks: list[SubtaskCreate] = []
 
 
@@ -111,6 +186,9 @@ class TaskUpdate(BaseModel):
     file_type: Optional[str] = Field(None, pattern="^(image|pdf)$")
     annotations_json: Optional[dict[str, Any]] = None
     vision_json: Optional[dict] = None
+    in_loop: Optional[bool] = None
+    at_close: Optional[bool] = None
+    fields: Optional[list[dict[str, Any]]] = None
 
 
 class TaskResponse(BaseModel):
@@ -129,6 +207,9 @@ class TaskResponse(BaseModel):
     file_type: Optional[str] = None
     annotations_json: Optional[dict[str, Any]] = None
     vision_json: Optional[dict] = None
+    in_loop: bool = False
+    at_close: bool = False
+    fields: Optional[list[dict[str, Any]]] = None
     subtasks: list[SubtaskResponse] = []
 
 

src/backend/models/api/task_result.py

@@ -0,0 +1,37 @@
+"""Pydantic schemas for task results (#54 check OK/NOK, #55 note con campi)."""
+from typing import Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from src.backend.models.api.tipi import UtcDateTime
+
+
+class TaskResultAttachment(BaseModel):
+    """Un allegato: `path` e' il percorso restituito da /api/files/upload."""
+    key: Optional[str] = Field(None, max_length=40)
+    path: str = Field(..., min_length=1, max_length=500)
+    name: str = Field(..., min_length=1, max_length=255)
+
+
+class TaskResultCreate(BaseModel):
+    task_id: int = Field(..., gt=0)
+    production_run_id: Optional[int] = Field(None, gt=0)
+    outcome: Literal["ok", "nok", "done"]
+    note: Optional[str] = None
+    values: Optional[dict[str, str]] = None
+    attachments: Optional[list[TaskResultAttachment]] = Field(None, max_length=50)
+
+
+class TaskResultResponse(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    task_id: int
+    version_id: int
+    production_run_id: Optional[int] = None
+    user_id: int
+    outcome: str
+    note: Optional[str] = None
+    values: Optional[dict[str, str]] = None
+    attachments: Optional[list[TaskResultAttachment]] = None
+    created_at: UtcDateTime

src/backend/models/api/user.py

@@ -50,6 +50,8 @@ class UserResponse(BaseModel):
     active: bool
     created_at: UtcDateTime
     last_login: Optional[UtcDateTime] = None
+    # #59: solo il fatto che il PIN esista, mai l'hash
+    has_pin: bool = False
 
 
 class UserPasswordChange(BaseModel):
@@ -68,3 +70,21 @@ class LoginResponse(BaseModel):
     user: UserResponse
     api_key: str
     message: str = "Login successful"
+
+
+class PinUserItem(BaseModel):
+    """Voce dell'elenco pubblico per il login con PIN (#59): solo id e nome."""
+    id: int
+    display_name: str
+
+
+class PinLoginRequest(BaseModel):
+    """Login con PIN (#59)."""
+    user_id: int
+    pin: str = Field(..., pattern=r"^[0-9]{4}$")
+
+
+class PinSet(BaseModel):
+    """Imposta (4 cifre) o toglie (null) il PIN di un utente (#59)."""
+    pin: Optional[str] = Field(None, pattern=r"^[0-9]{4}$")
+

src/backend/models/orm/__init__.py

@@ -3,6 +3,7 @@ from src.backend.models.orm.user import User
 from src.backend.models.orm.api_key import ApiKey
 from src.backend.models.orm.recipe import Recipe, RecipeVersion
 from src.backend.models.orm.task import RecipeTask, RecipeSubtask
+from src.backend.models.orm.task_result import TaskResult
 from src.backend.models.orm.measurement import Measurement
 from src.backend.models.orm.access_log import AccessLog
 from src.backend.models.orm.setting import SystemSetting, RecipeVersionAudit
@@ -17,6 +18,7 @@ __all__ = [
     "RecipeVersion",
     "RecipeTask",
     "RecipeSubtask",
+    "TaskResult",
     "Measurement",
     "AccessLog",
     "SystemSetting",

src/backend/models/orm/task.py

@@ -2,7 +2,7 @@
 from typing import TYPE_CHECKING, Optional
 
 from sqlalchemy import (
-    DECIMAL, Enum, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint,
+    Boolean, DECIMAL, Enum, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint,
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -15,12 +15,22 @@ if TYPE_CHECKING:
 # What a task is, declared by whoever writes the recipe instead of guessed from its
 # contents. The last two are not built yet; they are in the enum from the start
 # because widening a MySQL enum later is an ALTER on a live table.
-TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure")
+# 'check' = Controllo OK/NOK, senza quote (#54).
+TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure", "check")
 
 # Types the operator screen treats as "there are quotes to take here".
 # dxf_compare produce quote come camera_measure (#46)
 MEASURING_TASK_TYPES = ("measure", "camera_measure", "dxf_compare")
 
+# Tipi che entrano nel ciclo di produzione se in_loop non e' dichiarato (#53):
+# le misure di sempre piu' il controllo OK/NOK.
+DEFAULT_LOOP_TASK_TYPES = MEASURING_TASK_TYPES + ("check",)
+
+
+def _default_in_loop(context) -> bool:
+    """Default alla creazione di in_loop: dipende dal tipo del task (#53)."""
+    return context.get_current_parameters().get("task_type", "note") in DEFAULT_LOOP_TASK_TYPES
+
 
 class RecipeTask(Base):
     __tablename__ = "recipe_tasks"
@@ -55,6 +65,17 @@ class RecipeTask(Base):
     # Cartella immutabile (relativa a upload_path) con recipe.json e i file di
     # meta della ricetta VisionSuite v2 (#46).
     vision_recipe_dir: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    # #53: il task fa parte del ciclo di produzione (default dal tipo: misure e
+    # check si', gli altri no; un task 'note' si puo' mettere nel ciclo a mano).
+    in_loop: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=_default_in_loop, server_default="0"
+    )
+    # #53: task 'note' compilato alla fine produzione, prima della modale del capoturno.
+    at_close: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default="0"
+    )
+    # #55: campi extra del task 'note' ([{key,label,type,default,attachment}]).
+    fields: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
 
     # Relationships
     version: Mapped["RecipeVersion"] = relationship(back_populates="tasks")

src/backend/models/orm/task_result.py

@@ -0,0 +1,46 @@
+"""TaskResult: l'esito registrato da un task 'check' (OK/NOK) o 'note' (#54 #55)."""
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, JSON, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.backend.database import Base
+
+TASK_OUTCOMES = ("ok", "nok", "done")
+
+
+class TaskResult(Base):
+    __tablename__ = "task_results"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    task_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("recipe_tasks.id", ondelete="CASCADE"),
+        nullable=False, index=True,
+    )
+    # Copiata dal task: la versione in cui l'esito e' stato dato non cambia piu'.
+    version_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("recipe_versions.id"), nullable=False, index=True
+    )
+    production_run_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("production_runs.id", ondelete="SET NULL"),
+        nullable=True, index=True,
+    )
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("users.id"), nullable=False
+    )
+    outcome: Mapped[str] = mapped_column(
+        Enum(*TASK_OUTCOMES, name="task_outcome_enum"), nullable=False
+    )
+    note: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    # {chiave del campo: valore stringa}; allegati: [{key, path, name}]
+    values: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    attachments: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+    created_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"<TaskResult task={self.task_id} outcome={self.outcome}>"

src/backend/models/orm/user.py

@@ -22,6 +22,13 @@ class User(Base):
     # Admin flag separate from roles
     is_admin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
 
+    # PIN a 4 cifre (#59): solo l'hash; contatore errori e fine blocco per il lockout
+    pin_hash: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
+    pin_failed_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default="0"
+    )
+    pin_locked_until: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
+
     # User preferences
     language_pref: Mapped[str] = mapped_column(
         Enum("it", "en", name="language_enum"), nullable=False, default="it"
@@ -39,6 +46,11 @@ class User(Base):
     )
     last_login: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
 
+    @property
+    def has_pin(self) -> bool:
+        """True se l'utente ha un PIN impostato (#59); serve a UserResponse.has_pin."""
+        return self.pin_hash is not None
+
     def has_role(self, role: str) -> bool:
         """Check if user has a specific role."""
         return role in (self.roles or [])

src/backend/services/auth_service.py

@@ -54,7 +54,79 @@ async def authenticate_user(
     return user
 
 
-async def authorise_supervisor(db: AsyncSession, username: str, password: str) -> User:
+# #59: PIN a 4 cifre. Hash identico alle password (bcrypt); 5 errori -> blocco 15 minuti.
+PIN_MAX_FAILURES = 5
+PIN_LOCK_MINUTES = 15
+
+
+async def set_user_pin(db: AsyncSession, user: User, pin: str | None) -> None:
+    """Imposta o toglie (None) il PIN; azzera contatore e blocco. Admin esclusi (#59)."""
+    if pin is not None and user.is_admin:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Admin users cannot have a PIN",
+        )
+    user.pin_hash = hash_password(pin) if pin is not None else None
+    user.pin_failed_count = 0
+    user.pin_locked_until = None
+    await db.flush()
+
+
+async def authenticate_pin(db: AsyncSession, user_id: int, pin: str) -> User:
+    """Verifica user_id+PIN con blocco dopo 5 errori consecutivi (#59).
+
+    401 PIN errato (o utente non idoneo: stessa risposta, niente enumerazione),
+    423 utente bloccato. Il contatore va committato prima di sollevare: get_db fa
+    rollback su eccezione e l'errore andrebbe perso.
+    """
+    result = await db.execute(
+        select(User).where(User.id == user_id, User.active == True)  # noqa: E712
+    )
+    user = result.scalar_one_or_none()
+    if user is None or user.is_admin or user.pin_hash is None:
+        verify_password(pin, _DUMMY_PASSWORD_HASH)  # stesso costo di un PIN errato
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid PIN"
+        )
+    now = utcnow()
+    if user.pin_locked_until is not None:
+        if user.pin_locked_until > now:
+            raise HTTPException(
+                status_code=status.HTTP_423_LOCKED,
+                detail="PIN locked until "
+                + user.pin_locked_until.strftime("%Y-%m-%d %H:%M:%S")
+                + " UTC",
+            )
+        # blocco scaduto: si riparte da zero
+        user.pin_locked_until = None
+        user.pin_failed_count = 0
+    if verify_password(pin, user.pin_hash):
+        if user.pin_failed_count:
+            user.pin_failed_count = 0
+            await db.flush()
+        return user
+    user.pin_failed_count += 1
+    locked = user.pin_failed_count >= PIN_MAX_FAILURES
+    if locked:
+        user.pin_locked_until = now + timedelta(minutes=PIN_LOCK_MINUTES)
+    await db.commit()
+    if locked:
+        raise HTTPException(
+            status_code=status.HTTP_423_LOCKED,
+            detail="PIN locked until "
+            + user.pin_locked_until.strftime("%Y-%m-%d %H:%M:%S")
+            + " UTC",
+        )
+    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid PIN")
+
+
+async def authorise_supervisor(
+    db: AsyncSession,
+    username: str | None = None,
+    password: str | None = None,
+    user_id: int | None = None,
+    pin: str | None = None,
+) -> User:
     """Check the credentials of the supervisor authorising something.
 
     Used wherever the shop floor needs a capoturno: a line stop, a closure, an
@@ -65,7 +137,16 @@ async def authorise_supervisor(db: AsyncSession, username: str, password: str) -
     Uses authenticate_user rather than a full login: logging in would issue a
     needless API key for the supervisor.
     """
-    supervisor = await authenticate_user(db, username, password)
+    if user_id is not None and pin is not None:
+        # #59: via PIN, stesso blocco del login con PIN (401/423 sollevati da li')
+        supervisor = await authenticate_pin(db, user_id, pin)
+    elif username is not None and password is not None:
+        supervisor = await authenticate_user(db, username, password)
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Supervisor credentials required",
+        )
     if supervisor is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,

src/backend/services/production_service.py

@@ -145,7 +145,21 @@ async def measurement_task_ids(db: AsyncSession, version_id: int) -> list[int]:
         select(RecipeTask.id)
         .where(
             RecipeTask.version_id == version_id,
-            RecipeTask.task_type.in_(MEASURING_TASK_TYPES),
+            # #53: il ciclo lo dichiara il task (in_loop), non piu' il solo tipo.
+            RecipeTask.in_loop.is_(True),
+        )
+        .order_by(RecipeTask.order_index, RecipeTask.id)
+    )
+    return list(result.scalars().all())
+
+
+async def close_task_ids(db: AsyncSession, version_id: int) -> list[int]:
+    """Ids dei task da compilare alla fine produzione (at_close), in ordine (#53)."""
+    result = await db.execute(
+        select(RecipeTask.id)
+        .where(
+            RecipeTask.version_id == version_id,
+            RecipeTask.at_close.is_(True),
         )
         .order_by(RecipeTask.order_index, RecipeTask.id)
     )

src/backend/services/recipe_service.py

@@ -117,6 +117,10 @@ async def _copy_tasks_to_version(
             vision_json=task.vision_json,
             # Cartella immutabile: basta il riferimento (#46).
             vision_recipe_dir=task.vision_recipe_dir,
+            # #53 #55: ciclo, chiusura e campi extra seguono il task nella nuova versione.
+            in_loop=task.in_loop,
+            at_close=task.at_close,
+            fields=task.fields,
         )
         db.add(new_task)
         await db.flush()  # get new_task.id

src/backend/tests/test_migrations.py

@@ -149,3 +149,87 @@ class TestVisionRecipeDirMigration:
         assert "vision_recipe_dir" not in cols()
         command.upgrade(cfg, "head")  # e di nuovo su
         assert "vision_recipe_dir" in cols()
+
+
+class TestUserPinMigration:
+    """017_user_pin: users.pin_hash / pin_failed_count / pin_locked_until, su e giu (#59)."""
+
+    def test_017_upgrade_and_downgrade(self, sqlite_database_url):
+        db_path, _ = sqlite_database_url
+        cfg = _alembic_config()
+        new = {"pin_hash", "pin_failed_count", "pin_locked_until"}
+
+        def cols():
+            return {c["name"] for c in _inspect(db_path).get_columns("users")}
+
+        command.upgrade(cfg, "head")
+        assert new <= cols()
+        command.downgrade(cfg, "016_vision_recipe_dir")
+        assert not (new & cols())
+        command.upgrade(cfg, "head")  # e di nuovo su
+        assert new <= cols()
+
+
+_SEED_USER_RECIPE_VERSION = [
+    "INSERT INTO users (username, password_hash, display_name, roles, is_admin, active,"
+    " language_pref, theme_pref) VALUES ('u', 'x', 'U', '[]', 1, 1, 'en', 'light')",
+    "INSERT INTO recipes (code, name, active, created_by) VALUES ('R', 'R', 1, 1)",
+    "INSERT INTO recipe_versions (recipe_id, version_number, is_current, created_by)"
+    " VALUES (1, 1, 1, 1)",
+]
+
+
+class TestTaskLoopResultsMigration:
+    """018_task_loop_results: in_loop/at_close/fields, tipo check, task_results (#53 #54 #55)."""
+
+    def test_018_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 {"in_loop", "at_close", "fields"} <= cols()
+        assert "task_results" in _inspect(db_path).get_table_names()
+        command.downgrade(cfg, "017_user_pin")
+        assert not ({"in_loop", "at_close", "fields"} & cols())
+        assert "task_results" not in _inspect(db_path).get_table_names()
+        command.upgrade(cfg, "head")  # e di nuovo su
+
+    def test_018_marks_existing_measure_tasks_in_loop(self, sqlite_database_url):
+        db_path, _ = sqlite_database_url
+        cfg = _alembic_config()
+        command.upgrade(cfg, "017_user_pin")
+        engine = create_engine(f"sqlite:///{db_path}")
+        with engine.begin() as conn:
+            for stmt in _SEED_USER_RECIPE_VERSION:
+                conn.execute(text(stmt))
+            for i, t in enumerate(("measure", "camera_measure", "dxf_compare", "note", "drawing")):
+                conn.execute(text(
+                    "INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
+                    f" VALUES (1, {i}, '{t}', '{t}')"))
+        engine.dispose()
+
+        command.upgrade(cfg, "head")
+        engine = create_engine(f"sqlite:///{db_path}")
+        with engine.connect() as conn:
+            rows = dict(conn.execute(text("SELECT task_type, in_loop FROM recipe_tasks")).all())
+        engine.dispose()
+        assert rows == {"measure": 1, "camera_measure": 1, "dxf_compare": 1, "note": 0, "drawing": 0}
+
+    def test_018_downgrade_refused_when_a_check_task_exists(self, sqlite_database_url):
+        db_path, _ = sqlite_database_url
+        cfg = _alembic_config()
+        command.upgrade(cfg, "head")
+        engine = create_engine(f"sqlite:///{db_path}")
+        with engine.begin() as conn:
+            for stmt in _SEED_USER_RECIPE_VERSION:
+                conn.execute(text(stmt))
+            conn.execute(text(
+                "INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
+                " VALUES (1, 0, 'check', 'c')"))
+        engine.dispose()
+
+        with pytest.raises(RuntimeError, match="check"):
+            command.downgrade(cfg, "017_user_pin")

src/backend/tests/test_pin.py

@@ -0,0 +1,167 @@
+"""PIN a 4 cifre: login rapido, blocco, capoturno via PIN (#59)."""
+from datetime import timedelta
+
+import pytest
+from fastapi import HTTPException
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from src.backend.models.orm.user import User
+from src.backend.services import auth_service
+from src.backend.tests.conftest import _create_user, auth_headers
+from src.backend.tests.test_production_runs import (
+    _open, _recipe_with_interval, _station,
+)
+from src.backend.utils.tempo import utcnow
+
+
+async def _pin_user(db_session, username="operaio", pin="1234", **kw) -> User:
+    user = await _create_user(db_session, username=username, **kw)
+    await auth_service.set_user_pin(db_session, user, pin)
+    await db_session.commit()
+    return user
+
+
+async def _login_pin(client, user_id, pin):
+    return await client.post("/api/auth/login-pin", json={"user_id": user_id, "pin": pin})
+
+
+async def test_login_pin_ok_same_shape_as_login(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    resp = await _login_pin(client, user.id, "1234")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["user"]["username"] == "operaio"
+    assert data["user"]["has_pin"] is True
+    assert data["api_key"]
+    assert "pin_hash" not in data["user"]
+    # la api_key vale davvero
+    me = await client.get("/api/auth/me", headers={"X-API-Key": data["api_key"]})
+    assert me.status_code == 200
+
+
+async def test_login_pin_wrong_is_401(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    assert (await _login_pin(client, user.id, "9999")).status_code == 401
+
+
+async def test_login_pin_format_422(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    for bad in ("123", "12345", "12a4", ""):
+        assert (await _login_pin(client, user.id, bad)).status_code == 422
+
+
+async def test_lock_after_five_errors_even_with_right_pin(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    for _ in range(4):
+        assert (await _login_pin(client, user.id, "0000")).status_code == 401
+    resp = await _login_pin(client, user.id, "0000")  # 5 errore: scatta il blocco
+    assert resp.status_code == 423
+    assert "locked until" in resp.json()["detail"]
+    assert (await _login_pin(client, user.id, "1234")).status_code == 423
+
+
+async def test_lock_expires_and_counter_resets(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    for _ in range(5):
+        await _login_pin(client, user.id, "0000")
+    row = (await db_session.execute(select(User).where(User.id == user.id))).scalar_one()
+    await db_session.refresh(row)
+    assert row.pin_locked_until > utcnow() + timedelta(minutes=14)
+    # tempo simulato: il blocco e scaduto un secondo fa
+    row.pin_locked_until = utcnow() - timedelta(seconds=1)
+    await db_session.commit()
+    assert (await _login_pin(client, user.id, "1234")).status_code == 200
+    await db_session.refresh(row)
+    assert row.pin_failed_count == 0 and row.pin_locked_until is None
+
+
+async def test_success_resets_failed_counter(client: AsyncClient, db_session):
+    user = await _pin_user(db_session)
+    for _ in range(3):
+        await _login_pin(client, user.id, "0000")
+    assert (await _login_pin(client, user.id, "1234")).status_code == 200
+    for _ in range(4):  # il contatore e ripartito da zero: 4 errori non bloccano
+        assert (await _login_pin(client, user.id, "0000")).status_code == 401
+
+
+async def test_admin_cannot_have_pin_or_login_with_it(client: AsyncClient, admin_user, db_session):
+    resp = await client.put(
+        f"/api/users/{admin_user.id}/pin", headers=auth_headers(admin_user), json={"pin": "1234"},
+    )
+    assert resp.status_code == 422
+    # anche un admin con pin_hash scritto a mano non entra da login-pin
+    admin_user.pin_hash = auth_service.hash_password("1234")
+    await db_session.commit()
+    assert (await _login_pin(client, admin_user.id, "1234")).status_code == 401
+    listed = (await client.get("/api/auth/pin-users")).json()
+    assert all(u["id"] != admin_user.id for u in listed)
+
+
+async def test_put_pin_set_and_remove(client: AsyncClient, admin_user, db_session):
+    user = await _create_user(db_session, username="senzapin")
+    await db_session.commit()
+    h = auth_headers(admin_user)
+    assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": "4321"})).status_code == 200
+    assert (await _login_pin(client, user.id, "4321")).status_code == 200
+    assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": "12"})).status_code == 422
+    assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": None})).status_code == 200
+    assert (await _login_pin(client, user.id, "4321")).status_code == 401
+    assert (await client.put("/api/users/99999/pin", headers=h, json={"pin": "1234"})).status_code == 404
+    # non admin: 403
+    assert (await client.put(
+        f"/api/users/{user.id}/pin", headers=auth_headers(user), json={"pin": "1234"},
+    )).status_code == 403
+
+
+async def test_pin_users_list(client: AsyncClient, db_session):
+    a = await _pin_user(db_session, username="zeta", display_name="Zeta Rossi", pin="1111")
+    b = await _pin_user(db_session, username="alfa", display_name="Alfa Bianchi", pin="2222")
+    await _create_user(db_session, username="nopin")
+    await _pin_user(db_session, username="inattivo", pin="3333", active=False)
+    await db_session.commit()
+    resp = await client.get("/api/auth/pin-users")  # senza autenticazione
+    assert resp.status_code == 200
+    data = resp.json()
+    assert [u["display_name"] for u in data] == ["Alfa Bianchi", "Zeta Rossi"]
+    assert all(set(u) == {"id", "display_name"} for u in data)
+    assert {u["id"] for u in data} == {a.id, b.id}
+
+
+async def test_supervisor_via_pin(client: AsyncClient, measurement_tec_user, admin_user, db_session):
+    station = await _station(db_session, admin_user.id, code="ST-PIN")
+    recipe = await _recipe_with_interval(db_session, admin_user.id, 10, code="REC-PIN")
+    run_id = (await _open(client, measurement_tec_user, station, recipe)).json()["id"]
+    capo = await _pin_user(db_session, username="capo-pin", pin="5555", roles=["Supervisor"])
+    h = auth_headers(measurement_tec_user)
+
+    bad = await client.post(
+        f"/api/production-runs/{run_id}/pause", headers=h,
+        json={"supervisor_user_id": capo.id, "supervisor_pin": "0000"},
+    )
+    assert bad.status_code == 401
+    ok = await client.post(
+        f"/api/production-runs/{run_id}/pause", headers=h,
+        json={"supervisor_user_id": capo.id, "supervisor_pin": "5555"},
+    )
+    assert ok.status_code == 200
+    # nessuna coppia di credenziali: 422
+    none = await client.post(f"/api/production-runs/{run_id}/resume", headers=h, json={})
+    assert none.status_code == 422
+
+
+async def test_supervisor_pin_role_still_checked(db_session):
+    plain = await _pin_user(db_session, username="non-capo", pin="6666", roles=["Maker"])
+    with pytest.raises(HTTPException) as e:
+        await auth_service.authorise_supervisor(db_session, user_id=plain.id, pin="6666")
+    assert e.value.status_code == 403
+
+
+async def test_supervisor_pin_lockout(db_session):
+    capo = await _pin_user(db_session, username="capo-lock", pin="7777", roles=["Supervisor"])
+    codes = []
+    for pin in ["0000"] * 5 + ["7777"]:
+        with pytest.raises(HTTPException) as e:
+            await auth_service.authorise_supervisor(db_session, user_id=capo.id, pin=pin)
+        codes.append(e.value.status_code)
+    assert codes == [401, 401, 401, 401, 423, 423]

src/backend/tests/test_task_loop_fields.py

@@ -0,0 +1,210 @@
+"""Test di in_loop / at_close / fields e del tipo 'check' sui task (#53 #54 #55),
+di close_task_ids nella risposta del run e del bucket di rate limit del login PIN."""
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from src.backend.api.middleware.rate_limit import RateLimitMiddleware
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.recipe import RecipeVersion
+from src.backend.models.orm.station import Station
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.tests.conftest import auth_headers, create_test_recipe
+
+FIELDS = [
+    {"key": "peso", "label": "Peso", "type": "number", "default": "1.5", "attachment": False},
+    {"key": "foto", "label": "Foto", "type": "text", "default": None, "attachment": True},
+]
+
+
+async def _post_task(client, user, recipe_id, **body):
+    return await client.post(
+        f"/api/recipes/{recipe_id}/tasks", headers=auth_headers(user), json=body,
+    )
+
+
+async def test_create_defaults_in_loop_from_type(client: AsyncClient, maker_user, db_session):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L1")
+    await db_session.commit()
+    expected = {"note": False, "drawing": False, "measure": True,
+                "camera_measure": True, "dxf_compare": True, "check": True}
+    for task_type, in_loop in expected.items():
+        resp = await _post_task(client, maker_user, recipe.id, title=task_type, task_type=task_type)
+        assert resp.status_code == 201, resp.text
+        body = resp.json()
+        assert body["in_loop"] is in_loop, task_type
+        assert body["at_close"] is False
+        assert body["fields"] is None
+
+
+async def test_create_explicit_in_loop_and_note_fields(client: AsyncClient, maker_user, db_session):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L2")
+    await db_session.commit()
+    resp = await _post_task(
+        client, maker_user, recipe.id, title="Nota", task_type="note",
+        in_loop=True, at_close=True, fields=FIELDS,
+    )
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["in_loop"] is True and body["at_close"] is True
+    assert body["fields"] == FIELDS
+
+
+@pytest.mark.parametrize("task_type", ["measure", "check", "drawing"])
+async def test_create_at_close_or_fields_rejected_for_other_types(
+    client: AsyncClient, maker_user, db_session, task_type,
+):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code=f"REC-L3{task_type}")
+    await db_session.commit()
+    resp = await _post_task(client, maker_user, recipe.id, title="x", task_type=task_type, at_close=True)
+    assert resp.status_code == 422
+    assert isinstance(resp.json()["detail"], str)
+    resp = await _post_task(client, maker_user, recipe.id, title="x", task_type=task_type, fields=FIELDS)
+    assert resp.status_code == 422
+
+
+@pytest.mark.parametrize("bad", [
+    [{"key": "A b", "label": "x", "type": "text"}],                      # chiave non slug
+    [{"key": "k", "label": "", "type": "text"}],                         # etichetta vuota
+    [{"key": "k", "label": "x", "type": "date"}],                        # tipo ignoto
+    [{"key": "k", "label": "x", "type": "text"}, {"key": "k", "label": "y", "type": "text"}],  # duplicata
+    [{"key": "k", "label": "x", "type": "number", "default": "abc"}],    # default non numerico
+    [{"key": f"k{i}", "label": "x", "type": "text"} for i in range(21)],  # troppi
+])
+async def test_create_invalid_fields_rejected(client: AsyncClient, maker_user, db_session, bad):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L4")
+    await db_session.commit()
+    resp = await _post_task(client, maker_user, recipe.id, title="n", task_type="note", fields=bad)
+    assert resp.status_code == 422
+    assert isinstance(resp.json()["detail"], str)
+
+
+async def test_update_fields_null_clears_and_type_rules(client: AsyncClient, maker_user, db_session):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L5")
+    await db_session.commit()
+    created = (await _post_task(
+        client, maker_user, recipe.id, title="Nota", task_type="note",
+        at_close=True, fields=FIELDS,
+    )).json()
+    headers = auth_headers(maker_user)
+
+    # il frontend manda sempre i tre campi: fields null = nessun campo
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=headers,
+        json={"in_loop": True, "at_close": False, "fields": None},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["in_loop"] is True and body["at_close"] is False and body["fields"] is None
+
+    # tipo diverso da note con at_close o fields: 422
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=headers,
+        json={"task_type": "measure", "at_close": True, "fields": None},
+    )
+    assert resp.status_code == 422
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=headers,
+        json={"task_type": "check", "at_close": False, "fields": FIELDS},
+    )
+    assert resp.status_code == 422
+
+    # un task che ha gia' at_close non puo' passare a measure senza toglierlo
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=headers,
+        json={"at_close": True, "fields": FIELDS},
+    )
+    assert resp.status_code == 200
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=headers, json={"task_type": "measure"},
+    )
+    assert resp.status_code == 422
+
+
+async def test_copy_on_write_carries_the_three_fields(
+    client: AsyncClient, maker_user, db_session,
+):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L6")
+    await db_session.commit()
+    created = (await _post_task(
+        client, maker_user, recipe.id, title="Nota", task_type="note",
+        in_loop=True, at_close=True, fields=FIELDS,
+    )).json()
+
+    # una misura sulla versione corrente forza il copy-on-write alla prossima modifica
+    version = (await db_session.execute(
+        select(RecipeVersion).where(
+            RecipeVersion.recipe_id == recipe.id, RecipeVersion.is_current == True,  # noqa: E712
+        )
+    )).scalar_one()
+    subtask_id = (await db_session.execute(
+        select(RecipeSubtask.id).join(RecipeSubtask.task)
+        .where(RecipeSubtask.task.has(version_id=version.id))
+    )).scalars().first()
+    db_session.add(Measurement(
+        subtask_id=subtask_id, version_id=version.id, measured_by=maker_user.id,
+        value=10.0, pass_fail="pass", deviation=0.0, input_method="manual",
+    ))
+    await db_session.commit()
+    # sessione condivisa col client: la collezione version.tasks va riletta
+    db_session.expire_all()
+
+    resp = await client.put(
+        f"/api/tasks/{created['id']}", headers=auth_headers(maker_user),
+        json={"title": "Nota rinominata"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["version_id"] != version.id  # e' la copia nella nuova versione
+    assert body["in_loop"] is True and body["at_close"] is True
+    assert body["fields"] == FIELDS
+
+
+async def test_run_response_loop_and_close_task_ids(
+    client: AsyncClient, maker_user, measurement_tec_user, admin_user, db_session,
+):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L7")
+    db_session.add(Station(code="ST-L7", name="S", active=True, created_by=admin_user.id))
+    await db_session.commit()
+
+    async def mk(**body):
+        resp = await _post_task(client, maker_user, recipe.id, **body)
+        assert resp.status_code == 201, resp.text
+        return resp.json()
+
+    check = await mk(title="Check", task_type="check")
+    note_loop = await mk(title="Nota ciclo", task_type="note", in_loop=True)
+    note_out = await mk(title="Nota fuori", task_type="note")
+    close_a = await mk(title="Chiusura A", task_type="note", at_close=True)
+    measure_off = await mk(title="Misura fuori", task_type="measure", in_loop=False)
+    close_b = await mk(title="Chiusura B", task_type="note", at_close=True)
+
+    first_id = (await db_session.execute(
+        select(RecipeTask.id).where(RecipeTask.version_id == check["version_id"])
+        .order_by(RecipeTask.order_index)
+    )).scalars().first()
+
+    resp = await client.post(
+        "/api/production-runs", headers=auth_headers(measurement_tec_user),
+        json={"station_code": "ST-L7", "recipe_id": recipe.id},
+    )
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    # ciclo: solo in_loop, in ordine (misura della fixture, check, nota in ciclo)
+    assert body["measurement_task_ids"] == [first_id, check["id"], note_loop["id"]]
+    assert note_out["id"] not in body["measurement_task_ids"]
+    assert measure_off["id"] not in body["measurement_task_ids"]
+    assert body["close_task_ids"] == [close_a["id"], close_b["id"]]
+
+
+class TestLoginPinRateLimitBucket:
+    """#59: il login con PIN sta nel bucket 'login' come il login con password."""
+
+    def test_login_pin_is_login_bucket(self):
+        assert RateLimitMiddleware._is_login_bucket_request("POST", "/api/auth/login-pin")
+
+    def test_login_pin_get_is_not_login_bucket(self):
+        assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/auth/login-pin")
+
+    def test_pin_users_list_is_not_login_bucket(self):
+        assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/auth/pin-users")

src/backend/tests/test_task_results.py

@@ -0,0 +1,240 @@
+"""Test di POST/GET /api/task-results (#54 check OK/NOK, #55 note con campi)."""
+from unittest.mock import patch
+
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from src.backend.models.orm.production import ProductionRun
+from src.backend.models.orm.station import Station
+from src.backend.tests.conftest import auth_headers, create_test_recipe
+
+FIELDS = [
+    {"key": "peso", "label": "Peso", "type": "number", "default": None, "attachment": False},
+    {"key": "foto", "label": "Foto", "type": "text", "default": None, "attachment": True},
+]
+
+
+async def _setup(client, db_session, maker_user, admin_user, code):
+    recipe = await create_test_recipe(db_session, user_id=maker_user.id, code=code)
+    db_session.add(Station(code=f"ST-{code}", name="S", active=True, created_by=admin_user.id))
+    await db_session.commit()
+    h = auth_headers(maker_user)
+    check = (await client.post(
+        f"/api/recipes/{recipe.id}/tasks", headers=h, json={"title": "C", "task_type": "check"},
+    )).json()
+    note = (await client.post(
+        f"/api/recipes/{recipe.id}/tasks", headers=h,
+        json={"title": "N", "task_type": "note", "fields": FIELDS},
+    )).json()
+    return recipe, check, note
+
+
+async def _open_run(client, user, recipe, code):
+    resp = await client.post(
+        "/api/production-runs", headers=auth_headers(user),
+        json={"station_code": f"ST-{code}", "recipe_id": recipe.id},
+    )
+    assert resp.status_code == 201, resp.text
+    return resp.json()
+
+
+async def _post(client, user, **body):
+    return await client.post("/api/task-results", headers=auth_headers(user), json=body)
+
+
+async def test_requires_auth_and_role(client: AsyncClient, maker_user, admin_user, db_session):
+    _, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR0")
+    resp = await client.post("/api/task-results", json={"task_id": check["id"], "outcome": "ok"})
+    assert resp.status_code == 401
+    # un Maker senza ruolo di reparto non registra esiti
+    resp = await _post(client, maker_user, task_id=check["id"], outcome="ok")
+    assert resp.status_code == 403
+
+
+async def test_check_ok_and_nok_rules(
+    client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
+):
+    recipe, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR1")
+    run = await _open_run(client, measurement_tec_user, recipe, "TR1")
+
+    resp = await _post(client, measurement_tec_user, task_id=check["id"],
+                       production_run_id=run["id"], outcome="ok")
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["outcome"] == "ok"
+    assert body["version_id"] == check["version_id"]
+    assert body["user_id"] == measurement_tec_user.id
+    assert body["production_run_id"] == run["id"]
+    assert body["created_at"]
+
+    for note in (None, "", "   "):
+        resp = await _post(client, measurement_tec_user, task_id=check["id"],
+                           outcome="nok", note=note)
+        assert resp.status_code == 422
+        assert resp.json()["detail"] == "Per un NOK serve la nota"
+
+    resp = await _post(client, measurement_tec_user, task_id=check["id"],
+                       outcome="nok", note="  graffio  ")
+    assert resp.status_code == 201
+    assert resp.json()["note"] == "graffio"
+
+    # un check non si chiude con 'done'
+    resp = await _post(client, measurement_tec_user, task_id=check["id"], outcome="done")
+    assert resp.status_code == 422
+
+
+async def test_note_values_and_attachments(
+    client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
+):
+    recipe, _, note = await _setup(client, db_session, maker_user, admin_user, "TR2")
+    ok = await _post(
+        client, measurement_tec_user, task_id=note["id"], outcome="done",
+        values={"peso": "12,5"},
+        attachments=[{"key": "foto", "path": "task_results/2026/10/abc.jpg", "name": "foto.jpg"}],
+    )
+    assert ok.status_code == 201, ok.text
+    assert ok.json()["values"] == {"peso": "12,5"}
+    assert ok.json()["attachments"] == [{"key": "foto", "path": "task_results/2026/10/abc.jpg", "name": "foto.jpg"}]
+
+    # solo done
+    assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="ok")).status_code == 422
+    # chiave fuori dai campi
+    assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
+                        values={"altro": "1"})).status_code == 422
+    # numero non convertibile
+    assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
+                        values={"peso": "abc"})).status_code == 422
+    # allegati con percorso non sicuro
+    for bad in ("../x.jpg", "/etc/passwd", "a/../b.jpg", "C:/x.jpg", "a\\b.jpg"):
+        resp = await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
+                           attachments=[{"key": None, "path": bad, "name": "x"}])
+        assert resp.status_code == 422, bad
+    # chiave allegato fuori dai campi
+    resp = await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
+                       attachments=[{"key": "boh", "path": "task_results/a.jpg", "name": "x"}])
+    assert resp.status_code == 422
+
+
+async def test_other_task_types_rejected_and_unknown_task(
+    client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
+):
+    recipe, _, _ = await _setup(client, db_session, maker_user, admin_user, "TR3")
+    measure = (await client.post(
+        f"/api/recipes/{recipe.id}/tasks", headers=auth_headers(maker_user),
+        json={"title": "M", "task_type": "measure"},
+    )).json()
+    resp = await _post(client, measurement_tec_user, task_id=measure["id"], outcome="ok")
+    assert resp.status_code == 422
+    resp = await _post(client, measurement_tec_user, task_id=999999, outcome="ok")
+    assert resp.status_code == 404
+
+
+async def test_run_must_be_open_and_same_version(
+    client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
+):
+    recipe, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR4")
+    _, other_check, _ = await _setup(client, db_session, maker_user, admin_user, "TR4b")
+    run = await _open_run(client, measurement_tec_user, recipe, "TR4")
+
+    # task di un'altra versione
+    resp = await _post(client, measurement_tec_user, task_id=other_check["id"],
+                       production_run_id=run["id"], outcome="ok")
+    assert resp.status_code == 422
+    # run inesistente
+    resp = await _post(client, measurement_tec_user, task_id=check["id"],
+                       production_run_id=999999, outcome="ok")
+    assert resp.status_code == 404
+
+    # run chiuso: non registra piu'
+    row = (await db_session.execute(
+        select(ProductionRun).where(ProductionRun.id == run["id"])
+    )).scalar_one()
+    row.status = "closed"
+    await db_session.commit()
+    resp = await _post(client, measurement_tec_user, task_id=check["id"],
+                       production_run_id=run["id"], outcome="ok")
+    assert resp.status_code == 422
+
+
+async def test_list_filters_and_order(
+    client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
+):
+    recipe, check, note = await _setup(client, db_session, maker_user, admin_user, "TR5")
+    run = await _open_run(client, measurement_tec_user, recipe, "TR5")
+    a = (await _post(client, measurement_tec_user, task_id=check["id"],
+                     production_run_id=run["id"], outcome="ok")).json()
+    b = (await _post(client, measurement_tec_user, task_id=note["id"],
+                     production_run_id=run["id"], outcome="done")).json()
+    c = (await _post(client, measurement_tec_user, task_id=check["id"], outcome="ok")).json()
+    h = auth_headers(measurement_tec_user)
+
+    resp = await client.get(f"/api/task-results?production_run_id={run['id']}", headers=h)
+    assert [r["id"] for r in resp.json()] == [b["id"], a["id"]]  # piu' recente prima
+    resp = await client.get(f"/api/task-results?task_id={check['id']}", headers=h)
+    assert [r["id"] for r in resp.json()] == [c["id"], a["id"]]
+    resp = await client.get(
+        f"/api/task-results?task_id={check['id']}&production_run_id={run['id']}", headers=h)
+    assert [r["id"] for r in resp.json()] == [a["id"]]
+    assert (await client.get("/api/task-results")).status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# POST /api/task-results/attachments: l'operatore carica gli allegati (#55)
+# ---------------------------------------------------------------------------
+
+# firma PNG valida (magic bytes) piu' un po' di zeri
+PNG = bytes([0x89]) + b"PNG" + bytes([13, 10, 26, 10]) + bytes(32)
+
+
+async def test_attachment_upload_ok_for_measurement_tec(
+    client: AsyncClient, measurement_tec_user, tmp_path,
+):
+    with patch("src.backend.api.routers.task_results.settings") as mock_settings:
+        mock_settings.upload_path = tmp_path
+        mock_settings.max_upload_size_mb = 50
+        resp = await client.post(
+            "/api/task-results/attachments", headers=auth_headers(measurement_tec_user),
+            files={"file": ("foto bella (1).png", PNG, "image/png")},
+        )
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["name"] == "fotobella1.png"
+    assert body["file_path"].startswith("task_results/")
+    assert body["file_path"].endswith(".png")
+    assert (tmp_path / body["file_path"]).read_bytes() == PNG
+
+
+async def test_attachment_upload_rejects_wrong_type_size_and_role(
+    client: AsyncClient, measurement_tec_user, maker_user, tmp_path,
+):
+    h = auth_headers(measurement_tec_user)
+    with patch("src.backend.api.routers.task_results.settings") as mock_settings:
+        mock_settings.upload_path = tmp_path
+        mock_settings.max_upload_size_mb = 1
+        # tipo non ammesso (HTML dichiarato come png): come upload_file, 400
+        resp = await client.post(
+            "/api/task-results/attachments", headers=h,
+            files={"file": ("x.png", b"<html></html>", "image/png")},
+        )
+        assert resp.status_code == 400
+        # oltre il limite
+        resp = await client.post(
+            "/api/task-results/attachments", headers=h,
+            files={"file": ("x.png", PNG + bytes(1024 * 1024 + 1), "image/png")},
+        )
+        assert resp.status_code == 400
+        # un Maker senza ruolo di reparto non carica
+        resp = await client.post(
+            "/api/task-results/attachments", headers=auth_headers(maker_user),
+            files={"file": ("x.png", PNG, "image/png")},
+        )
+        assert resp.status_code == 403
+    assert not any(tmp_path.rglob("*.png"))
+
+
+async def test_task_results_dir_is_undeletable(client: AsyncClient, admin_user):
+    resp = await client.delete(
+        "/api/files/task_results/2026/10/abc.png", headers=auth_headers(admin_user),
+    )
+    assert resp.status_code in (403, 404)
+    assert resp.status_code == 403

src/frontend/flask_app/templates/maker/task_editor.html

@@ -91,6 +91,74 @@
 {% endblock %}
 
 {% block content %}
+{# #53 #55: parte del form task comune a creazione e modifica; `d` e il nome dello stato Alpine (newTask / editTaskData) #}
+{% macro task_flags_editor(d) %}
+<div class="sm:col-span-2 space-y-3">
+  <!-- #53: il task entra nel ciclo di misura (timer e cicalino) -->
+  <label class="flex items-start gap-2 text-sm cursor-pointer">
+    <input type="checkbox" class="mt-0.5"
+           x-model="{{ d }}.in_loop"
+           @change="{{ d }}.in_loop_touched = true">
+    <span>{{ _('Nel ciclo di misura (timer e cicalino)') }}</span>
+  </label>
+  <!-- #55: opzioni valide solo per il tipo Nota -->
+  <template x-if="{{ d }}.task_type === 'note'">
+    <div class="space-y-3">
+      <label class="flex items-start gap-2 text-sm cursor-pointer">
+        <input type="checkbox" class="mt-0.5" x-model="{{ d }}.at_close">
+        <span>{{ _('Compila a fine produzione') }}</span>
+      </label>
+      <div class="space-y-2">
+        <div class="flex items-center justify-between gap-2">
+          <label class="tmf-label mb-0">{{ _('Campi da compilare') }}</label>
+          <button type="button" class="btn btn-secondary text-xs gap-1"
+                  :disabled="{{ d }}.fields.length >= MAX_NOTE_FIELDS"
+                  @click="addNoteField({{ d }})">
+            + {{ _('Aggiungi campo') }}
+          </button>
+        </div>
+        <template x-for="(f, fi) in {{ d }}.fields" :key="f._uid">
+          <div class="grid grid-cols-1 sm:grid-cols-12 gap-2 p-2 rounded border border-[var(--border-color)] bg-[var(--bg-secondary)]">
+            <div class="sm:col-span-4">
+              <input type="text" class="tmf-input text-sm" maxlength="100"
+                     placeholder="{{ _('Etichetta') }}"
+                     x-model="f.label"
+                     @input="deriveNoteFieldKey({{ d }}, f)">
+              <span class="block mt-0.5 text-[10px] font-mono text-[var(--text-muted)]" x-text="f.key"></span>
+            </div>
+            <div class="sm:col-span-2">
+              <select class="tmf-input text-sm" x-model="f.type">
+                <option value="text">{{ _('Testo') }}</option>
+                <option value="number">{{ _('Numero') }}</option>
+              </select>
+            </div>
+            <div class="sm:col-span-3">
+              <input type="text" class="tmf-input text-sm"
+                     placeholder="{{ _('Valore precompilato') }}"
+                     x-model="f.default">
+            </div>
+            <label class="sm:col-span-1 flex items-center gap-1 text-xs cursor-pointer">
+              <input type="checkbox" x-model="f.attachment">
+              <span>{{ _('Allegato') }}</span>
+            </label>
+            <div class="sm:col-span-2 flex items-center justify-end gap-1">
+              <button type="button" class="p-1 rounded hover:bg-[var(--bg-card-hover)]"
+                      :disabled="fi === 0" @click="moveNoteField({{ d }}, fi, -1)"
+                      title="{{ _('Sposta su') }}">&uarr;</button>
+              <button type="button" class="p-1 rounded hover:bg-[var(--bg-card-hover)]"
+                      :disabled="fi === {{ d }}.fields.length - 1" @click="moveNoteField({{ d }}, fi, 1)"
+                      title="{{ _('Sposta giu') }}">&darr;</button>
+              <button type="button" class="p-1 rounded text-red-500 hover:bg-red-50 dark:hover:bg-red-900/20"
+                      @click="removeNoteField({{ d }}, fi)"
+                      title="{{ _('Rimuovi campo') }}">&times;</button>
+            </div>
+          </div>
+        </template>
+      </div>
+    </div>
+  </template>
+</div>
+{% endmacro %}
 <div class="tmf-page"
      x-data="taskEditor()"
      x-cloak>
@@ -273,12 +341,15 @@
         <!-- Tipo: dichiarato qui, non dedotto dal contenuto -->
         <div>
           <label class="tmf-label">{{ _('Tipo') }} <span class="text-red-500">*</span></label>
-          <select x-model="newTask.task_type" class="tmf-input">
+          <select x-model="newTask.task_type" class="tmf-input"
+                  @change="onTaskTypeChange(newTask, $event.target.value)">
             <option value="note">{{ _('Nota') }}</option>
             <option value="measure">{{ _('Misura') }}</option>
             <option value="drawing">{{ _('Disegno') }}</option>
             <option value="camera_measure">{{ _('Misura camera') }}</option>
             <option value="dxf_compare">{{ _('Confronto profilo') }}</option>
+            <!-- #54: task di controllo con esito OK/NOK, senza quote -->
+            <option value="check">{{ _('Controllo OK/NOK') }}</option>
           </select>
           <p class="mt-1 text-xs text-[var(--text-secondary)]">
             {{ _('Un task di misura resta tale anche prima di inserire le quote') }}
@@ -306,6 +377,8 @@
                     rows="3"
                     placeholder="{{ _('Descrizione opzionale...') }}"></textarea>
         </div>
+        {# #53 #55: ciclo di misura, compilazione a fine produzione e campi della nota #}
+        {{ task_flags_editor('newTask') }}
       </div>
     </div>
     <div class="tmf-card-footer flex items-center gap-2 justify-end">
@@ -387,8 +460,20 @@
                    x-ref="editTaskTitleInput">
           </template>
 
-          <!-- Badge: N misurazioni -->
+          <!-- #53: segno sui task nel ciclo di misura -->
+          <span x-show="task.in_loop" class="shrink-0 text-primary"
+                title="{{ _('Nel ciclo di misura (timer e cicalino)') }}">
+            <svg class="w-4 h-4" fill="none" stroke="currentColor" stroke-width="2" viewBox="0 0 24 24">
+              <path stroke-linecap="round" stroke-linejoin="round"
+                    d="M4 4v5h5M20 20v-5h-5M5.5 15A8 8 0 0018.4 17.4M18.5 9A8 8 0 005.6 6.6"/>
+            </svg>
+          </span>
+          <!-- #55: nota compilata a fine produzione -->
+          <span x-show="task.at_close" class="badge badge-neutral text-xs shrink-0">{{ _('A fine produzione') }}</span>
+
+          <!-- Badge: N misurazioni (non per i task di controllo, senza quote) -->
           <span class="badge badge-neutral text-xs shrink-0"
+                x-show="task.task_type !== 'check'"
                 x-text="(task.subtasks ? task.subtasks.length : 0) + ' {{ _('misurazioni') }}'">
           </span>
 
@@ -590,12 +675,14 @@
               <div class="grid grid-cols-1 sm:grid-cols-2 gap-3 p-3 rounded-lg bg-blue-50 dark:bg-blue-900/10 border border-blue-200 dark:border-blue-800">
                 <div>
                   <label class="tmf-label">{{ _('Tipo') }}</label>
-                  <select x-model="editTaskData.task_type" class="tmf-input text-sm">
+                  <select x-model="editTaskData.task_type" class="tmf-input text-sm"
+                          @change="onTaskTypeChange(editTaskData, $event.target.value)">
                     <option value="note">{{ _('Nota') }}</option>
                     <option value="measure">{{ _('Misura') }}</option>
                     <option value="drawing">{{ _('Disegno') }}</option>
                     <option value="camera_measure">{{ _('Misura camera') }}</option>
                     <option value="dxf_compare">{{ _('Confronto profilo') }}</option>
+                    <option value="check">{{ _('Controllo OK/NOK') }}</option>
                   </select>
                 </div>
                 <div>
@@ -618,12 +705,14 @@
                             rows="3"
                             placeholder="{{ _('Descrizione opzionale...') }}"></textarea>
                 </div>
+                {# #53 #55: ciclo di misura, compilazione a fine produzione e campi della nota #}
+                {{ task_flags_editor('editTaskData') }}
               </div>
             </template>
 
             <!-- ========== Subtask Table ========== -->
             <div class="subtask-table-wrap">
-              <table class="tmf-table" x-show="task.subtasks && task.subtasks.length > 0">
+              <table class="tmf-table" x-show="task.task_type !== 'check' && task.subtasks && task.subtasks.length > 0">
                 <thead>
                   <tr>
                     <th class="w-16 text-center">{{ _('#') }}</th>
@@ -710,7 +799,7 @@
             </div>
 
             <!-- ========== Empty Subtask State ========== -->
-            <div x-show="!task.subtasks || task.subtasks.length === 0"
+            <div x-show="task.task_type !== 'check' && (!task.subtasks || task.subtasks.length === 0)"
                  class="text-center py-8">
               <div class="inline-flex items-center justify-center w-14 h-14 rounded-full
                           bg-[var(--bg-secondary)] mb-3">
@@ -862,7 +951,7 @@
           <!-- ========== Task Card Footer ========== -->
           <div class="tmf-card-footer flex items-center justify-end">
             <button @click="openAddSubtask(task)"
-                    x-show="addingSubtaskForTask !== task.id"
+                    x-show="addingSubtaskForTask !== task.id && task.task_type !== 'check'"
                     class="btn btn-secondary text-xs gap-1.5">
               <svg class="w-3.5 h-3.5" fill="none" stroke="currentColor" stroke-width="2" viewBox="0 0 24 24">
                 <path stroke-linecap="round" stroke-linejoin="round" d="M12 4v16m8-8H4"/>
@@ -1210,8 +1299,10 @@ function taskEditor() {
 
     // ---- Task CRUD state ----
     editingTask: null,
-    editTaskData: { title: '', directive: '', description: '', task_type: 'note' },
-    newTask: { title: '', directive: '', description: '', task_type: 'note' },
+    editTaskData: { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] },
+    newTask: { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] },
+    MAX_NOTE_FIELDS: 20,  // #55: limite campi per nota (stesso del backend)
+    _fieldUid: 0,
 
     // ---- Subtask CRUD state ----
     editingSubtaskId: null,
@@ -1391,7 +1482,84 @@ function taskEditor() {
     // ============================================================
 
     resetNewTask() {
-      this.newTask = { title: '', directive: '', description: '', task_type: 'note' };
+      this.newTask = { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] };
+    },
+
+    // ---- #53 #55: ciclo di misura e campi della nota ----
+
+    // #53: default di "nel ciclo" per tipo (stessa regola del backend)
+    loopDefault(type) {
+      return ['measure', 'camera_measure', 'dxf_compare', 'check'].includes(type);
+    },
+
+    // Cambio tipo: il default del ciclo segue il tipo finche l'utente non lo tocca;
+    // at_close e campi esistono solo per la nota.
+    onTaskTypeChange(obj, type) {
+      obj.task_type = type;
+      if (!obj.in_loop_touched) obj.in_loop = this.loopDefault(type);
+      if (type !== 'note') { obj.at_close = false; obj.fields = []; }
+    },
+
+    // #55: key dall'etichetta (minuscole, [a-z0-9_], max 40), unica nella lista
+    _slugKey(label) {
+      const base = (label || '').normalize('NFD').replace(/[̀-ͯ]/g, '')
+        .toLowerCase().replace(/[^a-z0-9]+/g, '_').replace(/^_+|_+$/g, '').slice(0, 40);
+      return base || 'campo';
+    },
+
+    deriveNoteFieldKey(obj, f) {
+      if (f.locked) return;  // la key non cambia piu dopo il primo salvataggio
+      const used = new Set(obj.fields.filter(x => x !== f).map(x => x.key));
+      const base = this._slugKey(f.label);
+      let key = base, n = 2;
+      while (used.has(key)) {
+        const suffix = '_' + n++;
+        key = base.slice(0, 40 - suffix.length) + suffix;
+      }
+      f.key = key;
+    },
+
+    addNoteField(obj) {
+      if (obj.fields.length >= this.MAX_NOTE_FIELDS) return;
+      const f = { _uid: ++this._fieldUid, key: '', label: '', type: 'text', default: '', attachment: false, locked: false };
+      obj.fields.push(f);
+      this.deriveNoteFieldKey(obj, f);
+    },
+
+    removeNoteField(obj, i) {
+      obj.fields.splice(i, 1);
+    },
+
+    moveNoteField(obj, i, dir) {
+      const j = i + dir;
+      if (j < 0 || j >= obj.fields.length) return;
+      const [f] = obj.fields.splice(i, 1);
+      obj.fields.splice(j, 0, f);
+    },
+
+    // Campi salvati -> stato di modifica (key bloccate)
+    _fieldsToState(fields) {
+      return (fields || []).map(f => ({
+        _uid: ++this._fieldUid, key: f.key, label: f.label || '', type: f.type || 'text',
+        default: f.default == null ? '' : f.default, attachment: !!f.attachment, locked: true
+      }));
+    },
+
+    // Stato -> corpo API (fields null se non e una nota o e vuota); ok=false con errorMessage se manca un'etichetta.
+    _fieldsPayload(obj) {
+      if (obj.task_type !== 'note') return { ok: true, fields: null };
+      const out = [];
+      for (const f of obj.fields) {
+        if (!f.label.trim()) {
+          this.errorMessage = {{ _("Ogni campo deve avere un'etichetta")|tojson }};
+          return { ok: false };
+        }
+        out.push({
+          key: f.key, label: f.label.trim(), type: f.type || 'text',
+          default: (f.default || '').trim() || null, attachment: !!f.attachment
+        });
+      }
+      return { ok: true, fields: out.length ? out : null };
     },
 
     // ============================================================
@@ -1470,8 +1638,10 @@ function taskEditor() {
 
     async addTask() {
       if (!this.newTask.title.trim()) return;
-      this.saving = true;
       this.errorMessage = '';
+      const fp = this._fieldsPayload(this.newTask);  // #55
+      if (!fp.ok) return;
+      this.saving = true;
 
       try {
         const resp = await fetch(`/maker/api/recipes/${this.recipeId}/tasks`, {
@@ -1484,7 +1654,11 @@ function taskEditor() {
             title: this.newTask.title.trim(),
             task_type: this.newTask.task_type || 'note',
             directive: this.newTask.directive.trim() || null,
-            description: this.newTask.description.trim() || null
+            description: this.newTask.description.trim() || null,
+            // #53 #55: ciclo di misura, compilazione a fine produzione, campi della nota
+            in_loop: !!this.newTask.in_loop,
+            at_close: this.newTask.task_type === 'note' && !!this.newTask.at_close,
+            fields: fp.fields
           })
         });
 
@@ -1516,7 +1690,12 @@ function taskEditor() {
         title: task.title,
         task_type: task.task_type || 'note',
         directive: task.directive || '',
-        description: task.description || ''
+        description: task.description || '',
+        // #53 #55: valore salvato (task vecchi senza il campo: default del tipo)
+        in_loop: task.in_loop != null ? !!task.in_loop : this.loopDefault(task.task_type || 'note'),
+        in_loop_touched: false,
+        at_close: !!task.at_close,
+        fields: this._fieldsToState(task.fields)
       };
       // Ensure expanded
       this.expandedTask = task.id;
@@ -1524,13 +1703,15 @@ function taskEditor() {
 
     cancelEditTask() {
       this.editingTask = null;
-      this.editTaskData = { title: '', directive: '', description: '', task_type: 'note' };
+      this.editTaskData = { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] };
     },
 
     async updateTask(taskId) {
       if (!this.editTaskData.title.trim()) return;
-      this.saving = true;
       this.errorMessage = '';
+      const fp = this._fieldsPayload(this.editTaskData);  // #55
+      if (!fp.ok) return;
+      this.saving = true;
 
       try {
         const resp = await fetch(`/maker/api/tasks/${taskId}`, {
@@ -1543,7 +1724,11 @@ function taskEditor() {
             title: this.editTaskData.title.trim(),
             task_type: this.editTaskData.task_type || 'note',
             directive: this.editTaskData.directive.trim() || null,
-            description: this.editTaskData.description.trim() || null
+            description: this.editTaskData.description.trim() || null,
+            // #53 #55: ciclo di misura, compilazione a fine produzione, campi della nota
+            in_loop: !!this.editTaskData.in_loop,
+            at_close: this.editTaskData.task_type === 'note' && !!this.editTaskData.at_close,
+            fields: fp.fields
           })
         });
 

src/frontend/flask_app/tests/test_maker_task_fields.py

@@ -0,0 +1,79 @@
+"""Editor task del Maker: tipo check, in_loop, at_close, campi della nota (#53 #54 #55)."""
+import re
+
+from tests.conftest import api_get_router
+
+
+def _editor_html(logged_in_client, mock_api_client, tasks):
+    mock_api_client.get.side_effect = api_get_router({
+        "/api/recipes/3": {"id": 3, "code": "R", "name": "R"},
+        "/api/recipes/3/tasks": tasks,
+    })
+    resp = logged_in_client.get("/maker/recipes/3/tasks")
+    assert resp.status_code == 200
+    return resp.get_data(as_text=True)
+
+
+class TestTaskEditorFields:
+    def test_check_option_in_both_selects(self, logged_in_client, mock_api_client):
+        """#54: il tipo check e presente nella select di creazione e in quella di modifica."""
+        html = _editor_html(logged_in_client, mock_api_client, [{
+            "id": 11, "title": "T", "task_type": "note", "subtasks": [],
+            "file_path": None, "order_index": 1,
+        }])
+        assert html.count('<option value="check">') == 2
+        assert "Controllo OK/NOK" in html
+
+    def test_loop_and_note_controls_present(self, logged_in_client, mock_api_client):
+        """#53 #55: checkbox del ciclo, a fine produzione e editor dei campi in entrambi i form."""
+        html = _editor_html(logged_in_client, mock_api_client, [{
+            "id": 11, "title": "T", "task_type": "note", "subtasks": [],
+            "file_path": None, "order_index": 1, "in_loop": True,
+        }])
+        assert "Nel ciclo di misura (timer e cicalino)" in html
+        assert "Compila a fine produzione" in html
+        assert 'x-model="newTask.in_loop"' in html
+        assert 'x-model="editTaskData.in_loop"' in html
+        assert "addNoteField(newTask)" in html
+        assert "addNoteField(editTaskData)" in html
+        # segno sui task nel ciclo, e niente quote per check
+        assert 'x-show="task.in_loop"' in html
+        assert "task.task_type !== 'check'" in html
+
+    def test_save_bodies_carry_new_fields(self, logged_in_client, mock_api_client):
+        """Creazione e modifica inviano in_loop, at_close e fields nel corpo."""
+        html = _editor_html(logged_in_client, mock_api_client, [])
+        assert len(re.findall(r"in_loop: !!this\.(newTask|editTaskData)\.in_loop", html)) == 2
+        assert len(re.findall(r"fields: fp\.fields", html)) == 2
+        assert len(re.findall(r"at_close: this\.(newTask|editTaskData)\.task_type === 'note'", html)) == 2
+
+
+class TestTaskProxyForwardsFields:
+    """I proxy inoltrano il corpo cosi com'e: in_loop/at_close/fields arrivano al backend."""
+
+    BODY = {
+        "title": "Note finali", "task_type": "note", "in_loop": False, "at_close": True,
+        "fields": [{"key": "operatore", "label": "Operatore", "type": "text",
+                    "default": None, "attachment": False}],
+    }
+
+    def test_create_forwards(self, logged_in_client, mock_api_client):
+        mock_api_client.post.return_value = {"id": 5, **self.BODY}
+        resp = logged_in_client.post("/maker/api/recipes/3/tasks", json=self.BODY)
+        assert resp.status_code == 201
+        mock_api_client.post.assert_called_once_with("/api/recipes/3/tasks", data=self.BODY)
+
+    def test_update_forwards(self, logged_in_client, mock_api_client):
+        mock_api_client.put.return_value = {"id": 5, **self.BODY}
+        resp = logged_in_client.put("/maker/api/tasks/5", json=self.BODY)
+        assert resp.status_code == 200
+        mock_api_client.put.assert_called_once_with("/api/tasks/5", data=self.BODY)
+
+    def test_update_422_detail_returned(self, logged_in_client, mock_api_client):
+        """Un 422 del backend arriva al browser con il suo detail e lo status."""
+        mock_api_client.put.return_value = {
+            "error": True, "status_code": 422, "detail": "Massimo 20 campi",
+        }
+        resp = logged_in_client.put("/maker/api/tasks/5", json=self.BODY)
+        assert resp.status_code == 422
+        assert resp.get_json()["detail"] == "Massimo 20 campi"