PacketStation/server/packetstation/api/packet_children.py

"""Packet child rows: the BOM and its shorts, welds, heat tags, the heat table, the open stage (fields, checklist,
clock), the warehouse stage, the Q.C. inspection and change orders.

Every write goes through domain.pipeline.packet_write with its policy action: one transaction, the packet locked
first, the policy decides, then the version bump, packet_log, audit and NOTIFY. Row edits check the row's
row_version against If-Match (409 row_conflict); state changes check the state the caller expected (409). Every
mutating answer is the full read model.
"""

from __future__ import annotations

import uuid
from datetime import date
from decimal import Decimal
from typing import Any, Literal

import psycopg
from fastapi import APIRouter, Depends, Request
from psycopg.types.json import Jsonb
from pydantic import BaseModel, Field

from .. import policy
from ..domain import checklists, packets
from ..domain import qc as qc_dom
from ..domain.children import NEVER_VERIFY, active_depts, co_touched, current_wh, update_row, wh_move
from ..domain.pipeline import (
    SignBody,
    Write,
    apply_field_cas,
    check_row_version,
    jsonable,
    next_counter,
    packet_write,
    parse_if_match,
    qty_text,
    require_if_match,
)
from ..policy.actions import STATUS_LABELS
from ..text import LIMITS, RX_HEAT, RX_STAMP, clean_opt, clean_text
from .deps import Db, Principal, current_principal, get_db
from .errors import ApiError
from .packets import WRITE

router = APIRouter(prefix="/api/packets", tags=["packet-children"])

A = policy.Action

TEXT_SHORT = LIMITS["weld_text"]  # weld type, size, process, wo_base
QC_FIELD_MAX = 120

# The weld vocabulary (the old WELD_TYPES and PROCESSES).
WELD_TYPES = ("Fillet", "Groove", "Fill", "Seal", "Tack")
PROCESSES = ("GMAW", "FCAW", "SMAW", "GTAW", "SAW")
DEFAULT_SIZE = '5/16"'
POSITION = frozenset({"fx", "fy", "bfx", "bfy"})
# The weld fields in the order the change note lists them, with their note names.
WELD_NOTE_NAMES = {
    "type": "type",
    "size": "size",
    "process": "process",
    "items": "items",
    "wo_base": "wo",
    "piece": "piece",
    "stamp": "stamp",
    "page": "sheet",
}
# The heat table bounds (the old HT_MIN, HT_MAX and the drag clamp).
HT_XY_MAX = 0.97
HT_S_MIN, HT_S_MAX = 0.6, 2.1
STAGE_FIELDS = {
    "filler": LIMITS["filler"],
    "filler_lot": LIMITS["filler"],
    "preheat": LIMITS["preheat"],
    "notes": LIMITS["notes"],
}
STAGE_NOTE_NAMES = {"filler": "filler", "filler_lot": "filler lot", "preheat": "preheat", "notes": "notes"}

# The warehouse hold reasons, a closed set (the old build took any text). "Other" needs a note.
WH_HOLD_REASONS = (
    "Waiting on material (not in stock)",
    "Waiting on purchasing / PO",
    "Question for engineering or clerical",
    "Drawing / BOM problem",
    "Other",
)
# The short reasons, a closed set (the old build took any text).
SHORT_REASONS = (
    "Not in stock",
    "On order",
    "Back-ordered",
    "Damaged / rejected",
    "Wrong part received",
    "Needs cutting / prep",
    "Other",
)
WH_NOTE_MAX = 200
# The file types an MTR may be.
MTR_MIMES = frozenset({"application/pdf", "image/png", "image/jpeg"})
# The old warehouse board: which stage may follow which.
WH_NEXT: dict[str, frozenset[str]] = {
    "incoming": frozenset({"pulling", "hold"}),
    "pulling": frozenset({"pulled", "hold"}),
    "hold": frozenset({"pulling"}),
    "pulled": frozenset({"pulling"}),
}
WH_LABELS = {
    "pulling": "Started pulling",
    "hold": "Put on hold",
    "pulled": "Marked pulled, ready for production",
}
# Labels for a move into pulling from a stage other than incoming.
WH_PULLING_FROM = {"hold": "Pulling resumed", "pulled": "Back to pulling"}

MSG_NO_STAGE = "This packet has no open stage."
MSG_NO_INSPECTION = "This packet has no open Q.C. inspection."
MSG_SHORT_OPEN = "This line has an open short. Use the short to change heat, issued or location."
MSG_VOID = "This sheet is VOID. Mark the current sheet."


# ---------------------------------------------------------------- bodies


class It(BaseModel):
    it_reason: str | None = Field(default=None, max_length=1000)


class BomLineIn(BaseModel):
    item: int = Field(ge=1, le=100_000)
    qty: Decimal | None = Field(default=None, ge=0)
    part_id: str | None = Field(default=None, max_length=200)
    description: str | None = Field(default=None, max_length=1000)
    dwg_ref: str | None = Field(default=None, max_length=200)
    cl_item: str | None = Field(default=None, max_length=200)
    length: str | None = Field(default=None, max_length=200)
    material: str | None = Field(default=None, max_length=500)


class BomPutBody(It):
    bom: list[BomLineIn] = Field(max_length=1000)


class BomPatchBody(It):
    heat: str | None = Field(default=None, max_length=200)
    na: bool | None = None
    issued: Decimal | None = Field(default=None, ge=0)
    loc: str | None = Field(default=None, max_length=500)
    pulled: bool | None = None
    mtr_asset_id: uuid.UUID | None = None
    description: str | None = Field(default=None, max_length=1000)
    qty: Decimal | None = Field(default=None, ge=0)


class ShortBody(It):
    qty: int
    reason: str | None = Field(default=None, max_length=2000)
    ref: str | None = Field(default=None, max_length=500)
    eta: str | None = Field(default=None, max_length=40)
    note: str | None = Field(default=None, max_length=2000)
    heat: str | None = Field(default=None, max_length=200)
    expect_open: bool


class ShortCloseBody(It):
    expect_open: bool = True


class ReceiptBody(It):
    qty: int
    heat: str | None = Field(default=None, max_length=200)
    note: str | None = Field(default=None, max_length=2000)


class ShortApproveBody(It):
    sign: SignBody | None = None
    note: str | None = Field(default=None, max_length=2000)


class WeldBody(It):
    page: int | None = Field(default=None, ge=1, le=100_000)
    fx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    fy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    wo_base: str | None = Field(default=None, max_length=200)
    piece: int | None = Field(default=None, ge=1, le=100_000)
    items: list[int] = Field(min_length=1, max_length=200)
    type: str | None = Field(default=None, max_length=200)
    size: str | None = Field(default=None, max_length=200)
    process: str | None = Field(default=None, max_length=200)
    stamp: str | None = Field(default=None, max_length=40)


class WeldPatchBody(It):
    page: int | None = Field(default=None, ge=1, le=100_000)
    fx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    fy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    wo_base: str | None = Field(default=None, max_length=200)
    piece: int | None = Field(default=None, ge=1, le=100_000)
    items: list[int] | None = Field(default=None, min_length=1, max_length=200)
    type: str | None = Field(default=None, max_length=200)
    size: str | None = Field(default=None, max_length=200)
    process: str | None = Field(default=None, max_length=200)
    stamp: str | None = Field(default=None, max_length=40)


class RemoveBody(It):
    reason: str | None = Field(default=None, max_length=2000)


class TagBody(It):
    page: int | None = Field(default=None, ge=1, le=100_000)
    item: int = Field(ge=1, le=100_000)
    fx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    fy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)


class TagPatchBody(It):
    page: int | None = Field(default=None, ge=1, le=100_000)
    item: int | None = Field(default=None, ge=1, le=100_000)
    fx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    fy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfx: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)
    bfy: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False)


class HeatTableBody(It):
    x: float = Field(allow_inf_nan=False)
    y: float = Field(allow_inf_nan=False)
    s: float = Field(allow_inf_nan=False)
    carried_from: int | None = Field(default=None, ge=1, le=100_000)


class StageBody(It):
    """The flat body (with If-Match on the stage row), or `{set, base}` for field-level compare-and-set."""

    filler: str | None = Field(default=None, max_length=500)
    filler_lot: str | None = Field(default=None, max_length=500)
    preheat: str | None = Field(default=None, max_length=500)
    notes: str | None = Field(default=None, max_length=5000)
    set: dict[str, str | None] | None = Field(default=None, max_length=10)
    base: dict[str, str | None] | None = Field(default=None, max_length=10)


class ChecklistBody(It):
    done: bool


class ClockBody(It):
    op: Literal["start", "pause", "resume"]


class WhBody(It):
    stage: Literal["incoming", "pulling", "hold", "pulled"]
    expect_stage: Literal["incoming", "pulling", "hold", "pulled"]
    reason: str | None = Field(default=None, max_length=2000)
    note: str | None = Field(default=None, max_length=2000)


class WeldExamIn(BaseModel):
    result: Literal["accept", "reject", ""] | None = None
    note: str | None = Field(default=None, max_length=2000)


class QcBody(It):
    """`{set, base}` for field-level compare-and-set of the scalar fields, or the phase 2 flat body with If-Match on
    the inspection's row_version (checks and weld_exam merge per key)."""

    checks: dict[str, bool] | None = Field(default=None, max_length=50)
    weld_exam: dict[str, WeldExamIn] | None = Field(default=None, max_length=2000)
    nde_method: str | None = Field(default=None, max_length=500)
    nde_result: str | None = Field(default=None, max_length=500)
    nde_report: str | None = Field(default=None, max_length=500)
    hardness: str | None = Field(default=None, max_length=500)
    inspector: str | None = Field(default=None, max_length=500)
    stamp: str | None = Field(default=None, max_length=40)
    notes: str | None = Field(default=None, max_length=5000)
    set: dict[str, str | None] | None = Field(default=None, max_length=10)
    base: dict[str, str | None] | None = Field(default=None, max_length=10)


class QcCheckBody(It):
    done: bool


class QcWeldBody(It):
    result: Literal["accept", "reject", ""] | None = None
    note: str | None = Field(default=None, max_length=2000)


class CoItemIn(BaseModel):
    old_page: int | None = Field(default=None, ge=1, le=100_000)
    new_page: int | None = Field(default=None, ge=1, le=100_000)
    added: bool = False


class CoIssueBody(It):
    reason: str | None = Field(default=None, max_length=2000)
    ecn: str | None = Field(default=None, max_length=200)
    required_depts: list[str] | None = Field(default=None, max_length=30)
    items: list[CoItemIn] = Field(default_factory=list, max_length=200)
    hold: bool = False
    sign: SignBody | None = None


class SignOnly(It):
    sign: SignBody | None = None


class CoRejectBody(It):
    reason: str | None = Field(default=None, max_length=2000)
    sign: SignBody | None = None


class CoHoldBody(It):
    hold: bool


class CoVerifyBody(It):
    dept: str = Field(min_length=1, max_length=10)
    note: str | None = Field(default=None, max_length=2000)
    sign: SignBody | None = None


# ---------------------------------------------------------------- helpers


def _pp(d: Db, p: Principal) -> policy.Principal:
    return packets.policy_principal_for(d.conn, p)


def _read(d: Db, pp: policy.Principal, packet_id: int) -> dict[str, Any]:
    return packets.read_packet(d.conn, packet_id, pp)


def _heat(value: Any) -> str | None:
    """A cleaned heat number, None when empty, 422 invalid_heat when it does not fit RX_HEAT.

    Trailing white space is trimmed (a scanner may add it). A leading space is refused, as the pattern says.
    """
    text = clean_text(value, 200)
    if not text:
        return None
    lead = isinstance(value, str) and value[:1].isspace()
    if lead or len(text) > LIMITS["heat"] or not RX_HEAT.match(text):
        raise ApiError(
            422,
            "invalid_heat",
            f'Heat number "{text}" isn\'t valid: letters, numbers, space . / # _ - only, up to 40 characters.',
        )
    return text


def _stamp(value: Any) -> str | None:
    text = clean_text(value, 40)
    if not text:
        return None
    if not RX_STAMP.match(text):
        raise ApiError(422, "invalid_stamp", "A stamp is up to 10 letters, digits or dashes.")
    return text


def _user_stamp(conn: Any, user_id: Any) -> str | None:
    r = conn.execute("SELECT welder_stamp FROM users WHERE id = %s", (user_id,)).fetchone()
    return r[0] if r and r[0] else None


def _bom_line(conn: Any, packet_id: int, item: int) -> dict[str, Any]:
    row = packets.one(conn, "SELECT * FROM bom_lines WHERE packet_id = %s AND item = %s", (packet_id, item))
    if row is None:
        raise ApiError(404, "not_found", f"There is no BOM item {item} on this packet.")
    return row


def _open_short(conn: Any, line_id: int) -> dict[str, Any] | None:
    return packets.one(conn, "SELECT * FROM bom_shorts WHERE bom_line_id = %s AND closed_at IS NULL", (line_id,))


def _bom_items(conn: Any, packet_id: int) -> set[int]:
    return {int(r[0]) for r in conn.execute("SELECT item FROM bom_lines WHERE packet_id = %s", (packet_id,))}


def _check_items(conn: Any, packet_id: int, items: list[int]) -> list[int]:
    have = _bom_items(conn, packet_id)
    bad = sorted({i for i in items if i not in have})
    if bad:
        raise ApiError(
            422, "invalid_items", f"Item {', '.join(map(str, bad))} is not on this packet's BOM.", extra={"items": bad}
        )
    return sorted(set(items))


def _open_stage(conn: Any, packet_id: int) -> dict[str, Any] | None:
    return packets.one(conn, "SELECT * FROM packet_stages WHERE packet_id = %s AND closed_at IS NULL", (packet_id,))


def _need_stage(conn: Any, packet_id: int) -> dict[str, Any]:
    st = _open_stage(conn, packet_id)
    if st is None:
        raise ApiError(409, "no_stage", MSG_NO_STAGE)
    return st


def _num(value: Any) -> Decimal | None:
    return None if value is None else Decimal(str(value))


def _status_tail(conn: Any, row: dict[str, Any]) -> str:
    """The end of the receive note: " · packet ready to assign", plus " in {station}" when at a station."""
    out = f" · packet {STATUS_LABELS.get(str(row['status']), str(row['status'])).lower()}"
    if row["status"] in policy.ACTIVE and row["station_id"] is not None:
        st = conn.execute("SELECT name FROM stations WHERE id = %s", (row["station_id"],)).fetchone()
        if st and st[0]:
            out += f" in {st[0]}"
    return out


def _diff(row: dict[str, Any], values: dict[str, Any]) -> dict[str, Any]:
    """The values that differ from the row."""
    out: dict[str, Any] = {}
    for k, v in values.items():
        cur = row.get(k)
        if isinstance(v, Decimal) or isinstance(cur, Decimal):
            if _num(v) != _num(cur):
                out[k] = v
        elif (str(v) if isinstance(v, uuid.UUID) else v) != (str(cur) if isinstance(cur, uuid.UUID) else cur):
            out[k] = v
    return out


def _auto_pull(w: Write) -> None:
    """Ticking a line Pulled while the warehouse is waiting or on hold starts pulling, as the old build did."""
    if w.packet["status"] != "warehouse":
        return
    cur = current_wh(w.conn, w.id)
    stage = cur["stage"] if cur else "incoming"
    if stage not in ("incoming", "hold"):
        return
    note = "pulling resumed" if stage == "hold" else "started pulling"
    wh_move(w.conn, w.id, "pulling", w.user_id, note=note, auto=True)
    w.changed("wh")
    w.note = f"{w.note} · {note}" if w.note else note


def _set_pulled(w: Write, line: dict[str, Any], values: dict[str, Any]) -> None:
    """values gets pulled true, pulled_at and pulled_by (kept when already pulled), issued = qty when unset."""
    values["pulled"] = True
    if not line["pulled"]:
        values["pulled_at"] = "now"
        values["pulled_by"] = w.user_id


def _apply_line(w: Write, line: dict[str, Any], values: dict[str, Any]) -> dict[str, Any]:
    """Write the changed values to a BOM line (row_version + 1). pulled_at 'now' is the database clock."""
    if values.get("pulled_at") == "now":
        values["pulled_at"] = w.conn.execute("SELECT now()").fetchone()[0]
    row = update_row(w.conn, "bom_lines", line["id"], values)
    assert row is not None
    return row


def _weld(conn: Any, packet_id: int, no: int) -> dict[str, Any]:
    row = packets.one(
        conn, "SELECT * FROM welds WHERE packet_id = %s AND weld_no = %s AND removed_at IS NULL", (packet_id, no)
    )
    if row is None:
        raise ApiError(404, "not_found", f"There is no weld W{no} on this packet.")
    return row


def _tag(conn: Any, packet_id: int, no: int) -> dict[str, Any]:
    row = packets.one(
        conn, "SELECT * FROM heat_tags WHERE packet_id = %s AND tag_no = %s AND removed_at IS NULL", (packet_id, no)
    )
    if row is None:
        raise ApiError(404, "not_found", f"There is no heat tag H{no} on this packet.")
    return row


def _page_row(conn: Any, packet_id: int, page: int) -> dict[str, Any] | None:
    return packets.one(
        conn, "SELECT n, type, dwg, rev, void_at FROM packet_pages WHERE packet_id = %s AND n = %s", (packet_id, page)
    )


def _drawing_page(conn: Any, packet_id: int, page: int) -> dict[str, Any]:
    """A drawing page of the packet, live or VOID. 422 invalid_page otherwise."""
    pg = _page_row(conn, packet_id, page)
    if pg is None or pg["type"] != "drawing":
        raise ApiError(422, "invalid_page", f"There is no drawing sheet {page} on this packet.")
    return pg


def _live_page(conn: Any, packet_id: int, page: int | None) -> None:
    """A mark goes on a live drawing page. 422 void_page on a VOID sheet; nothing is checked without a page."""
    if page is None:
        return
    if _drawing_page(conn, packet_id, page)["void_at"] is not None:
        raise ApiError(422, "void_page", MSG_VOID)


def _not_void(conn: Any, packet_id: int, page: int | None) -> None:
    """An existing mark on a VOID sheet is never changed."""
    if page is None:
        return
    pg = _page_row(conn, packet_id, page)
    if pg is not None and pg["void_at"] is not None:
        raise ApiError(422, "void_page", MSG_VOID)


def _page_text(pg: dict[str, Any] | None, page: int) -> str:
    """A sheet in a log note: its drawing number, else "sheet 2"."""
    dwg = (pg or {}).get("dwg")
    return str(dwg) if dwg else f"sheet {page}"


def _num_text(value: Any) -> str:
    return f"{round(float(value), 2):g}"


def _co_no_text(no: int) -> str:
    return packets.co_label(no)


def _co(conn: Any, packet_id: int, no: int) -> dict[str, Any]:
    row = packets.one(
        conn,
        "SELECT *, origin::text AS origin, review::text AS review FROM change_orders"
        " WHERE packet_id = %s AND co_no = %s",
        (packet_id, no),
    )
    if row is None:
        raise ApiError(404, "not_found", f"There is no change order {_co_no_text(no)} on this packet.")
    return row


def _co_state(row: dict[str, Any], want: str, what: str) -> None:
    if row["review"] != want:
        raise ApiError(
            409,
            "co_state",
            f"Change order {_co_no_text(row['co_no'])} is {row['review']}. Only a {want} change order can be {what}.",
            extra={"current": {"co_no": row["co_no"], "review": row["review"], "hold": row["hold"]}},
        )


# ---------------------------------------------------------------- BOM


@router.put("/{packet_id}/bom", dependencies=WRITE)
def put_bom(
    packet_id: int,
    body: BomPutBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Full replace of a draft's BOM (Clerical). If-Match is the packet version."""
    if_match = require_if_match(request)
    seen: set[int] = set()
    lines: list[dict[str, Any]] = []
    for i, bl in enumerate(body.bom):
        if bl.item in seen:
            raise ApiError(422, "duplicate_item", f"BOM item {bl.item} is listed twice.")
        seen.add(bl.item)
        lines.append(
            {
                "item": bl.item,
                "qty": bl.qty,
                "part_id": clean_opt(bl.part_id, LIMITS["part_id"]),
                "description": clean_opt(bl.description, LIMITS["description"]),
                "dwg_ref": clean_opt(bl.dwg_ref, LIMITS["drawing"]),
                "cl_item": clean_opt(bl.cl_item, LIMITS["part_id"]),
                "length": clean_opt(bl.length, LIMITS["length"]),
                "material": clean_opt(bl.material, LIMITS["material"]),
                "sort": i,
            }
        )
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_BOM, request=request, it_reason=body.it_reason, if_match=if_match, policy_principal=pp
    ) as w:
        if w.packet["status"] != "draft":
            raise ApiError(403, "wrong_status", "The whole BOM is replaced only while the packet is a draft.")
        conn.execute("DELETE FROM bom_lines WHERE packet_id = %s", (packet_id,))
        for b in lines:
            conn.execute(
                "INSERT INTO bom_lines (packet_id, item, qty, part_id, description, dwg_ref, cl_item, length, material,"
                " src, sort) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'hand', %s)",
                (
                    packet_id,
                    b["item"],
                    b["qty"],
                    b["part_id"],
                    b["description"],
                    b["dwg_ref"],
                    b["cl_item"],
                    b["length"],
                    b["material"],
                    b["sort"],
                ),
            )
        w.changed("bom")
        w.log("BOM replaced", f"{len(lines)} line{'s' if len(lines) != 1 else ''}")
    return _read(d, pp, packet_id)


@router.patch("/{packet_id}/bom/{item}", dependencies=WRITE)
def patch_bom_line(
    packet_id: int,
    item: int,
    body: BomPatchBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """One BOM line: heat, na, issued, loc, pulled, mtr_asset_id, description, qty. If-Match is the line's
    row_version."""
    if_match = parse_if_match(request)
    given = body.model_fields_set - {"it_reason"}
    values: dict[str, Any] = {}
    if "heat" in given:
        values["heat"] = _heat(body.heat)
    if "na" in given:
        if body.na is None:
            raise ApiError(422, "invalid_request", "N/A is true or false.")
        values["na"] = body.na
        if body.na:
            values["heat"] = None
    if "issued" in given:
        values["issued"] = body.issued
    if "loc" in given:
        values["loc"] = clean_opt(body.loc, LIMITS["location"])
    if "description" in given:
        values["description"] = clean_opt(body.description, LIMITS["description"])
    if "qty" in given:
        values["qty"] = body.qty
    if "mtr_asset_id" in given:
        values["mtr_asset_id"] = body.mtr_asset_id
    if "pulled" in given and body.pulled is None:
        raise ApiError(422, "invalid_request", "Pulled is true or false.")
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_BOM, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        line = _bom_line(conn, packet_id, item)
        check_row_version(line, if_match)
        if _open_short(conn, line["id"]) is not None and given & {"heat", "issued", "loc"}:
            raise ApiError(409, "short_open", MSG_SHORT_OPEN)
        if values.get("mtr_asset_id") is not None:
            asset = conn.execute("SELECT mime FROM assets WHERE id = %s", (values["mtr_asset_id"],)).fetchone()
            if asset is None:
                raise ApiError(422, "invalid_asset", "That MTR file is not on the server. Upload it first.")
            if asset[0] not in MTR_MIMES:
                raise ApiError(422, "invalid_asset", "An MTR is a PDF, PNG or JPG file.")
        newly_pulled = False
        if "pulled" in given:
            if body.pulled:
                if not line["pulled"]:
                    newly_pulled = True
                    _set_pulled(w, line, values)
                    issued = values.get("issued", line["issued"])
                    if issued is None:
                        values["issued"] = line["qty"]
            else:
                values["pulled"] = False
                values["pulled_at"] = None
                values["pulled_by"] = None
        changes = _diff(line, values)
        if not changes:
            w.skip()
            return _read(d, pp, packet_id)
        _apply_line(w, line, changes)
        w.changed(*[f"bom.{k}" for k in changes if k not in ("pulled_at", "pulled_by")])
        w.detail["item"] = item
        _log_line(w, item, line, changes, given)
        if newly_pulled:
            _auto_pull(w)
    return _read(d, pp, packet_id)


def _log_line(w: Write, item: int, line: dict[str, Any], changes: dict[str, Any], given: set[str]) -> None:
    """The packet_log text of a line edit. Fields the server set along the way (issued on a Pulled tick, the heat
    cleared by N/A) do not count as a second field."""
    kinds = {k for k in changes if k not in ("pulled_at", "pulled_by")}
    if "pulled" in kinds and "issued" not in given:
        kinds.discard("issued")
    if "na" in kinds and "heat" not in given:
        kinds.discard("heat")
    if len(kinds) != 1:
        w.log(f"Line changed · item {item}")
        return
    kind = next(iter(kinds))
    if kind == "pulled":
        w.log(f"Pulled · item {item}" if changes["pulled"] else f"Pulled undone · item {item}")
    elif kind == "heat":
        if changes["heat"]:
            w.log(f"Heat recorded · item {item}", changes["heat"])
        else:
            w.log(f"Heat cleared · item {item}")
    elif kind == "na":
        w.log(f"Marked N/A · item {item}" if changes["na"] else f"Traceable again · item {item}")
    elif kind == "issued":
        w.log(f"Issued qty changed · item {item}", f"{qty_text(changes['issued'])} of {qty_text(line['qty'])}")
    elif kind == "loc":
        w.log(f"Location set · item {item}", changes["loc"] or None)
    elif kind == "mtr_asset_id":
        w.log(f"MTR attached · item {item}" if changes["mtr_asset_id"] else f"MTR removed · item {item}")
    else:
        w.log(f"Line changed · item {item}")


@router.post("/{packet_id}/bom/{item}/short", dependencies=WRITE)
def save_short(
    packet_id: int,
    item: int,
    body: ShortBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Open a short on a line, or update the open one. expect_open says which the caller saw."""
    reason = clean_opt(body.reason, LIMITS["reason"])
    if reason not in SHORT_REASONS:
        raise ApiError(400, "invalid_reason", "Pick one of the short reasons.")
    ref = clean_opt(body.ref, LIMITS["po"])
    note = clean_opt(body.note, LIMITS["note"])
    eta = clean_opt(body.eta, 40)
    if eta is not None:
        try:
            date.fromisoformat(eta)
        except ValueError:
            raise ApiError(422, "invalid_date", "The ETA is a date (YYYY-MM-DD).") from None
    heat = _heat(body.heat) if body.heat is not None else None
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_BOM, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        line = _bom_line(conn, packet_id, item)
        short = _open_short(conn, line["id"])
        if (short is not None) != body.expect_open:
            raise ApiError(
                409,
                "short_conflict",
                "Someone else changed this short. Reload and try again.",
                extra={"current": jsonable({"open": short is not None, "short": short})},
            )
        need = _num(line["qty"])
        if need is None or need < 1:
            raise ApiError(422, "invalid_qty", "This line has no quantity, so it cannot be short.")
        if not (1 <= body.qty <= need):
            raise ApiError(422, "invalid_qty", f"Short qty must be 1 to {need}.")
        if short is None:
            conn.execute(
                "INSERT INTO bom_shorts (packet_id, bom_line_id, qty, qty_open, reason, ref, eta, note, opened_by)"
                " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
                (packet_id, line["id"], body.qty, body.qty, reason, ref, eta, note, w.user_id),
            )
            label = f"Short recorded · item {item}"
        else:
            received = (_num(short["qty"]) or Decimal(0)) - (_num(short["qty_open"]) or Decimal(0))
            update_row(
                conn,
                "bom_shorts",
                short["id"],
                {
                    "qty": received + body.qty,
                    "qty_open": body.qty,
                    "reason": reason,
                    "ref": ref,
                    "eta": eta,
                    "note": note,
                },
            )
            label = f"Short updated · item {item}"
        values: dict[str, Any] = {"issued": need - body.qty}
        _set_pulled(w, line, values)
        if heat is not None and not line["na"]:
            values["heat"] = heat
        newly_pulled = not line["pulled"]
        changes = _diff(line, values)
        if changes:
            _apply_line(w, line, changes)
        w.changed("bom.short")
        w.detail.update({"item": item, "qty": body.qty})
        parts = [f"{body.qty} of {qty_text(need)} short"]
        parts += [x for x in (reason, ref, f"ETA {eta}" if eta else None) if x]
        w.log(label, " · ".join(parts))
        if newly_pulled:
            _auto_pull(w)
    return _read(d, pp, packet_id)


@router.delete("/{packet_id}/bom/{item}/short", dependencies=WRITE)
def clear_short(
    packet_id: int,
    item: int,
    request: Request,
    body: ShortCloseBody | None = None,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Close the open short: the whole quantity counts as issued."""
    body = body or ShortCloseBody()
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_BOM, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        line = _bom_line(conn, packet_id, item)
        short = _open_short(conn, line["id"])
        if short is None or not body.expect_open:
            raise ApiError(
                409,
                "short_conflict",
                "Someone else changed this short. Reload and try again.",
                extra={"current": jsonable({"open": short is not None, "short": short})},
            )
        conn.execute(
            "UPDATE bom_shorts SET closed_at = now(), row_version = row_version + 1 WHERE id = %s", (short["id"],)
        )
        changes = _diff(line, {"issued": line["qty"]})
        if changes:
            _apply_line(w, line, changes)
        w.changed("bom.short")
        w.detail.update({"item": item, "qty": float(short["qty_open"])})
        w.log(f"Short cleared · item {item}")
    return _read(d, pp, packet_id)


@router.post("/{packet_id}/bom/{item}/short/receipts", dependencies=WRITE)
def receive_short(
    packet_id: int,
    item: int,
    body: ReceiptBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Receive short parts. The short closes when nothing is open.

    Receive by custody: a warehouse member receives in any status but filed (WH_RECEIVE_SHORT); anyone else
    receives as the holder of the packet (EDIT_BOM), and the policy refuses everyone who is neither.
    """
    heat = _heat(body.heat)
    note = clean_opt(body.note, LIMITS["note"])
    pp = _pp(d, p)
    conn = d.conn
    action = A.WH_RECEIVE_SHORT if "WH" in pp.support else A.EDIT_BOM
    with packet_write(d, p, packet_id, action, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        line = _bom_line(conn, packet_id, item)
        short = _open_short(conn, line["id"])
        if short is None:
            raise ApiError(404, "not_found", f"Item {item} has no open short.")
        left = _num(short["qty_open"]) or Decimal(0)
        if not (1 <= body.qty <= left):
            raise ApiError(422, "invalid_qty", f"Received qty must be 1 to {left}.")
        if heat is None and not line["na"]:
            raise ApiError(422, "heat_required", "Scan or type the heat number of the part received.")
        conn.execute(
            "INSERT INTO short_receipts (short_id, packet_id, qty, heat, note, by) VALUES (%s, %s, %s, %s, %s, %s)",
            (short["id"], packet_id, body.qty, heat, note, w.user_id),
        )
        still = left - body.qty
        conn.execute(
            "UPDATE bom_shorts SET qty_open = %s, closed_at = CASE WHEN %s <= 0 THEN now() END,"
            " row_version = row_version + 1 WHERE id = %s",
            (still, still, short["id"]),
        )
        values: dict[str, Any] = {"issued": (_num(line["issued"]) or Decimal(0)) + body.qty}
        if heat:
            heats = [h for h in (line["heat"] or "").split(" / ") if h]
            if heat not in heats:
                heats.append(heat)
            values["heat"] = " / ".join(heats)
        _apply_line(w, line, values)
        w.changed("bom.short")
        w.detail.update({"item": item, "qty": body.qty})
        tail = f"{qty_text(still)} still short" if still > 0 else "short closed"
        w.log(
            f"Short received · item {item}",
            " · ".join(x for x in (f"{body.qty} pc", heat and f"heat {heat}", tail) if x)
            + _status_tail(conn, w.packet),
        )
    return _read(d, pp, packet_id)


def _not_approved(conn: Any, receipt_id: int) -> None:
    if conn.execute("SELECT 1 FROM short_approvals WHERE receipt_id = %s", (receipt_id,)).fetchone():
        raise ApiError(409, "already_approved", "This receipt is already approved.")


@router.post("/{packet_id}/shorts/receipts/{receipt_id}/approve", dependencies=WRITE)
def approve_short_receipt(
    packet_id: int,
    receipt_id: int,
    body: ShortApproveBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """The department supervisor approves one received short (C7): the part and its heat number match what the
    warehouse issued. One approval per receipt; a second answers 409 already_approved."""
    note = clean_opt(body.note, LIMITS["note"])
    pp = _pp(d, p)
    conn = d.conn
    receipt = packets.one(
        conn,
        "SELECT r.*, b.item FROM short_receipts r JOIN bom_shorts s ON s.id = r.short_id"
        " JOIN bom_lines b ON b.id = s.bom_line_id WHERE r.id = %s AND r.packet_id = %s",
        (receipt_id, packet_id),
    )
    if receipt is None:
        raise ApiError(404, "not_found", "No such short receipt on this packet.")
    _not_approved(conn, receipt_id)  # before the signature: a second approval costs no attempt
    content = {
        "action": str(A.SHORT_APPROVE),
        "packet_id": packet_id,
        "receipt_id": receipt_id,
        "item": receipt["item"],
        "qty": qty_text(receipt["qty"]),
        "heat": receipt["heat"],
        "note": note,
    }
    with packet_write(
        d,
        p,
        packet_id,
        A.SHORT_APPROVE,
        request=request,
        it_reason=body.it_reason,
        sign=body.sign,
        sign_content=content,
        policy_principal=pp,
    ) as w:
        _not_approved(conn, receipt_id)  # again under the packet lock: a racing approval waits, then sees the first
        conn.execute(
            "INSERT INTO short_approvals (receipt_id, short_id, packet_id, dept_code, note, signature_id, by)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s)",
            (
                receipt_id,
                receipt["short_id"],
                packet_id,
                w.state.dept,
                note,
                w.signature.id if w.signature else None,
                w.user_id,
            ),
        )
        w.changed("bom.short")
        w.detail.update({"item": receipt["item"], "receipt_id": receipt_id})
        parts = [f"{qty_text(receipt['qty'])} pc", receipt["heat"] and f"heat {receipt['heat']}", note]
        w.log(f"Short approved · item {receipt['item']}", " · ".join(x for x in parts if x))
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- welds


def _weld_values(conn: Any, packet_id: int, body: WeldBody | WeldPatchBody, given: set[str]) -> dict[str, Any]:
    values: dict[str, Any] = {}
    for k in ("page", "fx", "fy", "bfx", "bfy", "piece"):
        if k in given:
            values[k] = getattr(body, k)
    for k in ("wo_base", "type", "size", "process"):
        if k in given:
            values[k] = clean_opt(getattr(body, k), TEXT_SHORT)
    if values.get("type") is not None and values["type"] not in WELD_TYPES:
        raise ApiError(422, "invalid_weld", "Pick a weld type from the list.")
    if values.get("process") is not None and values["process"] not in PROCESSES:
        raise ApiError(422, "invalid_weld", "Pick a process from the list.")
    if values.get("wo_base") is not None:
        bases = {
            r[0] for r in conn.execute("SELECT wo_base FROM packet_work_orders WHERE packet_id = %s", (packet_id,))
        }
        if values["wo_base"] not in bases:
            raise ApiError(422, "invalid_wo", "Pick one of this packet's work orders.")
    if "stamp" in given:
        values["stamp"] = _stamp(body.stamp)
    if "items" in given:
        if body.items is None:
            raise ApiError(422, "invalid_items", "Pick the BOM items this weld joins.")
        values["items"] = _check_items(conn, packet_id, body.items)
    return values


def _items_text(items: Any) -> str:
    return "items " + " + ".join(map(str, items or []))


def _weld_note(values: dict[str, Any]) -> str:
    """The note of a new weld: Fillet 5/16" · items 1 + 2 · L662001 #1."""
    desc = " ".join(str(x) for x in (values.get("type"), values.get("size")) if x)
    wo = values.get("wo_base")
    if wo and values.get("piece") is not None:
        wo = f"{wo} #{values['piece']}"
    return " · ".join(x for x in (desc, _items_text(values["items"]), wo) if x)


def _change_note(changes: dict[str, Any]) -> str | None:
    """The changed weld fields: type Groove · items 1 + 3."""
    parts: list[str] = []
    for k, name in WELD_NOTE_NAMES.items():
        if k not in changes:
            continue
        v = changes[k]
        if k == "items":
            parts.append(_items_text(v))
        else:
            parts.append(f"{name} {v}" if v not in (None, "") else f"{name} cleared")
    return " · ".join(parts) or None


@router.post("/{packet_id}/welds", dependencies=WRITE)
def add_weld(
    packet_id: int,
    body: WeldBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """A new weld. The weld number comes from the packet counter."""
    pp = _pp(d, p)
    conn = d.conn
    given = set(body.model_fields_set) | {"items"}
    with packet_write(
        d, p, packet_id, A.EDIT_WELDS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        values = _weld_values(conn, packet_id, body, given)
        _live_page(conn, packet_id, values.get("page"))
        if values.get("size") is None:
            values["size"] = DEFAULT_SIZE
        dept = str(w.packet["dept_code"])
        worker = pp.level(dept) == 1
        welder_id = w.user_id if worker else w.packet["welder_id"]
        if values.get("stamp") is None:
            values["stamp"] = (_user_stamp(conn, w.user_id) if worker else None) or w.packet["welder_stamp"]
        stage = _open_stage(conn, packet_id)
        no = next_counter(conn, packet_id, "weld")
        conn.execute(
            "INSERT INTO welds (packet_id, weld_no, page, fx, fy, bfx, bfy, wo_base, piece, items, type, size, process,"
            " stamp, welder_id, dept_code, stage_id, by) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,"
            " %s, %s, %s, %s, %s)",
            (
                packet_id,
                no,
                values.get("page"),
                values.get("fx"),
                values.get("fy"),
                values.get("bfx"),
                values.get("bfy"),
                values.get("wo_base"),
                values.get("piece"),
                values["items"],
                values.get("type"),
                values.get("size"),
                values.get("process"),
                values.get("stamp"),
                welder_id,
                dept,
                stage["id"] if stage else None,
                w.user_id,
            ),
        )
        w.changed("welds")
        w.detail["weld_no"] = no
        w.log(f"Weld W{no} marked", _weld_note(values))
    return _read(d, pp, packet_id)


@router.patch("/{packet_id}/welds/{no}", dependencies=WRITE)
def patch_weld(
    packet_id: int,
    no: int,
    body: WeldPatchBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    if_match = parse_if_match(request)
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_WELDS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        row = _weld(conn, packet_id, no)
        check_row_version(row, if_match)
        _not_void(conn, packet_id, row["page"])
        values = _weld_values(conn, packet_id, body, set(body.model_fields_set))
        changes = _diff(row, values)
        if "items" in changes and sorted(row["items"] or []) == changes["items"]:
            del changes["items"]
        if not changes:
            w.skip()
            return _read(d, pp, packet_id)
        if "page" in changes:
            _live_page(conn, packet_id, changes["page"])
        update_row(conn, "welds", row["id"], changes)
        w.changed(*[f"weld.{k}" for k in changes])
        w.detail["weld_no"] = no
        if set(changes) <= POSITION:
            w.log(f"Weld W{no} moved")
        else:
            w.log(f"Weld W{no} changed", _change_note(changes))
    return _read(d, pp, packet_id)


@router.delete("/{packet_id}/welds/{no}", dependencies=WRITE)
def remove_weld(
    packet_id: int,
    no: int,
    request: Request,
    body: RemoveBody | None = None,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Soft delete. If-Match is checked when given."""
    body = body or RemoveBody()
    if_match = parse_if_match(request)
    reason = clean_opt(body.reason, LIMITS["reason"])
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_WELDS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        row = _weld(conn, packet_id, no)
        if if_match is not None:
            check_row_version(row, if_match)
        _not_void(conn, packet_id, row["page"])
        conn.execute(
            "UPDATE welds SET removed_at = now(), removed_by = %s, removed_reason = %s, row_version = row_version + 1"
            " WHERE id = %s",
            (w.user_id, reason, row["id"]),
        )
        w.changed("welds")
        w.detail["weld_no"] = no
        w.log(f"Weld W{no} removed", reason)
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- heat tags and the heat table


def _tag_values(conn: Any, packet_id: int, body: TagBody | TagPatchBody, given: set[str]) -> dict[str, Any]:
    values: dict[str, Any] = {k: getattr(body, k) for k in ("page", "fx", "fy", "bfx", "bfy") if k in given}
    if "item" in given:
        if body.item is None:
            raise ApiError(422, "invalid_items", "Pick the BOM item this tag marks.")
        values["item"] = _check_items(conn, packet_id, [body.item])[0]
    return values


@router.post("/{packet_id}/heat-tags", dependencies=WRITE)
def add_tag(
    packet_id: int,
    body: TagBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_HEATS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        values = _tag_values(conn, packet_id, body, set(body.model_fields_set) | {"item"})
        _live_page(conn, packet_id, values.get("page"))
        no = next_counter(conn, packet_id, "heat")
        conn.execute(
            "INSERT INTO heat_tags (packet_id, tag_no, page, item, fx, fy, bfx, bfy, by)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
            (
                packet_id,
                no,
                values.get("page"),
                values["item"],
                values.get("fx"),
                values.get("fy"),
                values.get("bfx"),
                values.get("bfy"),
                w.user_id,
            ),
        )
        w.changed("heat_tags")
        w.detail["tag_no"] = no
        w.log(f"Heat tag H{no} placed", f"item {values['item']}")
    return _read(d, pp, packet_id)


@router.patch("/{packet_id}/heat-tags/{no}", dependencies=WRITE)
def patch_tag(
    packet_id: int,
    no: int,
    body: TagPatchBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    if_match = parse_if_match(request)
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_HEATS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        row = _tag(conn, packet_id, no)
        check_row_version(row, if_match)
        _not_void(conn, packet_id, row["page"])
        changes = _diff(row, _tag_values(conn, packet_id, body, set(body.model_fields_set)))
        if not changes:
            w.skip()
            return _read(d, pp, packet_id)
        if "page" in changes:
            _live_page(conn, packet_id, changes["page"])
        update_row(conn, "heat_tags", row["id"], changes)
        w.changed(*[f"heat_tag.{k}" for k in changes])
        w.detail["tag_no"] = no
        item = changes.get("item", row["item"])
        w.log(f"Heat tag H{no} {'moved' if set(changes) <= POSITION else 'changed'}", f"item {item}")
    return _read(d, pp, packet_id)


@router.delete("/{packet_id}/heat-tags/{no}", dependencies=WRITE)
def remove_tag(
    packet_id: int,
    no: int,
    request: Request,
    body: RemoveBody | None = None,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    body = body or RemoveBody()
    if_match = parse_if_match(request)
    reason = clean_opt(body.reason, LIMITS["reason"])
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_HEATS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        row = _tag(conn, packet_id, no)
        if if_match is not None:
            check_row_version(row, if_match)
        _not_void(conn, packet_id, row["page"])
        conn.execute(
            "UPDATE heat_tags SET removed_at = now(), removed_by = %s, removed_reason = %s,"
            " row_version = row_version + 1 WHERE id = %s",
            (w.user_id, reason, row["id"]),
        )
        w.changed("heat_tags")
        w.detail["tag_no"] = no
        w.log(f"Heat tag H{no} removed", " · ".join(x for x in (f"item {row['item']}", reason) if x))
    return _read(d, pp, packet_id)


@router.put("/{packet_id}/heat-table/{page}", dependencies=WRITE)
def put_heat_table(
    packet_id: int,
    page: int,
    body: HeatTableBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Place, move or resize the heat table on a drawing page. The same placement again changes nothing."""
    if page < 1:
        raise ApiError(422, "invalid_page", "Pages start at 1.")
    if not (0 <= body.x <= HT_XY_MAX and 0 <= body.y <= HT_XY_MAX and HT_S_MIN <= body.s <= HT_S_MAX):
        raise ApiError(
            422,
            "invalid_heat_table",
            f"Keep the heat table on the sheet: x and y from 0 to {HT_XY_MAX}, size from {HT_S_MIN} to {HT_S_MAX}.",
        )
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_HEAT_TABLE, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        pg = _drawing_page(conn, packet_id, page)
        if pg["void_at"] is not None:
            raise ApiError(422, "void_page", MSG_VOID)
        if body.carried_from is not None and _page_row(conn, packet_id, body.carried_from) is None:
            raise ApiError(422, "invalid_page", f"There is no sheet {body.carried_from} on this packet.")
        cur = packets.one(
            conn,
            "SELECT x, y, s, carried_from FROM heat_table_placements WHERE packet_id = %s AND page = %s",
            (packet_id, page),
        )
        if cur is not None and (cur["x"], cur["y"], cur["s"]) == (body.x, body.y, body.s):
            w.skip()
            return _read(d, pp, packet_id)
        carried = body.carried_from if "carried_from" in body.model_fields_set else (cur or {}).get("carried_from")
        conn.execute(
            "INSERT INTO heat_table_placements (packet_id, page, x, y, s, carried_from, by)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s)"
            " ON CONFLICT (packet_id, page) DO UPDATE SET x = EXCLUDED.x, y = EXCLUDED.y, s = EXCLUDED.s,"
            " carried_from = EXCLUDED.carried_from, by = EXCLUDED.by, at = now()",
            (packet_id, page, body.x, body.y, body.s, carried, w.user_id),
        )
        w.changed("heat_table")
        w.detail["page"] = page
        where = _page_text(pg, page)
        if cur is None:
            w.log("Heat table placed on drawing", where)
        elif (cur["x"], cur["y"]) != (body.x, body.y):
            w.log("Heat table moved", where)
        else:
            w.log("Heat table resized", f"{'A+' if body.s > cur['s'] else 'A-'} {_num_text(body.s)}")
    return _read(d, pp, packet_id)


@router.delete("/{packet_id}/heat-table/{page}", dependencies=WRITE)
def remove_heat_table(
    packet_id: int,
    page: int,
    request: Request,
    body: It | None = None,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    body = body or It()
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_HEAT_TABLE, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        _not_void(conn, packet_id, page)
        gone = conn.execute(
            "DELETE FROM heat_table_placements WHERE packet_id = %s AND page = %s RETURNING id", (packet_id, page)
        ).fetchone()
        if gone is None:
            raise ApiError(404, "not_found", f"There is no heat table on sheet {page}.")
        w.changed("heat_table")
        w.detail["page"] = page
        w.log("Heat table removed from drawing", _page_text(_page_row(conn, packet_id, page), page))
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- the open stage


@router.patch("/{packet_id}/stage", dependencies=WRITE)
def patch_stage(
    packet_id: int,
    body: StageBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Filler, filler lot, preheat and notes of the open stage (the weld record).

    Either the flat body with If-Match (the stage's row_version), or `{set, base}`: field-level compare-and-set,
    409 field_conflict with `current` for a field someone else changed since `base`.
    """
    cas = body.set is not None
    if cas:
        assert body.set is not None
        bad = sorted(k for k in body.set if k not in STAGE_FIELDS)
        if bad:
            raise ApiError(422, "invalid_field", f"Not a weld record field: {', '.join(bad)}.")
        values = {k: clean_opt(v, STAGE_FIELDS[k]) for k, v in body.set.items()}
        base = {k: clean_opt(v, STAGE_FIELDS[k]) for k, v in (body.base or {}).items() if k in STAGE_FIELDS}
        if_match = None
    else:
        values = {k: clean_opt(getattr(body, k), n) for k, n in STAGE_FIELDS.items() if k in body.model_fields_set}
        base = {}
        if_match = parse_if_match(request)
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_STAGE_FIELDS, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        stage = _need_stage(conn, packet_id)
        if cas:
            changes = apply_field_cas({k: stage[k] for k in STAGE_FIELDS}, values, base)
        else:
            check_row_version(stage, if_match)
            changes = _diff(stage, values)
        if not changes:
            w.skip()
            return _read(d, pp, packet_id)
        update_row(conn, "packet_stages", stage["id"], changes)
        w.changed(*[f"stage.{k}" for k in changes])
        w.log("Weld record changed", _stage_note(changes))
    return _read(d, pp, packet_id)


def _stage_note(changes: dict[str, Any]) -> str | None:
    """The changed weld record fields: filler ER70S-6 · filler lot 1234."""
    parts: list[str] = []
    for k, name in STAGE_NOTE_NAMES.items():
        if k not in changes:
            continue
        v = changes[k]
        if k == "notes":
            parts.append("notes changed" if v else "notes cleared")
        else:
            parts.append(f"{name} {v}" if v else f"{name} cleared")
    return " · ".join(parts) or None


@router.put("/{packet_id}/stage/checklist/{key}", dependencies=WRITE)
def put_checklist(
    packet_id: int,
    key: str,
    body: ChecklistBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Tick or untick one checklist key. Per-key writes never conflict, so there is no If-Match."""
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.EDIT_CHECKLIST, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        weld = bool(w.state.weld_dept)
        templates = w.children.templates if w.children is not None else None
        if key not in checklists.checklist_keys(weld, templates):
            raise ApiError(422, "invalid_key", "That is not an item on this department's checklist.")
        stage = _need_stage(conn, packet_id)
        cur = (stage["checklist"] or {}).get(key)
        if checklists.is_done(cur) == body.done:
            w.skip()
            return _read(d, pp, packet_id)
        if body.done:
            stamp = _user_stamp(conn, w.user_id) or w.packet["welder_stamp"]
            conn.execute(
                "UPDATE packet_stages SET checklist = jsonb_set(checklist, ARRAY[%s]::text[],"
                " jsonb_build_object('done', true, 'at', now(), 'by', %s::text, 'stamp', %s::text)) WHERE id = %s",
                (key, str(w.user_id), stamp, stage["id"]),
            )
        else:
            conn.execute(
                "UPDATE packet_stages SET checklist = jsonb_set(checklist, ARRAY[%s]::text[], %s) WHERE id = %s",
                (key, Jsonb({"done": False}), stage["id"]),
            )
        items = checklists.checklist_for(weld, w.packet["drawing"], w.packet["rev"], w.packet["wps"], templates)
        text = next((i["text"] for i in items if i["key"] == key), key)
        w.changed(f"checklist.{key}")
        w.log(("Checked: " if body.done else "Unchecked: ") + text)
    return _read(d, pp, packet_id)


@router.post("/{packet_id}/stage/clock", dependencies=WRITE)
def clock(
    packet_id: int,
    body: ClockBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Start, pause or resume the stage clock. No version check: the clock state itself is the check."""
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.CLOCK, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        stage = _need_stage(conn, packet_id)
        running = stage["clock_running_since"] is not None
        bad = (
            (body.op == "start" and running)
            or (body.op == "pause" and not running)
            or (body.op == "resume" and (running or stage["clock_started_at"] is None))
        )
        if bad:
            raise ApiError(
                409,
                "clock_state",
                "The clock is running." if running else "The clock is stopped.",
                extra={"current": {"running": running, "started": stage["clock_started_at"] is not None}},
            )
        if body.op == "start":
            sets = "clock_running_since = now(), clock_started_at = coalesce(clock_started_at, now())"
            label = "Clock started"
        elif body.op == "pause":
            sets = (
                "clock_elapsed_ms = clock_elapsed_ms"
                " + (extract(epoch FROM (now() - clock_running_since)) * 1000)::bigint, clock_running_since = NULL"
            )
            label = "Paused"
        else:
            sets = "clock_running_since = now()"
            label = "Resumed"
        conn.execute(f"UPDATE packet_stages SET {sets} WHERE id = %s", (stage["id"],))  # noqa: S608 (fixed fragments)
        w.changed("stage.clock")
        w.log(label)
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- the warehouse stage


@router.post("/{packet_id}/wh", dependencies=WRITE)
def set_wh_stage(
    packet_id: int,
    body: WhBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Move the warehouse board: incoming, pulling, hold, pulled. expect_stage is what the caller saw."""
    reason = clean_opt(body.reason, LIMITS["reason"])
    note = clean_opt(body.note, WH_NOTE_MAX)
    if body.stage == "hold":
        if reason not in WH_HOLD_REASONS:
            raise ApiError(400, "invalid_reason", "Pick one of the hold reasons.")
        if reason == "Other" and note is None:
            raise ApiError(400, "reason_required", "Say what it's waiting on.")
    pp = _pp(d, p)
    conn = d.conn
    extra = {"reason": reason, "note": note}
    with packet_write(
        d, p, packet_id, A.SET_WH_STAGE, request=request, extra=extra, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        cur = current_wh(conn, packet_id)
        stage = cur["stage"] if cur else "incoming"
        if stage != body.expect_stage:
            raise ApiError(
                409,
                "stale_step",
                "Someone else already moved this packet on the warehouse board.",
                extra={"current": {"stage": stage, "version": int(w.packet["version"])}},
            )
        if body.stage not in WH_NEXT[stage]:
            raise ApiError(409, "wh_transition", f"A packet that is {stage} cannot move to {body.stage}.")
        wh_move(
            conn,
            packet_id,
            body.stage,
            w.user_id,
            reason=reason if body.stage == "hold" else None,
            note=note,
            auto=False,
        )
        if body.stage == "pulling" and stage in WH_PULLING_FROM:
            label = WH_PULLING_FROM[stage]
        else:
            label = WH_LABELS.get(body.stage, "")
        w.changed("wh")
        w.detail.update({"wh_from": stage, "wh_to": body.stage})
        w.log(
            f"Warehouse: {label}",
            " · ".join(x for x in (reason, note) if x) if body.stage == "hold" else note,
        )
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- Q.C.


QC_FIELDS = ("inspector", "stamp", "nde_method", "nde_result", "nde_report", "hardness", "notes")
QC_NOTE_NAMES = {"inspector": "inspector", "stamp": "stamp", "hardness": "hardness", "nde_report": "report"}


def _open_inspection(conn: Any, packet_id: int) -> dict[str, Any]:
    insp = packets.one(conn, "SELECT * FROM qc_inspections WHERE packet_id = %s AND closed_at IS NULL", (packet_id,))
    if insp is None:
        raise ApiError(409, "no_inspection", MSG_NO_INSPECTION)
    return insp


def _qc_value(key: str, value: Any) -> Any:
    """One scalar Q.C. field, cleaned and checked."""
    if key == "notes":
        return clean_opt(value, LIMITS["notes"])
    if key == "stamp":
        return _stamp(value)
    text = clean_opt(value, QC_FIELD_MAX)
    if key == "nde_method" and text is not None and text not in qc_dom.NDE_METHODS:
        raise ApiError(422, "invalid_field", "Pick an NDE method from the list.")
    if key == "nde_result" and text is not None and text not in qc_dom.NDE_RESULTS:
        raise ApiError(422, "invalid_field", "Pick an NDE result from the list.")
    return text


def _qc_logs(w: Write, changes: dict[str, Any]) -> None:
    """The log lines of a Q.C. field change: "NDE method: RT", "NDE result: Acceptable", then "Q.C. record changed"
    with the other fields. A method set to None required clears the result and report without a line of their own."""
    none = changes.get("nde_method") == qc_dom.NDE_NONE
    lines: list[tuple[str, str | None]] = []
    if "nde_method" in changes:
        lines.append((f"NDE method: {changes['nde_method'] or '-'}", None))
    if "nde_result" in changes and not (none and not changes["nde_result"]):
        lines.append((f"NDE result: {changes['nde_result'] or '-'}", None))
    parts: list[str] = []
    for k in ("inspector", "stamp", "hardness", "nde_report", "notes"):
        if k not in changes or (k == "nde_report" and none and not changes[k]):
            continue
        v = changes[k]
        if k == "notes":
            parts.append("notes changed" if v else "notes cleared")
        else:
            parts.append(f"{QC_NOTE_NAMES[k]} {v}" if v else f"{QC_NOTE_NAMES[k]} cleared")
    if parts:
        lines.append(("Q.C. record changed", " · ".join(parts)))
    if not lines:
        lines.append(("Q.C. record changed", None))
    w.log(*lines[0])
    for more in lines[1:]:
        w.log_also(*more)


@router.put("/{packet_id}/qc/checks/{key}", dependencies=WRITE)
def put_qc_check(
    packet_id: int,
    key: str,
    body: QcCheckBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Tick or untick one Q.C. check. Per-key writes never conflict, so there is no If-Match."""
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_QC, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        items = w.children.qc_items if w.children is not None and w.children.qc_items else None
        if key not in checklists.qc_keys(items):
            raise ApiError(422, "invalid_key", "That is not a Q.C. check item.")
        insp = _open_inspection(conn, packet_id)
        if checklists.is_done((insp["checks"] or {}).get(key)) == body.done:
            w.skip()
            return _read(d, pp, packet_id)
        if body.done:
            conn.execute(
                "UPDATE qc_inspections SET checks = jsonb_set(checks, ARRAY[%s]::text[],"
                " jsonb_build_object('done', true, 'at', now(), 'by', %s::text, 'stamp', %s::text)),"
                " row_version = row_version + 1 WHERE id = %s",
                (key, str(w.user_id), insp["stamp"], insp["id"]),
            )
        else:
            conn.execute(
                "UPDATE qc_inspections SET checks = jsonb_set(checks, ARRAY[%s]::text[], %s),"
                " row_version = row_version + 1 WHERE id = %s",
                (key, Jsonb({"done": False}), insp["id"]),
            )
        pk = w.packet
        texts = checklists.qc_items_for(pk["drawing"], pk["rev"], pk["wps"], pk["welder_stamp"], items)
        text = next((i["text"] for i in texts if i["key"] == key), key)
        w.changed(f"qc.checks.{key}")
        w.log(("Q.C. checked: " if body.done else "Q.C. unchecked: ") + text)
    return _read(d, pp, packet_id)


@router.put("/{packet_id}/qc/welds/{no}", dependencies=WRITE)
def put_qc_weld(
    packet_id: int,
    no: int,
    body: QcWeldBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """The visual exam of one weld: accept, reject or cleared (empty), and a note."""
    given = body.model_fields_set
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_QC, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        live = conn.execute(
            "SELECT 1 FROM welds WHERE packet_id = %s AND weld_no = %s AND removed_at IS NULL", (packet_id, no)
        ).fetchone()
        if live is None:
            raise ApiError(422, "invalid_weld", "No such weld on this packet.")
        insp = _open_inspection(conn, packet_id)
        k = str(no)
        cur = dict((insp["weld_exam"] or {}).get(k) or {})
        result = (body.result or "") if "result" in given else str(cur.get("result") or "")
        note = clean_text(body.note, LIMITS["note"]) if "note" in given else str(cur.get("note") or "")
        result_changed = result != str(cur.get("result") or "")
        note_changed = note != str(cur.get("note") or "")
        if not result_changed and not note_changed:
            w.skip()
            return _read(d, pp, packet_id)
        conn.execute(
            "UPDATE qc_inspections SET weld_exam = jsonb_set(weld_exam, ARRAY[%s]::text[],"
            " %s::jsonb || jsonb_build_object('at', now(), 'by', %s::text)), row_version = row_version + 1"
            " WHERE id = %s",
            (k, Jsonb({"result": result, "note": note}), str(w.user_id), insp["id"]),
        )
        label = qc_dom.weld_label(no)
        w.changed(f"qc.weld_exam.{k}")
        if result_changed:
            w.log(f"VT {result} · {label}" if result else f"VT cleared · {label}", note if result and note else None)
        else:
            w.log("Q.C. record changed", f"{label} note")
    return _read(d, pp, packet_id)


@router.patch("/{packet_id}/qc", dependencies=WRITE)
def patch_qc(
    packet_id: int,
    body: QcBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """The open inspection's fields.

    `{set, base}`: inspector, stamp, nde_method, nde_result, nde_report, hardness and notes with field-level
    compare-and-set (409 field_conflict with `current`). The phase 2 flat body with If-Match (the inspection's
    row_version) still works and merges checks and weld_exam per key. Setting the NDE method to None required clears
    the result and the report. The NACE flag is not written: the read model computes it from the WPS.
    """
    cas = body.set is not None
    values: dict[str, Any] = {}
    base: dict[str, Any] = {}
    if_match: int | None = None
    if cas:
        assert body.set is not None
        bad = sorted(k for k in body.set if k not in QC_FIELDS)
        if bad:
            raise ApiError(422, "invalid_field", f"Not a Q.C. field: {', '.join(bad)}.")
        values = {k: _qc_value(k, v) for k, v in body.set.items()}
        for k, v in (body.base or {}).items():
            if k in QC_FIELDS:
                base[k] = clean_opt(v, LIMITS["notes"] if k == "notes" else QC_FIELD_MAX)
    else:
        given = body.model_fields_set
        if body.checks:
            bad = sorted(set(body.checks) - set(checklists.qc_keys(checklists.load_qc_items(d.conn))))
            if bad:
                raise ApiError(422, "invalid_key", f"Not a Q.C. item: {', '.join(bad)}.")
        values = {k: _qc_value(k, getattr(body, k)) for k in QC_FIELDS if k in given}
        if_match = parse_if_match(request)
    if values.get("nde_method") == qc_dom.NDE_NONE:
        values["nde_result"] = None
        values["nde_report"] = None
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(d, p, packet_id, A.EDIT_QC, request=request, it_reason=body.it_reason, policy_principal=pp) as w:
        insp = _open_inspection(conn, packet_id)
        if cas:
            changes = apply_field_cas({k: insp[k] for k in QC_FIELDS}, values, base)
        else:
            check_row_version(insp, if_match)
            changes = _diff(insp, values)
        scalar = dict(changes)
        at = conn.execute("SELECT now()").fetchone()[0].isoformat()
        by = str(w.user_id)
        if not cas and body.checks:
            stamp = values.get("stamp", insp["stamp"])
            checks = dict(insp["checks"] or {})
            for k, done in body.checks.items():
                if checklists.is_done(checks.get(k)) == done:
                    continue
                checks[k] = {"done": True, "at": at, "by": by, "stamp": stamp} if done else {"done": False}
            if checks != (insp["checks"] or {}):
                changes["checks"] = Jsonb(checks)
        if not cas and body.weld_exam:
            live = {
                str(r[0])
                for r in conn.execute(
                    "SELECT weld_no FROM welds WHERE packet_id = %s AND removed_at IS NULL", (packet_id,)
                )
            }
            bad = sorted(set(body.weld_exam) - live)
            if bad:
                raise ApiError(422, "invalid_weld", f"No such weld on this packet: {', '.join(bad)}.")
            exam = dict(insp["weld_exam"] or {})
            for no, entry in body.weld_exam.items():
                cur = dict(exam.get(no) or {})
                new = dict(cur)
                if "result" in entry.model_fields_set:
                    new["result"] = entry.result or ""
                if "note" in entry.model_fields_set:
                    new["note"] = clean_text(entry.note, LIMITS["note"])
                if new != cur:
                    new.update({"at": at, "by": by})
                    exam[no] = new
            if exam != (insp["weld_exam"] or {}):
                changes["weld_exam"] = Jsonb(exam)
        if not changes:
            w.skip()
            return _read(d, pp, packet_id)
        update_row(conn, "qc_inspections", insp["id"], changes)
        w.changed(*[f"qc.{k}" for k in changes])
        if scalar:
            _qc_logs(w, scalar)
        else:
            w.log("Q.C. record changed")
    return _read(d, pp, packet_id)


# ---------------------------------------------------------------- change orders


@router.get("/{packet_id}/change-orders")
def list_change_orders(
    packet_id: int,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = _read(d, _pp(d, p), packet_id)["change_orders"]
    return out


def _required(conn: Any, codes: list[str]) -> list[str]:
    want: list[str] = []
    for c in codes:
        c = clean_text(c, 10).upper()
        if c and c not in want:
            want.append(c)
    bad = [c for c in want if c in NEVER_VERIFY or c not in active_depts(conn, want)]
    if bad:
        raise ApiError(
            422,
            "invalid_dept",
            f"Not a department that verifies change orders: {', '.join(bad)}.",
            extra={"depts": bad},
        )
    return want


def _co_log_detail(w: Write, no: int) -> None:
    w.changed("change_orders")
    w.detail["co_no"] = no


@router.post("/{packet_id}/change-orders", dependencies=WRITE)
def issue_co(
    packet_id: int,
    body: CoIssueBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Clerical issues a change order by hand. It is approved at once and holds the packet until every required
    department verifies it."""
    reason = clean_opt(body.reason, LIMITS["reason"])
    if reason is None:
        raise ApiError(400, "reason_required", "Say what changed and why.")
    ecn = clean_opt(body.ecn, LIMITS["ecn"])
    pp = _pp(d, p)
    conn = d.conn
    content = {
        "action": str(A.CO_ISSUE),
        "packet_id": packet_id,
        "reason": reason,
        "ecn": ecn,
        "required_depts": body.required_depts,
        "items": [i.model_dump() for i in body.items],
    }
    with packet_write(
        d,
        p,
        packet_id,
        A.CO_ISSUE,
        request=request,
        it_reason=body.it_reason,
        sign=body.sign,
        sign_content=content,
        policy_principal=pp,
    ) as w:
        required = (
            _required(conn, body.required_depts) if body.required_depts is not None else co_touched(conn, w.packet)
        )
        pages = {
            int(r["n"]): r
            for r in packets.rows(conn, "SELECT n, dwg, rev FROM packet_pages WHERE packet_id = %s", (packet_id,))
        }
        for it in body.items:
            for n in (it.old_page, it.new_page):
                if n is not None and n not in pages:
                    raise ApiError(422, "invalid_page", f"There is no page {n} on this packet.")
        no = next_counter(conn, packet_id, "co")
        sig = w.signature.id if w.signature else None
        co_id = conn.execute(
            "INSERT INTO change_orders (packet_id, co_no, origin, review, hold, held_by, held_at, ecn, reason,"
            " issued_by, issued_signature_id, approved_at, approved_by, approved_signature_id)"
            " VALUES (%s, %s, 'manual', 'approved', %s, %s, CASE WHEN %s THEN now() END, %s, %s, %s, %s, now(), %s, %s)"
            " RETURNING id",
            (
                packet_id,
                no,
                body.hold,
                w.user_id if body.hold else None,
                body.hold,
                ecn,
                reason,
                w.user_id,
                sig,
                w.user_id,
                sig,
            ),
        ).fetchone()[0]
        for it in body.items:
            old = pages.get(it.old_page) if it.old_page is not None else None
            new = pages.get(it.new_page) if it.new_page is not None else None
            conn.execute(
                "INSERT INTO co_items (co_id, old_page, old_dwg, old_rev, new_page, new_dwg, new_rev, added)"
                " VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
                (
                    co_id,
                    it.old_page,
                    old["dwg"] if old else None,
                    old["rev"] if old else None,
                    it.new_page,
                    new["dwg"] if new else None,
                    new["rev"] if new else None,
                    it.added,
                ),
            )
        for code in required:
            conn.execute("INSERT INTO co_required_depts (co_id, dept_code) VALUES (%s, %s)", (co_id, code))
        _co_log_detail(w, no)
        w.detail["required"] = required
        verify = f"verify: {', '.join(required)}" if required else "no departments to verify"
        w.log(f"Change order {_co_no_text(no)} issued", " · ".join(x for x in (reason, ecn, verify) if x))
    return _read(d, pp, packet_id)


FEED_ORIGINS = ("visual", "drawing")
MSG_CO_NOT_STAGED = "The new drawing has not been fetched from M-Files yet."


def _vis_text(leg: Any, *names: str) -> str:
    for n in names:
        v = leg.get(n) if isinstance(leg, dict) else None
        if v is not None and v != "":
            return str(v)
    return ""


def _leg_row(leg: Any) -> dict[str, Any]:
    """One work order row from a change order's `vis.legs` entry. Accepts a flat leg ({wo|wo_base|base, lot, split,
    sub, seq, ...}) or one with a nested key ({key: {base, lot, split, sub}, seq, ...})."""
    if not isinstance(leg, dict):
        raise ApiError(422, "bad_change_order", "The change order's work order list cannot be read.")
    key = leg.get("key") if isinstance(leg.get("key"), dict) else leg
    base = _vis_text(key, "base", "wo_base", "wo")
    if not base:
        raise ApiError(422, "bad_change_order", "The change order's work order list cannot be read.")
    seq_raw = leg.get("seq")
    return {
        "wo_base": base,
        "lot": _vis_text(key, "lot"),
        "split": _vis_text(key, "split"),
        "sub": _vis_text(key, "sub"),
        "seq": int(seq_raw) if seq_raw not in (None, "") else None,
        "qty": _num(leg.get("qty")),
        "target_hrs": _num(leg.get("tgt", leg.get("target_hrs"))),
        "op": clean_opt(leg.get("op"), LIMITS["op"]),
        "dwg": clean_opt(leg.get("dwg"), LIMITS["drawing"]),
        "rev": clean_opt(leg.get("rev"), LIMITS["rev"]),
        "partspecs": clean_opt(leg.get("partspecs"), 4000),
        "visual_status": clean_opt(leg.get("st", leg.get("visual_status")), 10),
    }


def _apply_legs(conn: Any, packet_id: int, legs: list[Any], title: str | None) -> dict[str, int]:
    """Replace the packet's work orders with the change order's legs. Rows whose key and seq match are updated in
    place (their ids and gone marks stay), new ones are added, the rest are deleted."""
    want = [_leg_row(lg) for lg in legs]
    have = packets.rows(conn, "SELECT * FROM packet_work_orders WHERE packet_id = %s", (packet_id,))

    def k(r: dict[str, Any]) -> tuple[Any, ...]:
        return (r["wo_base"], r["lot"], r["split"], r["sub"], r["seq"])

    by_key = {k(r): r for r in have}
    seen: set[tuple[Any, ...]] = set()
    counts = {"added": 0, "updated": 0, "removed": 0}
    for sort, r in enumerate(want):
        key = k(r)
        if key in seen:
            continue
        seen.add(key)
        cur = by_key.get(key)
        if cur is not None:
            conn.execute(
                "UPDATE packet_work_orders SET qty = %s, target_hrs = %s, op = %s, dwg = %s, rev = %s, partspecs = %s,"
                " visual_status = %s, sort = %s, row_version = row_version + 1 WHERE id = %s",
                (
                    r["qty"],
                    r["target_hrs"],
                    r["op"],
                    r["dwg"],
                    r["rev"],
                    r["partspecs"],
                    r["visual_status"],
                    sort,
                    cur["id"],
                ),
            )
            counts["updated"] += 1
        else:
            conn.execute(
                "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq, qty, target_hrs, op,"
                " description, visual_status, dwg, rev, partspecs, sort)"
                " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
                (
                    packet_id,
                    r["wo_base"],
                    r["lot"],
                    r["split"],
                    r["sub"],
                    r["seq"],
                    r["qty"],
                    r["target_hrs"],
                    r["op"],
                    title,
                    r["visual_status"],
                    r["dwg"],
                    r["rev"],
                    r["partspecs"],
                    sort,
                ),
            )
            counts["added"] += 1
    gone = [r["id"] for r in have if k(r) not in seen]
    if gone:
        conn.execute("DELETE FROM packet_work_orders WHERE id = ANY(%s)", (gone,))
        counts["removed"] = len(gone)
    return counts


def _plural(n: int, word: str) -> str:
    return f"{n} {word}{'' if n == 1 else 's'}"


def _apply_feed_co(w: Write, co: dict[str, Any]) -> str | None:
    """Apply a VISUAL or M-Files change order on approval: work orders, BOM, header drawing and rev, the fingerprint,
    and the staged drawing (old pages voided, new pages carry the change order). Returns a note for the log."""
    from . import packet_visual as pv

    conn = w.conn
    pid = w.id
    vis = co["vis"] if isinstance(co["vis"], dict) else {}
    staged = co["staged"] if isinstance(co["staged"], dict) else None
    row = w.packet
    visual = dict(row["visual"] or {})
    notes: list[str] = []
    sets: dict[str, Any] = {}
    if isinstance(vis.get("legs"), list) and vis["legs"]:
        c = _apply_legs(conn, pid, vis["legs"], row["title"])
        notes.append(f"work orders: {c['added']} added, {c['updated']} kept, {c['removed']} removed")
    if isinstance(vis.get("bom"), list) and not visual.get("manual_bom"):
        c = pv.merge_visual_bom(conn, pid, vis["bom"])
        notes.append(f"BOM: {c['added']} added, {c['updated']} kept, {c['removed'] + c['na']} dropped")
    if co["origin"] == "visual":
        if vis.get("to_dwg") not in (None, ""):
            sets["drawing"] = clean_opt(vis["to_dwg"], LIMITS["drawing"])
        if vis.get("to_rev") not in (None, ""):
            sets["rev"] = clean_opt(vis["to_rev"], LIMITS["rev"])
        if co["target_fp"]:
            sets["visual_fp_applied"] = co["target_fp"]
        for col, key in (("visual_want", "want"), ("visual_rel", "rel")):
            if key in vis:
                sets[col] = vis[key] or None
        for key in ("fp_parts", "folded", "partspecs"):
            if key in vis:
                visual[key] = vis[key]
    flags = list(row["flags"] or [])
    if staged and staged.get("asset_id"):
        asset = pv.pdf_asset(conn, staged["asset_id"])
        dwg = sets.get("drawing") or staged.get("dwg") or row["drawing"]
        rev = str(staged["rev"]) if staged.get("rev") not in (None, "") else (sets.get("rev") or row["rev"])
        first, count, old = pv.replace_drawing_pages(
            conn, pid, asset, dwg=dwg, rev=rev, co_id=int(co["id"]), void_by=w.user_id
        )
        _carry_heat_tables(w, old, first, count)
        for i in range(max(len(old), count)):
            o = old[i] if i < len(old) else None
            new = i < count
            conn.execute(
                "INSERT INTO co_items (co_id, old_page, old_dwg, old_rev, new_page, new_dwg, new_rev, added)"
                " VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
                (
                    co["id"],
                    o["n"] if o else None,
                    o["dwg"] if o else None,
                    o["rev"] if o else None,
                    first + i if new else None,
                    dwg if new else None,
                    rev if new else None,
                    o is None,
                ),
            )
        sets["drawing_state"] = "attached"
        flags = pv.flags_without(flags, (*packets.DRAWING_FLAGS, "hand_drawing"))
        visual["rev_confirmed"] = True
        visual["vault"] = {
            "asset_id": str(asset["id"]),
            "name": staged.get("name"),
            "rev": rev,
            "at": pv.now_iso(),
            "co_no": co["co_no"],
        }
        visual.pop("manual_dwg", None)
        notes.append(f"drawing {dwg} rev {rev}, {_plural(count, 'page')}; {_plural(len(old), 'old page')} voided")
    sets["flags"] = flags
    sets["visual"] = Jsonb(visual)
    cols = ", ".join(f"{c} = %s" + ("::drawing_state" if c == "drawing_state" else "") for c in sets)
    conn.execute(f"UPDATE packets SET {cols} WHERE id = %s", (*sets.values(), pid))  # noqa: S608 (fixed names)
    w.changed(*(c for c in sets if c not in ("flags", "visual")), "change_orders")
    return "; ".join(notes) or None


def _carry_heat_tables(w: Write, old: list[dict[str, Any]], first: int, count: int) -> None:
    """Copy each heat table placement from a voided sheet to the sheet that replaces it (the old htCarry), with
    carried_from set to the old sheet. Welds and heat tags stay on the VOID sheet."""
    for i, o in enumerate(old):
        new_n = first + i if i < count else first
        cur = packets.one(
            w.conn, "SELECT x, y, s FROM heat_table_placements WHERE packet_id = %s AND page = %s", (w.id, o["n"])
        )
        if cur is None:
            continue
        done = w.conn.execute(
            "INSERT INTO heat_table_placements (packet_id, page, x, y, s, carried_from, by)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (packet_id, page) DO NOTHING RETURNING id",
            (w.id, new_n, cur["x"], cur["y"], cur["s"], o["n"], w.user_id),
        ).fetchone()
        if done is not None:
            w.changed("heat_table")
            w.log_also("Heat table copied to new sheet", "from the VOID copy, check its position on the new revision")


@router.post("/{packet_id}/change-orders/{no}/approve", dependencies=WRITE)
def approve_co(
    packet_id: int,
    no: int,
    body: SignOnly,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Approve a pending change order. A VISUAL or M-Files one is applied to the packet here; a minor one (target
    hours or want date only) needs no department verification."""
    pp = _pp(d, p)
    conn = d.conn
    pre = packets.one(
        conn,
        "SELECT origin::text AS origin, staged, target_fp FROM change_orders WHERE packet_id = %s AND co_no = %s",
        (packet_id, no),
    )
    content: dict[str, Any] = {"action": str(A.CO_APPROVE), "packet_id": packet_id, "co_no": no}
    if pre is not None and pre["origin"] in FEED_ORIGINS:
        content["target_fp"] = pre["target_fp"]
        content["staged_sha256"] = pre["staged"].get("sha256") if isinstance(pre["staged"], dict) else None
    with packet_write(
        d,
        p,
        packet_id,
        A.CO_APPROVE,
        request=request,
        it_reason=body.it_reason,
        sign=body.sign,
        sign_content=content,
        policy_principal=pp,
    ) as w:
        co = _co(conn, packet_id, no)
        _co_state(co, "pending", "approved")
        note: str | None = None
        if co["origin"] in FEED_ORIGINS:
            vis = co["vis"] if isinstance(co["vis"], dict) else {}
            if vis.get("dwg_change") is True and not co["staged"]:
                raise ApiError(409, "co_not_staged", MSG_CO_NOT_STAGED, extra={"waiting": co["waiting"]})
            note = _apply_feed_co(w, co)
        update_row(
            conn,
            "change_orders",
            co["id"],
            {
                "review": "approved",
                "approved_at": conn.execute("SELECT now()").fetchone()[0],
                "approved_by": w.user_id,
                "approved_signature_id": w.signature.id if w.signature else None,
            },
        )
        has_required = conn.execute("SELECT 1 FROM co_required_depts WHERE co_id = %s LIMIT 1", (co["id"],)).fetchone()
        if has_required is None and not co["minor"]:
            for code in co_touched(conn, w.packet):
                conn.execute("INSERT INTO co_required_depts (co_id, dept_code) VALUES (%s, %s)", (co["id"], code))
        _co_log_detail(w, no)
        w.log(f"Change order {_co_no_text(no)} approved", note)
    return _read(d, pp, packet_id)


@router.post("/{packet_id}/change-orders/{no}/reject", dependencies=WRITE)
def reject_co(
    packet_id: int,
    no: int,
    body: CoRejectBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    pp = _pp(d, p)
    conn = d.conn
    reason = clean_opt(body.reason, LIMITS["reason"])
    content = {"action": str(A.CO_REJECT), "packet_id": packet_id, "co_no": no, "reason": reason}
    with packet_write(
        d,
        p,
        packet_id,
        A.CO_REJECT,
        request=request,
        extra={"reason": reason},
        it_reason=body.it_reason,
        sign=body.sign,
        sign_content=content,
        policy_principal=pp,
    ) as w:
        co = _co(conn, packet_id, no)
        _co_state(co, "pending", "rejected")
        update_row(
            conn,
            "change_orders",
            co["id"],
            {
                "review": "rejected",
                "rejected_at": conn.execute("SELECT now()").fetchone()[0],
                "rejected_by": w.user_id,
                "rejected_reason": reason,
                "rejected_signature_id": w.signature.id if w.signature else None,
            },
        )
        _co_log_detail(w, no)
        w.log(f"Change order {_co_no_text(no)} rejected", reason)
    return _read(d, pp, packet_id)


@router.post("/{packet_id}/change-orders/{no}/hold", dependencies=WRITE)
def hold_co(
    packet_id: int,
    no: int,
    body: CoHoldBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """Hold the packet on a pending change order, or release it."""
    pp = _pp(d, p)
    conn = d.conn
    with packet_write(
        d, p, packet_id, A.CO_HOLD_TOGGLE, request=request, it_reason=body.it_reason, policy_principal=pp
    ) as w:
        co = _co(conn, packet_id, no)
        _co_state(co, "pending", "held or released")
        if bool(co["hold"]) == body.hold:
            w.skip()
            return _read(d, pp, packet_id)
        now = conn.execute("SELECT now()").fetchone()[0]
        update_row(
            conn,
            "change_orders",
            co["id"],
            {"hold": body.hold, "held_by": w.user_id if body.hold else None, "held_at": now if body.hold else None},
        )
        _co_log_detail(w, no)
        w.log(f"Change order {_co_no_text(no)} {'holds the packet' if body.hold else 'no longer holds the packet'}")
    return _read(d, pp, packet_id)


@router.post("/{packet_id}/change-orders/{no}/verify", dependencies=WRITE)
def verify_co(
    packet_id: int,
    no: int,
    body: CoVerifyBody,
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    """A supervisor of a required department verifies an approved change order. Verifications are never
    withdrawn."""
    dept = clean_text(body.dept, 10).upper()
    note = clean_opt(body.note, LIMITS["note"])
    pp = _pp(d, p)
    conn = d.conn
    content = {"action": str(A.CO_VERIFY), "packet_id": packet_id, "co_no": no, "dept": dept, "note": note}
    with packet_write(
        d,
        p,
        packet_id,
        A.CO_VERIFY,
        request=request,
        extra={"dept": dept},
        it_reason=body.it_reason,
        sign=body.sign,
        sign_content=content,
        policy_principal=pp,
    ) as w:
        co = _co(conn, packet_id, no)
        _co_state(co, "approved", "verified")
        required = {r[0] for r in conn.execute("SELECT dept_code FROM co_required_depts WHERE co_id = %s", (co["id"],))}
        if dept not in required:
            raise ApiError(422, "invalid_dept", f"Change order {_co_no_text(no)} does not need verification by {dept}.")
        done = conn.execute(
            "SELECT 1 FROM co_verifications WHERE co_id = %s AND dept_code = %s", (co["id"], dept)
        ).fetchone()
        if done is not None:
            raise ApiError(409, "already_verified", f"{dept} already verified change order {_co_no_text(no)}.")
        try:
            conn.execute(
                "INSERT INTO co_verifications (co_id, dept_code, by, note, signature_id, it_reason)"
                " VALUES (%s, %s, %s, %s, %s, %s)",
                (co["id"], dept, w.user_id, note, w.signature.id if w.signature else None, w.it_reason),
            )
        except psycopg.errors.UniqueViolation:
            raise ApiError(
                409, "already_verified", f"{dept} already verified change order {_co_no_text(no)}."
            ) from None
        _co_log_detail(w, no)
        w.detail["dept"] = dept
        names = packets.dept_names(conn)
        verified = {r[0] for r in conn.execute("SELECT dept_code FROM co_verifications WHERE co_id = %s", (co["id"],))}
        left = [names.get(c, c) for c in sorted(required) if c not in verified]
        tail = f"still waiting on {', '.join(left)}" if left else "change order fully verified"
        w.log(
            f"Change order {_co_no_text(no)} verified by {names.get(dept, dept)}",
            " · ".join(x for x in (note, tail) if x),
        )
    return _read(d, pp, packet_id)