#!/usr/bin/env python3
"""
zorros_print_agent.py — Local Kitchen Print Bridge
Run this on ANY computer at Zorro's on the same WiFi as the printer.

Polls Railway every 10 seconds for new orders → prints ESC/POS tickets
directly to 192.168.20.68:9100

Usage:
    pip install requests
    export PRINT_BRIDGE_TOKEN=your-token
    export PRINTER_IP=192.168.20.68  # optional; this is the default
    export BACKUP_PRINTER_IP=192.168.213.110  # optional 2nd printer; used if primary is down
    export RECEIPT_PRINTER_IP=192.168.20.60  # optional receipt station; customer/store/driver copies print there
    python3 zorros_print_agent.py --status            # no-token readiness check
    python3 zorros_print_agent.py --test-print         # local printer test
    python3 zorros_print_agent.py --test-bridge-print  # Railway + token + printer test
    python3 zorros_print_agent.py --doctor            # no-paper readiness check
    python3 zorros_print_agent.py --probe-completion   # certify GS ( H support (prints 1 slip)
    python3 zorros_print_agent.py               # start the live spooler
"""

import time, socket, datetime, json, os, re, sys, hashlib, requests

try:
    import fcntl
except ImportError:  # pragma: no cover - store agent runs on macOS/Linux
    fcntl = None

try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass


def _load_agent_env():
    """Fill missing config from print-agent.env, the file the wrapper sources.

    Owner report 2026-08-06: hand-run prints (importing this module directly to
    push a single ticket) came out all red and routed every duplicate to the
    kitchen printer.  Cause: only the launchd wrapper sources print-agent.env,
    so an ad-hoc interpreter had no KITCHEN_DARKNESS_BOOST (under-driven head =>
    red-only on two-color paper) and no RECEIPT_PRINTER_IP (copies fell back to
    the kitchen rail).  Reading the file here makes every entry point — wrapper,
    stale launchd plist, or a bare `import` — configure itself identically.
    Real environment always wins; this only supplies what is absent.
    """
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "print-agent.env")
    try:
        with open(path, "r", encoding="utf-8") as fh:
            lines = fh.readlines()
    except OSError:
        return
    for line in lines:
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[7:]
        key, sep, val = line.partition("=")
        if not sep:
            continue
        key = key.strip()
        val = val.strip().strip("'\"")
        if key and not os.environ.get(key):
            os.environ[key] = val


_load_agent_env()

RAILWAY_URL   = os.getenv("RAILWAY_URL", "https://zorro-phone-agent-production.up.railway.app").rstrip("/")
PRINTER_IP    = os.getenv("PRINTER_IP", "192.168.20.68").strip()
PRINTER_PORT  = int(os.getenv("PRINTER_PORT", "9100"))
# Optional second printer on the same LAN. When set, a ticket that fails to
# reach the primary is retried on the backup before the order is left for the
# next poll — so a single jammed/offline printer doesn't stall the whole line.
BACKUP_PRINTER_IP   = os.getenv("BACKUP_PRINTER_IP", "").strip()
BACKUP_PRINTER_PORT = int(os.getenv("BACKUP_PRINTER_PORT", str(PRINTER_PORT)))
# Optional receipt-station printer (owner station split 2026-08-02: kitchen
# ticket stays on the primary; CUSTOMER/STORE/DRIVER copies go here). When
# unset, or when this printer is down, copies print on the primary as before —
# a dead receipt printer must never cost the cashier their PAID/COLLECT copy.
RECEIPT_PRINTER_IP   = os.getenv("RECEIPT_PRINTER_IP", "").strip()
RECEIPT_PRINTER_PORT = int(os.getenv("RECEIPT_PRINTER_PORT", str(PRINTER_PORT)))
_PACK_COPY_ON = os.getenv("PRINT_PACK_COPY", "0").strip().lower() in ("1", "true", "yes", "on")   # owner 2026-09-29: off
# WHEN the delivery DRIVER COPY prints (Noe 2026-10-05: "when he assigns it,
# it doesn't print"). 'order' (default, today's behaviour) = one second after
# the kitchen ticket at order time, where it lands in the kitchen pile hours
# before a driver is booked. 'assign' = skip it here; the server queues a
# 'driver_slip' bridge job the moment a driver is named (ASSIGN / REASSIGN /
# LEAVING), so the paper count stays "1 delivery" (owner 9/29) instead of two.
DRIVER_COPY_AT = os.getenv("DRIVER_COPY_AT", "order").strip().lower() or "order"
POLL_INTERVAL = int(os.getenv("PRINT_POLL_INTERVAL", "10"))
PRINT_BRIDGE_TOKEN = os.getenv("PRINT_BRIDGE_TOKEN", "").strip()
# Real-time printer status gate. TCP-send success does NOT mean a ticket printed:
# a powered, networked printer that is out of paper or has its cover open accepts
# the bytes and silently drops the job. When enabled (default), we ask the printer
# whether it can actually print (ESC/POS DLE EOT) before marking an order printed.
# Set PRINTER_STATUS_CHECK=0 to disable if a printer misbehaves in the store.
STATUS_CHECK_ENABLED = os.getenv("PRINTER_STATUS_CHECK", "1").strip().lower() not in ("0", "false", "no", "off")
# Post-send STATUS SAMPLE. socket.sendall() only proves the OS accepted the
# bytes into the LOCAL send buffer — over the store VPN the tunnel can drop
# right after sendall returns and the ticket never reaches the printer while
# the order still gets marked printed (order #59, 2026-07-26 18:25Z). The
# pre-send status gate above cannot catch this: the connection was healthy
# when it ran. So after the ticket bytes we request real-time status ON THE
# SAME connection. IMPORTANT LIMITS (Epson data-processing reference): DLE EOT
# is a REAL-TIME command — the printer answers it as soon as it is received,
# while the ticket bytes may still sit unprocessed in its receive/print
# buffers. A reply is therefore evidence the link survived past the send plus
# a sample of current printer status — NOT proof the ticket was parsed or
# physically printed. Silence from a printer that is known to answer DLE EOT
# means the link likely died mid-stream: the ticket enters a bounded
# sent-unconfirmed retry (below) instead of being marked printed.
# Set PRINT_SEND_STATUS_CHECK=0 to disable if a printer misbehaves.
SEND_STATUS_CHECK_ENABLED = os.getenv("PRINT_SEND_STATUS_CHECK", "1").strip().lower() not in ("0", "false", "no", "off")
SEND_STATUS_TIMEOUT = float(os.getenv("PRINT_SEND_STATUS_TIMEOUT", "5"))
# ── ONE SAFE RETRY (audit 2026-09-26 item 5) ────────────────────────────────
# A sent-but-unconfirmed ticket (bytes left this Mac, nothing proved paper)
# gets EXACTLY ONE retry: after UNCONFIRMED_RETRY_BACKOFF seconds it is sent
# once more under a "*** POSSIBLE DUPLICATE — CHECK THE RAIL ***" banner.
# A second unconfirmed send HOLDS the order — never marked printed, staff
# paged ONCE — and there is never a third: not on the next poll, not after a
# restart (the send count and the hold live in printed_state.json). Only a
# deliberate human requeue (a POS correction/reprint, which carries the
# bridge's UPDATED ORDER marker) releases a hold, for one more bannered
# print. A hard rule, not an env knob — the 9/5 pile was retries nobody
# bounded, and a printer whose replies get lost must never become a storm.
MAX_UNCONFIRMED_SENDS = 2
UNCONFIRMED_RETRY_BACKOFF = float(os.getenv("PRINT_UNCONFIRMED_RETRY_BACKOFF", "30"))
# ── A TICKET NEVER PRINTS IN PIECES (D-22, 2026-09-26 17:07–17:57Z) ──────────
# The Epson at .60 dropped its link for 50 minutes. ping_printer logged
# "[PRINTER CHECK] Not reachable" every poll, yet every poll ALSO called
# _send_ticket for each pending order (check_printer_fault treats a failed
# connect as "can't tell" and lets the send go): 139 "[PRINTER ERROR] timed
# out" lines, 37 "Failed #999900027 — will retry". Each attempt that got PART
# of the ticket across before the 5 s socket timeout put a PARTIAL ticket on
# paper — D-22 (5 items) reached the driver with items missing, and several
# half pages piled on the rail. Rules now:
#   1. a send never starts unless the LAST reachability probe of that printer
#      (check_printer_fault / probe_printer_link, run right before every
#      order) succeeded — while it fails, tickets WAIT, nothing is sent and
#      nothing is marked, and the wait is logged once a minute, not per try;
#   2. ticket bytes go out in a tracked chunked loop with a socket timeout
#      scaled to the ticket length; a mid-send error logs
#      "[PARTIAL SEND] #id bytes=<sent>/<total>", re-queues the order and the
#      retry carries a top "*** REPRINT — PREVIOUS COPY WAS INCOMPLETE ***"
#      banner so the line discards the partial page; retries back off
#      30 s → 60 s → 120 s … capped at 5 min — never once per 10 s poll;
#   3. after PRINT_PARTIAL_PAGE_AFTER partial sends of the same order the
#      owner is paged ONCE (print-failed report + fault heartbeat) naming the
#      order and "printer link unstable at <ip>:<port>".
PARTIAL_PAGE_AFTER = int(os.getenv("PRINT_PARTIAL_PAGE_AFTER", "3"))
PARTIAL_BACKOFF_BASE = float(os.getenv("PRINT_PARTIAL_BACKOFF_BASE", "30"))
PARTIAL_BACKOFF_MAX = float(os.getenv("PRINT_PARTIAL_BACKOFF_MAX", "300"))
WAIT_LOG_SECONDS = float(os.getenv("PRINT_WAIT_LOG_SECONDS", "60"))
# Socket timeout for the ticket bytes: base + per-KB, capped. 5 s flat was
# too short for a 40-line ticket over a flapping link — the timeout fired
# with the ticket half sent.
SEND_TIMEOUT_BASE = float(os.getenv("PRINT_SEND_TIMEOUT_BASE", "5"))
SEND_TIMEOUT_PER_KB = float(os.getenv("PRINT_SEND_TIMEOUT_PER_KB", "2"))
SEND_TIMEOUT_MAX = float(os.getenv("PRINT_SEND_TIMEOUT_MAX", "60"))
SEND_CHUNK = max(64, int(os.getenv("PRINT_SEND_CHUNK", "1024")))
PROBE_TIMEOUT = float(os.getenv("PRINT_PROBE_TIMEOUT", "3"))
# A probe older than this no longer vouches for the link; print_ticket
# re-probes before it sends.
PROBE_FRESH_SECONDS = float(os.getenv("PRINT_PROBE_FRESH_SECONDS", "5"))
# TRUE job-completion barrier (ESC/POS GS ( H fn=48 "process ID response"):
# unlike real-time DLE EOT, GS ( H is an ordinary buffered command — the
# printer transmits the echoed process ID only AFTER all receive/print data
# before it has been processed, so its reply DOES confirm the ticket was
# consumed. That echo is the strongest acknowledgement a ticket is marked
# printed VERIFIED on (audit 2026-09-26 item 5). Owner 10/5 (Noe: "it printed
# normally"): a DLE EOT status byte answered ONLINE after the send also marks
# the ticket VERIFIED (see classify_post_send) — only silence, noise or a
# fault byte leaves it UNVERIFIED / errored, and always with the reason.
# A printer that does not know the command may print the trailing ID bytes
# as garbage on a live kitchen ticket, so a live ticket only ever carries
# the request to a printer that answered --probe-completion once (prints one
# small slip; run owner-approved, in a quiet window). That answer is
# CERTIFICATION, recorded in printer_capabilities.json beside
# printed_state.json, and from then on the completion check is the DEFAULT
# for that printer. PRINT_COMPLETION_CHECK:
#   unset / "auto" / "certification"  (DEFAULT) certified printers only;
#   "1"  legacy force — every printer (only if you know they all speak it);
#   "0"  kill switch — never, even certified printers.
def _completion_mode(raw):
    v = (raw or "").strip().lower()
    if v in ("1", "true", "yes", "on"):
        return "on"
    if v in ("0", "false", "no", "off"):
        return "off"
    return "auto"


COMPLETION_CHECK_MODE = _completion_mode(os.getenv("PRINT_COMPLETION_CHECK", ""))
# Legacy name kept for readers/tests: True only under the force flag.
COMPLETION_CHECK_ENABLED = COMPLETION_CHECK_MODE == "on"
COMPLETION_TIMEOUT = float(os.getenv("PRINT_COMPLETION_TIMEOUT", "15"))
# Cash-drawer kick (owner spec 2026-07-24, drawer control): the drawer plugs
# into the receipt printer's DK port; an ESC p pulse pops it. OFF by default —
# enable per store with DRAWER_KICK_ENABLED=1 once a drawer is physically
# attached, else the pulse harmlessly no-ops on drawerless printers.
DRAWER_KICK_ENABLED = os.getenv("DRAWER_KICK_ENABLED", "").strip().lower() in ("1", "true", "yes", "on")
# Reported in every heartbeat: after copying a new build to the store Mac,
# the bridge status page shows this string, so "is the fix live?" has an
# answer that is not a guess.
SPOOLER_VERSION = "2026-09-26-physical-ack-v8"
SPOOLER_LOCK_PATH = os.getenv(
    "PRINT_SPOOLER_LOCK_PATH", "/tmp/zorros_print_agent.lock"
).strip()

ESC=b'\x1b'; GS=b'\x1d'; DLE=b'\x10'; EOT=b'\x04'
INIT=ESC+b'@'; CUT=GS+b'V\x41\x05'
# Same pin-2 pulse backend/printer_client.py uses for its owner-gated kick.
OPEN_DRAWER=ESC+b'p\x00\x19\xfa'
BOLD_ON=ESC+b'E\x01'; BOLD_OFF=ESC+b'E\x00'
ALIGN_CTR=ESC+b'a\x01'; ALIGN_LEFT=ESC+b'a\x00'
DOUBLE_ON=GS+b'!\x11'; DOUBLE_OFF=GS+b'!\x00'
# Double HEIGHT only (GS ! 0x01). DOUBLE_ON is 0x11 — double width AND height —
# which halves the line to 16 columns and destroys the right-aligned price
# column. Tall-only keeps all 32 columns, so an item line reads twice as big
# across the kitchen while its price still lands flush right.
TALL_ON=GS+b'!\x01'; TALL_OFF=GS+b'!\x00'
INVERSE_ON=GS+b'B\x01'; INVERSE_OFF=GS+b'B\x00'
# Second print color (ESC r n). The BTP-M300A honors it on two-color thermal
# paper (black + red OR black + blue — the accent color is a property of the
# paper roll, not selectable). Ignored harmlessly on plain paper.
COLOR2_ON=ESC+b'r\x01'; COLOR2_OFF=ESC+b'r\x00'
LF=b'\n'
# Kitchen darkness boost (owner report 2026-08-02: BTP printed EVERYTHING red
# — on two-color paper, black only develops at high head energy; red develops
# at low energy, so an under-driven head turns the whole ticket red). When
# KITCHEN_DARKNESS_BOOST=1, kitchen-path tickets get GS ( K speed=2 +
# density=8 injected right after ESC @ (INIT resets these, so they must
# follow it). Non-persistent per ESC/POS — a power-cycle or ESC @ clears
# them — and printers that don't implement GS ( K ignore the bytes.
DARKNESS_BOOST_ENABLED = os.getenv("KITCHEN_DARKNESS_BOOST", "").strip().lower() in ("1", "true", "yes", "on")
DARKNESS_BOOST = GS+b'(K\x02\x00\x32\x02' + GS+b'(K\x02\x00\x31\x08'

def _apply_darkness_boost(data):
    """Inject the boost after the leading ESC @ so INIT can't wipe it."""
    if not DARKNESS_BOOST_ENABLED:
        return data
    if data.startswith(INIT):
        return INIT + DARKNESS_BOOST + data[len(INIT):]
    return DARKNESS_BOOST + data
# ESC/POS real-time status transmission (DLE EOT n): the printer answers with one
# status byte even while offline. n=2 = offline cause (cover open / paper end),
# n=4 = roll-paper sensor (paper end/near-end).
STATUS_OFFLINE = DLE + EOT + b'\x02'
STATUS_PAPER   = DLE + EOT + b'\x04'
# GS ( H fn=48: request a "process ID response" — the printer echoes the 4-byte
# ID (framed 0x37 ... 0x00) only after everything received before the command
# has been processed. Ordinary buffered command, NOT real-time: that is what
# makes it a completion barrier. Format: GS ( H pL pH fn m d1..d4.
COMPLETION_REQ_PREFIX = GS + b'(H\x06\x00\x30\x30'
COMPLETION_RSP_HEADER = b'\x37'

# ── Printed-order memory SURVIVES restarts (owner 2026-09-05: "please dont
# send milion prints for 1 single order"). This set used to be in-memory
# only, so every restart forgot everything and re-printed whatever the
# server still listed as pending — T-4 Kelly got six tickets. Backed by a
# small JSON file now: load at boot, append on every print, keep the last
# 2000 ids. A restart can no longer cause a duplicate ticket.
# ZORROS_PRINT_STATE_DIR lets tests (and a relocated agent) keep the registry
# files somewhere other than beside this script. Default: beside the script,
# exactly where the store's printed_ids.json has always lived.
_STATE_DIR = (os.getenv("ZORROS_PRINT_STATE_DIR") or "").strip() or \
    os.path.dirname(os.path.abspath(__file__))
_PRINTED_STORE = os.path.join(_STATE_DIR, "printed_ids.json")


def _load_printed():
    try:
        with open(_PRINTED_STORE, "r", encoding="utf-8") as fh:
            ids = json.load(fh)
        return set(ids if isinstance(ids, list) else [])
    except Exception:
        return set()


def _persist_printed():
    try:
        ids = sorted(_printed, key=lambda x: str(x))[-2000:]
        tmp = _PRINTED_STORE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump(ids, fh)
        os.replace(tmp, _PRINTED_STORE)
    except Exception as e:
        log(f"[PRINTED STORE WARN] could not persist printed ids: {e}")


def note_manually_printed(order_id, order=None):
    """Record an order printed by hand (operator script) so the spooler
    never prints it again — even after a restart.

    Pass the order payload too when you have it: the requeue detector below
    then knows WHAT was printed, so a later counter correction or payment on
    that order still gets its one update ticket instead of a silent heal."""
    _printed.add(order_id)
    _persist_printed()
    if isinstance(order, dict):
        try:
            _record_print_state(order_id, order, marked=False)
        except Exception as e:
            log(f"[PRINT STATE WARN] could not record hand-print state for #{order_id}: {e}")


_printed = _load_printed()
# Ids whose server "printed" flag this process has already tried to heal
# (printed locally, server still said pending) — once per id per process.
_healed_marks = set()

# ── Deliberate REQUEUE vs lost mark (2026-09-16) ──────────────────────────────
# The registry above answers "did this Mac ever print #N?" — and since 9/6 a
# "yes" turned every re-served id into a HEAL: repair the server's flag, print
# nothing. Right for T-24 (the mark-printed POST was lost on a blip), wrong
# for every DELIBERATE requeue: a counter correction (order_mods.replace_items
# → kitchen_printed=0 + reprint_marker, served with is_update_reprint and an
# '*** UPDATED ORDER — NOT A NEW ORDER ***' item), a payment landing on a
# ticket already on the rail (webhook → kitchen_printed=0 for the PAID
# reprint), a scheduled order released at prep time. 75 heals by 9/15 — every
# one followed a CLEAN mark, i.e. the server had acked the print and a writer
# then cleared the flag on purpose. T-43/T-44/D-51/T-62/T-64 on 9/15 were
# payments settling minutes after the print; their PAID tickets never reached
# paper, and no correction has printed since 9/6.
#
# printed_state.json remembers, per id, WHAT was printed (a fingerprint of the
# ticket-relevant payload), whether it was PAID, and whether the server ACKED
# the mark. A re-served id is then:
#   • the very ticket we last printed and the mark never acked → a lost mark:
#     heal the flag, print nothing (the 9/6 behaviour);
#   • bridge marker present / payload changed / mark previously acked → a
#     deliberate requeue: print ONE kitchen ticket bannered NOT A NEW ORDER,
#     mark again. No PACK / DRIVER / PAID copies — the owner's 9/6 maximum
#     (2 kitchen + 1 driver + 1 paid) was already spent on the first print.
# MAX_REQUEUE_PRINTS caps update tickets per id so a server that keeps
# re-clearing a flag can never become a ticket storm.
_STATE_STORE = os.path.join(_STATE_DIR, "printed_state.json")
MAX_REQUEUE_PRINTS = int(os.getenv("PRINT_MAX_REQUEUE_PRINTS", "3"))
_PAID_WORDS = ("paid", "captured", "settled")
_UPDATE_MARKER_WORDS = ("updated order", "not a new order")

# ── Printer capabilities SURVIVE restarts: which printers answered the GS ( H
# completion probe. --probe-completion prints one slip and, on the echo,
# records the printer here; from then on every ticket to it is completion-
# checked by default and a mark says VERIFIED only on that echo.
_CAPS_STORE = os.path.join(_STATE_DIR, "printer_capabilities.json")


def _caps_key(ip, port):
    return "%s:%d" % (ip, int(port))


def _load_caps():
    try:
        with open(_CAPS_STORE, "r", encoding="utf-8") as fh:
            data = json.load(fh)
        return {str(k): v for k, v in data.items() if isinstance(v, dict)} \
            if isinstance(data, dict) else {}
    except Exception:
        return {}


def _persist_caps():
    try:
        tmp = _CAPS_STORE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump(_printer_caps, fh)
        os.replace(tmp, _CAPS_STORE)
    except Exception as e:
        log(f"[CAPS WARN] could not persist printer capabilities: {e}")


_printer_caps = _load_caps()


def note_completion_certified(ip, port, elapsed=None):
    """The printer at ip:port echoed a GS ( H process ID (--probe-completion):
    remember it, so the completion check is ON for it from now on."""
    entry = dict(_printer_caps.get(_caps_key(ip, port)) or {})
    entry["completion_certified_at"] = _log_ts()
    if elapsed is not None:
        entry["completion_echo_ms"] = int(float(elapsed) * 1000)
    _printer_caps[_caps_key(ip, port)] = entry
    _persist_caps()


def _completion_certified(ip, port):
    return bool((_printer_caps.get(_caps_key(ip, port)) or {}).get("completion_certified_at"))


def _completion_enabled_for(ip, port):
    """Whether a ticket to ip:port carries the GS ( H completion request."""
    if COMPLETION_CHECK_MODE == "on":
        return True
    if COMPLETION_CHECK_MODE == "off":
        return False
    return _completion_certified(ip, port)


def _load_print_state():
    try:
        with open(_STATE_STORE, "r", encoding="utf-8") as fh:
            data = json.load(fh)
        return {str(k): v for k, v in data.items() if isinstance(v, dict)} \
            if isinstance(data, dict) else {}
    except Exception:
        return {}


def _persist_print_state():
    try:
        keys = sorted(_print_state, key=lambda k: str(_print_state[k].get("ts", "")))[-2000:]
        tmp = _STATE_STORE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump({k: _print_state[k] for k in keys}, fh)
        os.replace(tmp, _STATE_STORE)
    except Exception as e:
        log(f"[PRINT STATE WARN] could not persist print state: {e}")


def _merge_print_state_from_disk():
    """Another process (a hand-print script) may have recorded state since we
    loaded — newest timestamp wins, same idea as re-reading printed_ids.json."""
    for k, v in _load_print_state().items():
        cur = _print_state.get(k)
        if cur is None or str(v.get("ts", "")) > str(cur.get("ts", "")):
            _print_state[k] = v


def _is_paid_payload(o):
    return str(o.get("payment_status") or "").strip().lower() in _PAID_WORDS


def _is_bridge_banner_item(it):
    return "***" in str(it.get("name") or it.get("display_name") or "")


def _requeue_marker_text(o):
    """The bridge's UPDATED ORDER marker when this payload carries one, else ''.

    print_bridge_api prepends orders.reprint_marker as the first item and sets
    is_update_reprint=True. build_ticket drops every '***' pseudo-item, so the
    marker is rendered here as the requeue banner instead."""
    try:
        for it in _coerce_items(o.get("items") or o.get("order_items") or []):
            if not isinstance(it, dict) or not _is_bridge_banner_item(it):
                continue
            name = str(it.get("name") or it.get("display_name") or "")
            if any(w in name.lower() for w in _UPDATE_MARKER_WORDS):
                return name.strip("* ").strip()
    except Exception:
        pass
    if o.get("is_update_reprint") in (True, 1, "1", "true", "True"):
        return "UPDATED ORDER"
    return ""


def _ticket_fingerprint(o):
    """Hash of everything on this payload that changes what the ticket SAYS.

    Deliberately a curated subset, not the whole payload: money display
    fields and server-derived extras may be recomputed between polls, and a
    spurious difference here would print a duplicate ticket."""
    try:
        items = [it for it in _coerce_items(o.get("items") or o.get("order_items") or [])
                 if isinstance(it, dict) and not _is_bridge_banner_item(it)]
    except Exception:
        items = []
    basis = {
        "items": items,
        "paid": _is_paid_payload(o),
        "type": str(o.get("order_type") or "").strip().lower(),
        "sched": str(o.get("scheduled_for") or o.get("scheduled_time") or "").strip(),
        "notes": str(o.get("special_instructions") or "").strip(),
        "staff_notes": str(o.get("staff_print_notes") or "").strip(),
        "packing": o.get("packing_instructions") or [],
        "name": str(o.get("customer_name") or "").strip(),
        "addr": str(o.get("address") or o.get("delivery_address") or "").strip(),
        "marker": _requeue_marker_text(o),
    }
    raw = json.dumps(basis, sort_keys=True, default=str, ensure_ascii=True)
    return hashlib.sha1(raw.encode("utf-8")).hexdigest()


def _record_print_state(order_id, o, marked, requeue=False, **fields):
    """Remember what was printed for this id — and, since the physical-ack
    build, what PROVED it (`verified`/`evidence`, consumed from the evidence
    the last print left in _evidence_by_id) plus the one-safe-retry
    bookkeeping (`sends`, `held`) that must survive a restart. `fields`
    override any of those explicitly."""
    key = str(order_id)
    prev = _print_state.get(key) or {}
    ev = _evidence_by_id.pop(key, None)
    st = {
        "fp": _ticket_fingerprint(o),
        "paid": _is_paid_payload(o),
        "marked": bool(marked),
        "reprints": int(prev.get("reprints") or 0) + (1 if requeue else 0),
        "sends": int(prev.get("sends") or 0),
        "held": bool(prev.get("held")),
        "verified": bool(ev["verified"]) if ev else bool(prev.get("verified")),
        "evidence": str(ev["evidence"] if ev else (prev.get("evidence") or ""))[:300],
        "ts": _log_ts(),
    }
    st.update(fields)
    _print_state[key] = st
    _persist_print_state()


def _update_print_state(order_id, **fields):
    """Change a few fields of an existing state entry (no payload needed)."""
    st = _print_state.get(str(order_id))
    if st is None:
        return
    st.update(fields)
    st["ts"] = _log_ts()
    _persist_print_state()


def _note_state_marked(order_id):
    st = _print_state.get(str(order_id))
    if st is not None:
        st["marked"] = True
        st["ts"] = _log_ts()
        _persist_print_state()


def _classify_relisted(order_id, o):
    """Decide what a re-served id that is already in the registry means.

    Returns ("requeue", <banner head>) — print ONE update ticket — or
    ("heal", <reason>) — repair the server flag, print nothing."""
    st = _print_state.get(str(order_id))
    marker = _requeue_marker_text(o)
    if st is None:
        # Printed before this build, or by hand without a payload: the only
        # proof of a deliberate requeue is the bridge's own marker.
        if marker:
            return "requeue", "UPDATED ORDER"
        return "heal", "no local print state and no update marker"
    if int(st.get("reprints") or 0) >= MAX_REQUEUE_PRINTS:
        return "heal", f"requeue cap of {MAX_REQUEUE_PRINTS} update tickets reached"
    fp = _ticket_fingerprint(o)
    if fp == st.get("fp") and not st.get("marked"):
        # The exact ticket we last printed, and the server never acked the
        # mark: the T-24 lost-POST case. Not a requeue.
        return "heal", "same ticket, mark never acked"
    if marker:
        return "requeue", "UPDATED ORDER"
    if fp != st.get("fp"):
        if _is_paid_payload(o) and not st.get("paid"):
            return "paidcopy", "NOW PAID"   # owner 2026-09-29: no second kitchen ticket; customer PAID copy only
        return "requeue", "UPDATED ORDER"
    # Same content, and the server HAD acked our mark: only a writer clearing
    # kitchen_printed on purpose puts an acked ticket back in the feed.
    return "requeue", "REPRINT"


def _requeue_banner(head):
    """Two big centred lines above an update ticket: the reason, then
    NOT A NEW ORDER — so the line never makes the food twice."""
    b = bytearray(INIT + ALIGN_CTR + BOLD_ON + DOUBLE_ON)
    b += enc(str(head)[:16]) + LF
    b += enc("NOT A NEW ORDER") + LF
    b += DOUBLE_OFF + BOLD_OFF
    b += enc("=" * 32) + LF + ALIGN_LEFT
    return bytes(b)


# Top banner on the retry of a ticket that died mid-send. The printer already
# put PART of that ticket on paper (D-22, 9/26); the line must know to throw
# the partial page away and cook from this one.
REPRINT_BANNER = "*** REPRINT — PREVIOUS COPY WAS INCOMPLETE ***"


def _reprint_banner():
    """Inverse, double-size REPRINT head + the two lines the line must read,
    above a ticket whose previous copy was cut off mid-send (REPRINT_BANNER
    split so every line fits: 16 columns double-width, 32 single)."""
    b = bytearray(INIT + ALIGN_CTR + BOLD_ON + INVERSE_ON + DOUBLE_ON)
    b += enc("*** REPRINT ***") + LF
    b += DOUBLE_OFF
    b += enc("PREVIOUS COPY WAS INCOMPLETE") + LF
    b += enc("DISCARD THE PARTIAL PAGE") + LF
    b += INVERSE_OFF + BOLD_OFF
    b += enc("=" * 32) + LF + ALIGN_LEFT
    return bytes(b)


# Top banner on the ONE safe retry of a ticket whose first send nothing
# confirmed: the same ticket may already be on the rail (audit 9/26 item 5).
DUPLICATE_BANNER = "*** POSSIBLE DUPLICATE — CHECK THE RAIL ***"


def _duplicate_banner():
    """Inverse, double-size POSSIBLE / DUPLICATE head + the lines the line
    must read, above the retry of an unconfirmed send (16 columns double
    width, 32 single)."""
    b = bytearray(INIT + ALIGN_CTR + BOLD_ON + INVERSE_ON + DOUBLE_ON)
    b += enc("*** POSSIBLE ***") + LF
    b += enc("DUPLICATE") + LF
    b += DOUBLE_OFF
    b += enc("CHECK THE RAIL") + LF
    b += enc("FIRST SEND WAS NOT CONFIRMED") + LF
    b += enc("SAME TICKET MAY BE THERE") + LF
    b += INVERSE_OFF + BOLD_OFF
    b += enc("=" * 32) + LF + ALIGN_LEFT
    return bytes(b)


def _clock():
    """Monotonic seconds. A function so tests can drive the backoff clock."""
    return time.monotonic()


_print_state = _load_print_state()
_last_printer_error = ""
_last_pending_orders_seen = 0
_last_print_failure = ""
# True when the last print_ticket failure happened AFTER ticket bytes were
# sent (silence/reset during the post-send sample) — i.e. the ticket may or
# may not have printed. Distinguishes the bounded sent-unconfirmed path from
# plain connect/send errors, which never delivered bytes and retry freely.
_last_send_unconfirmed = False
# True when the last successful print_ticket was confirmed by the GS ( H
# completion barrier (not just status-sampled).
_last_print_completed = False
# True when the last successful print_ticket was confirmed by an ONLINE
# status answer after the send (owner 10/5: that ticket printed).
_last_print_confirmed_online = False
# Per-order count of sent-but-unconfirmed attempts, and orders held once the
# count hits MAX_UNCONFIRMED_SENDS (dedup: no more automatic reprints).
_unconfirmed_sends = {}
_held_unconfirmed = set()
_completion_seq = 0
# What the last successful print_ticket PROVED, in words — goes to the server
# in the existing mark-printed call (printed_verified + print_evidence) — and
# which printer (ip:port) the bytes went to (the backup, on a fallback).
_last_print_evidence = ""
_last_print_printer = ""
# Evidence per order id from the print the loop just made, consumed by the
# mark-printed payload and folded into printed_state.json.
_evidence_by_id = {}


def _restore_unconfirmed_from_state():
    """A restart must not be the third send: rebuild the send counters and
    the holds from printed_state.json (9/5 lesson — in-memory counters
    reset and the server still lists the order)."""
    for k, st in _print_state.items():
        try:
            oid = int(k)
        except (TypeError, ValueError):
            oid = k
        if oid in _printed:
            continue
        if st.get("held"):
            _held_unconfirmed.add(oid)
        elif int(st.get("sends") or 0) > 0 and not st.get("marked"):
            _unconfirmed_sends[oid] = int(st.get("sends") or 0)


_restore_unconfirmed_from_state()
# Printers seen answering an identifying DLE EOT byte this process lifetime.
# Only for these does post-send silence mean "connection died"; a printer that
# has never spoken (a mute Star model, an lpr relay) keeps legacy behavior so
# the status sample cannot put it into an infinite reprint loop.
_status_capable = set()
# ── Link state per (ip, port), written ONLY by reachability probes
# (check_printer_fault / probe_printer_link) and by a fully delivered send.
# {"up": bool, "at": monotonic seconds, "reason": str}. No entry = never
# probed in this process (hand-print scripts, tests) → legacy send-and-see.
_link_state = {}
# Last "[PRINTER WAIT]" line per printer key — one line per WAIT_LOG_SECONDS,
# never one per poll or per order.
_wait_logged_at = {}
# Orders reported print-failed ("waiting: link down") during the CURRENT
# outage — once per order per outage; cleared when the link comes back.
_wait_reported = set()
# (sent, total) of the last print_ticket attempt that died MID-SEND with some
# bytes already handed to the printer, else None. Set by print_ticket; read by
# the polling loop, which owns the order id for the log line.
_last_partial_send = None
# Per-order count of partial sends, per-order earliest retry time (monotonic),
# and orders whose partial-send page already went out.
_partial_sends = {}
_retry_not_before = {}
_partial_paged = set()
# Set when the bridge rejects our print token (401/403). A rejected agent used
# to be INVISIBLE: fetch_orders swallowed the 401, returned no orders, and the
# next heartbeat reported a reachable printer — identical to a healthy store
# with nothing to print. Tickets piled up in the cloud and nobody was paged.
_auth_rejected = ""
_spooler_lock_file = None


def acquire_spooler_lock():
    """Allow exactly one live polling process on this store computer.

    The LaunchAgent and a manually started copy previously could poll the same
    pending order at the same time. Both printed the full kitchen/customer/store
    packet before either one marked the order printed, producing six slips for
    one pickup. The OS releases this lock automatically if the process exits or
    crashes, so it cannot leave printing permanently wedged.
    """
    global _spooler_lock_file
    if fcntl is None:
        log("[FATAL] Single-instance locking is unavailable on this computer")
        return False
    lock_path = SPOOLER_LOCK_PATH or "/tmp/zorros_print_agent.lock"
    lock_file = None
    try:
        lock_file = open(lock_path, "a+", encoding="ascii")
        fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        lock_file.seek(0)
        lock_file.truncate()
        lock_file.write(str(os.getpid()))
        lock_file.flush()
        _spooler_lock_file = lock_file
        return True
    except (OSError, IOError):
        try:
            if lock_file is not None:
                lock_file.close()
        except Exception:
            pass
        log(
            "[FATAL] Another Zorro's print agent is already running. "
            "This copy will exit to prevent duplicate tickets."
        )
        return False

def _log_ts():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def log(msg):
    """Timestamped stdout for the unattended polling loop.

    This agent runs headless behind a VPN that drops (7/25: two flapping
    episodes inside store hours). Undated lines made it impossible to tell a
    printer-unreachable burst from this minute apart from one from this
    morning — the log had to be dated by cross-referencing
    teleport-watchdog.log. Interactive commands keep plain print().
    """
    print(f"{_log_ts()} {msg}")

_TRANSLIT = str.maketrans({
    "—": "-", "–": "-", "‘": "'", "’": "'",
    "“": '"', "”": '"', "…": "...", " ": " ",
    # Web item names carry typographic marks the phone path never sent:
    # "Grandma Pizza — Sicilian 16×16" printed its × as "?".
    "×": "x", "″": '"', "′": "'", "•": "-",
    "½": "1/2", "¼": "1/4", "¾": "3/4",
})
def enc(t):
    # Transliterate common punctuation BEFORE ascii-replace: em-dashes were
    # printing as '?' on live tickets (owner photos 2026-08-03, T-31).
    return str(t).translate(_TRANSLIT).encode("ascii", "replace")

# HOT/COLD item marks (owner 2026-07-24). Self-contained keyword classifier —
# the on-prem agent has no menu brain; COLD keywords win over HOT ones so a
# grilled-chicken SALAD still reads cold. Default HOT (kitchen looks at it).
COLD_WORDS=("take-and-bake","take and bake","ready to bake","iced","cold",
            "bottled","chilled","salad","juice","lemonade","water","matcha",
            "yogurt","fruit","cookie","muffin","cannoli")
HOT_WORDS=("hot","grilled","toasted","baked","fries","soup","wings","pizza",
           "calzone","stromboli","burger","cheesesteak","parm","nashville",
           "quesadilla","meatball","nuggets","tenders","cutlet","breakfast",
           "egg","bacon","sausage")

def item_temperature(item):
    name=str(item.get("name") or "").lower()
    for w in COLD_WORDS:
        if w in name: return "COLD"
    for w in HOT_WORDS:
        if w in name: return "HOT"
    cat=str(item.get("category") or item.get("category_id") or "").lower()
    if cat in ("drinks","salad","cold-subs","cookies","muffins",
               "sweet-croissants","bread","dough-ready-to-bake","fresh-fruit-cups"):
        return "COLD"
    return "HOT"

def temperature_tag(temp):
    # ESC r (second colour) is a LINE attribute on this printer class, not a
    # character one: switching it on anywhere in a line paints the WHOLE line,
    # and switching it back off mid-line does not undo that.  The HOT tag used
    # COLOR2 and leads the item line, so every item name printed red too —
    # owner report 2026-08-06, "all modifiers and the item name print in red".
    # The item name is the line the kitchen reads from across the room and it
    # is meant to be big and BLACK, so the tag is emphasised without colour.
    # Red now survives only on lines that are red end-to-end: removals, sides,
    # extras and the payment banner (owner ruling 2026-07-28).
    if temp=="COLD":
        return INVERSE_ON+enc(" COLD ")+INVERSE_OFF+enc(" ")
    return INVERSE_ON+enc(" HOT ")+INVERSE_OFF+enc(" ")

# ── Ticket width (standard 32-column ESC/POS at normal size) ────────────────
TICKET_WIDTH = 32
# Modifier lines print DOUBLE width + height to match the POS ticket the owner
# holds as the standard (photos IMG_0591/IMG_0592, 2026-08-06) — that size is
# what the make station reads across the room.  Double width halves the roll to
# 16 columns, which is why the POS ticket truncates ("fresh apple slic",
# "ITALIAN SOFT ROL").  We wrap at 16 instead of truncating, so the last word
# of a build never goes missing.
MOD_WIDTH = 16

# ── Half-and-half modifier grouping (mirrors backend/printer_client.py) ─────
# The server flattens pizza builds into modifier strings like
# "left half: pepperoni" / "1st half no garlic". Those strings are a contract
# (the KDS board stores them verbatim), so grouping into a LEFT:/RIGHT: block
# happens here at print time only.
HALF_LINE_PREFIXES = (
    # (prefix, side label, render topping as a removal)
    ("left half: ",  "LEFT",     False),
    ("right half: ", "RIGHT",    False),
    ("1st half no ", "1ST HALF", True),
    ("2nd half no ", "2ND HALF", True),
    ("1st half: ",   "1ST HALF", False),
    ("2nd half: ",   "2ND HALF", False),
    ("half: ",       "HALF",     False),
)

def parse_half_modifier(mod_str):
    """Return (side_label, TOPPING_TEXT) for a half-scoped modifier, else None."""
    mod_lower = mod_str.lower()
    for prefix, label, is_removal in HALF_LINE_PREFIXES:
        if mod_lower.startswith(prefix):
            topping = mod_str[len(prefix):].strip().upper()
            if not topping:
                return None
            return label, ("NO " + topping) if is_removal else topping
    return None

def group_half_modifiers(mods):
    """Split modifier strings into ({side: [toppings…]}, [non-half mods])."""
    half_groups, other_mods = {}, []
    for mod in mods:
        if isinstance(mod, dict):
            mod = mod.get("name", str(mod))
        mod_str = str(mod).strip()
        # The bridge's pending-print payload stringifies web-order modifier
        # dicts, so "{'name': 'Well Done', 'price': 0}" arrives as TEXT and
        # printed verbatim (owner photos 2026-08-07, D-17: nobody in the
        # kitchen could read the tickets).  Parse the repr back to its name;
        # a paid add-on keeps its price visible.
        if mod_str.startswith("{") and "name" in mod_str:
            try:
                import ast
                parsed = ast.literal_eval(mod_str)
                if isinstance(parsed, dict):
                    nm = str(parsed.get("name") or parsed.get("NAME") or "").strip()
                    try:
                        pr = float(parsed.get("price") or parsed.get("PRICE") or 0)
                    except Exception:
                        pr = 0
                    if nm:
                        mod_str = f"{nm} (+${pr:.0f})" if pr else nm
            except Exception:
                pass
        if not mod_str:
            continue
        parsed = parse_half_modifier(mod_str)
        if parsed:
            half_groups.setdefault(parsed[0], []).append(parsed[1])
        else:
            other_mods.append(mod_str)
    return half_groups, other_mods

def wrap_lines(text, avail):
    """Word-wrap text to at most `avail` chars per line, hard-breaking any
    single word longer than `avail` (wrap, never truncate)."""
    avail = max(1, int(avail))
    lines, cur = [], ''
    for word in text.split():
        while len(word) > avail:
            if cur:
                lines.append(cur); cur = ''
            lines.append(word[:avail]); word = word[avail:]
        if not word:
            continue
        if len(cur) + len(word) + (1 if cur else 0) <= avail:
            cur = (cur + ' ' + word).strip()
        else:
            lines.append(cur); cur = word
    if cur:
        lines.append(cur)
    return lines or ['']

def mod_block(prefix, text, width=TICKET_WIDTH):
    """Render one modifier line wrapped to the ticket width; continuation
    lines indent to align under the text."""
    indent = len(prefix)
    parts = wrap_lines(text, width - indent)
    out = enc(prefix + parts[0]) + LF
    for cont in parts[1:]:
        out += enc(' ' * indent + cont) + LF
    return out

# Words that carry no build meaning — the marker glyph already says "side",
# and articles/politeness never change what the kitchen does.
_NOTE_FILLER = {"a", "an", "the", "of", "on", "in", "it", "and", "to", "with",
                "my", "our", "please", "pls", "thanks", "thank", "you",
                "side", "sides", "order", "just"}


def _note_words(text):
    return {w for w in re.findall(r"[a-z]+", str(text).lower())
            if w not in _NOTE_FILLER}


def _note_already_said(special, mods):
    """True when a modifier already carries the note's whole meaning.

    Owner 2026-08-06, salad sample: "side of caesar dressing" printed as a
    side AND "dressing on the side" printed again as a note — the same
    instruction twice, on the one line that has to be unambiguous.  The old
    check was exact string equality, so any rewording slipped through.
    Compare significant words instead: suppress only when every meaningful
    word of the note also appears in one modifier, so a note that adds
    anything new ("no croutons" against "extra croutons") still prints.
    """
    note = _note_words(special)
    if not note:
        return True
    return any(note <= _note_words(m) for m in (mods or []))


_BUILDS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "menu_builds.json")
_builds_cache = {"mtime": None, "index": {}}


def _load_menu_builds():
    """Read menu_builds.json, re-reading it whenever the file changes.

    Regenerated by build_menu_index.py from the menu descriptions.  A missing
    or unreadable file is not an error — tickets simply print without build
    lists, exactly as they did before.
    """
    try:
        mtime = os.path.getmtime(_BUILDS_PATH)
    except OSError:
        _builds_cache["mtime"], _builds_cache["index"] = None, {}
        return _builds_cache["index"]
    if _builds_cache["mtime"] != mtime:
        try:
            with open(_BUILDS_PATH, "r", encoding="utf-8") as fh:
                _builds_cache["index"] = json.load(fh) or {}
        except Exception as e:
            log(f"[BUILD INDEX] unreadable, printing without builds — {e}")
            _builds_cache["index"] = {}
        _builds_cache["mtime"] = mtime
    return _builds_cache["index"]


def _normalize_item_name(name):
    """Strip size and shape noise so an ordered item matches its menu row."""
    n = str(name or "").lower()
    n = re.sub(r'\d+\s*(?:"|inch|in\b)', " ", n)
    n = re.sub(r"\b(?:small|medium|large|lg|sm|md|personal|sicilian|round|"
               r"half|whole|slice|x\d+|\d+)\b", " ", n)
    n = re.sub(r"[^a-z ]", " ", n)
    return re.sub(r"\s+", " ", n).strip()


def _removed_words(mods):
    """Significant words of every "no X" / "without X" modifier on an item."""
    out = []
    for mod in mods or []:
        if isinstance(mod, dict):
            mod = mod.get("name", "")
        text = str(mod).strip().lower()
        if not text.startswith(("no ", "without ")):
            continue
        words = {w for w in re.findall(r"[a-z]+", text)} - {"no", "without"}
        if words:
            out.append(words)
    return out


def menu_build_for(name):
    """Component list for an ordered item, or [] when we have no recipe."""
    index = _load_menu_builds()
    if not index:
        return []
    entry = index.get(_normalize_item_name(name))
    return list(entry.get("components") or []) if entry else []


def ticket_build_for(name, item):
    """Recipe components safe to print beside this explicit order line."""
    components = menu_build_for(name)
    # Size has one authoritative, emphasized line from pizza_size_word().
    # Recipe prose must never contradict it (P-43: stored 12-inch line,
    # stale recipe printed MAKE: 16-INCH). Remove dimensions from MAKE rather
    # than letting two independent size sources reach the cook.
    cleaned = []
    for component in components:
        text = re.sub(r"\b(?:9|10|12|14|16|18|24)\s*(?:-|\s)?(?:inch\b|in\b|\")", "", str(component), flags=re.I)
        text = re.sub(r"\s{2,}", " ", text).strip(" ,-—")
        if text:
            cleaned.append(text)
    return cleaned


_COOKIE_PACKAGE_MARKERS = (
    "baker's bag", "bakers bag", "half dozen", "dozen", "cookie cake",
    "cookie pizza", "cookie bar", "cookie sandwich", "cookie dough",
    "party pack", "catering", "tray",
)
_STAFF_FRAGMENT_PATTERNS = (
    r"UNCONFIRMED\s*[—-]\s*CALL ENDED EARLY\s*[—-]\s*VERIFY WITH CUSTOMER IF POSSIBLE",
    r"UNCONFIRMED\s*[—-]\s*caller hung up before confirming(?:\s*;\s*owner ruling)?",
    r"UNCONFIRMED\s*[—-]\s*TEXT ORDER\s*[—-]\s*CUSTOMER MUST REPLY YES",
    r"CONFIRMED BY SMS", r"CONFIRMED BY STAFF",
    r"(?:⚠\s*)?CHECK BUILD\s*:[^\n]*", r"\bSTAFF\s*:[^\n]*",
    r"BUILD NOT VERIFIED[^\n]*", r"STAFF (?:CONFIRM|REVIEW|MUST)[^\n]*",
    r"PAY-LINK FAILED[^\n]*", r"REVIEW_REASON[^\n]*",
    r"SUPERSEDED BY[^\n]*", r"OWNER RULING[^\n]*",
    r"VERIFY WITH CUSTOMER IF POSSIBLE", r"PRICE REVIEW[^;.|]*",
    r"\[UPDATED[^\]]*\]",
)
_STAFF_FRAGMENT_RE = re.compile("|".join(_STAFF_FRAGMENT_PATTERNS), re.I)


def _packing_instructions(o, items):
    """Structured owner packaging from the server, with a safe local fallback."""
    raw = o.get("packing_instructions")
    if isinstance(raw, str):
        try:
            raw = json.loads(raw)
        except Exception:
            raw = []
    if isinstance(raw, list) and raw:
        return raw
    count = 0.0
    for item in items or []:
        if not isinstance(item, dict):
            continue
        name = str(item.get("name") or item.get("display_name") or "").strip().lower()
        if not name or "***" in name or "cookie" not in name:
            continue
        if any(marker in name for marker in _COOKIE_PACKAGE_MARKERS):
            continue
        try:
            count += max(float(item.get("quantity") or item.get("qty") or 1), 0.0)
        except (TypeError, ValueError):
            continue
    if count not in (1.0, 2.0):
        return []
    return [{"code": "BG-SM", "quantity": 1,
             "label": "SMALLEST PAPER BAG - NO BOX"}]


def _ticket_notes_for_surface(o, include_specs):
    """Keep CHECK BUILD and other diagnostics on kitchen copies only."""
    customer, staff = [], []
    seen_customer, seen_staff = set(), set()
    raw = str(o.get("notes") or o.get("special_instructions") or "")
    for match in _STAFF_FRAGMENT_RE.finditer(raw):
        text = " ".join(match.group(0).split()).strip(" ;|.")
        if any(marker in text.casefold() for marker in (
                "check build", "couldn't read", "could not read", "incomplete order")):
            text = "CHECK BUILD - REVIEW IN POS BEFORE MAKING"
        key = text.casefold()
        if text and key not in seen_staff:
            seen_staff.add(key)
            staff.append(text)
    customer_raw = _STAFF_FRAGMENT_RE.sub("", raw)
    customer_raw = re.sub(r"[;|]\s*(?=[;|])", "", customer_raw)
    # Semicolons and pipes separate audiences. A final period belongs to the
    # customer's instruction ("Call on arrival.") and must stay on paper.
    customer_raw = re.sub(r"(?:^[\s;|]+)|(?:[\s;|]+$)", "", customer_raw)
    for line in customer_raw.splitlines():
        text = " ".join(line.split()).strip()
        if not text:
            continue
        key = text.casefold()
        if key not in seen_customer:
            seen_customer.add(key)
            customer.append(text)
    if include_specs:
        for line in str(o.get("staff_print_notes") or "").splitlines():
            text = " ".join(line.split()).strip()
            if any(marker in text.casefold() for marker in (
                    "check build", "couldn't read", "could not read", "incomplete order")):
                text = "CHECK BUILD - REVIEW IN POS BEFORE MAKING"
            key = text.casefold()
            if text and key not in seen_staff:
                seen_staff.add(key)
                staff.append(text)
        return "\n".join(customer + staff)
    return "\n".join(customer)


def _coerce_items(raw):
    """Accept an items list OR the raw JSON string the orders table stores.

    The print bridge hands back parsed lists, so build_ticket only ever saw
    lists — until a ticket was pushed straight from a database row, where
    `items` is a JSON *string*.  list("[{...}]") then yields single characters,
    every one of which fails the isinstance(dict) filter, and the ticket
    printed with NO ITEMS AT ALL between the rules (owner report 2026-08-06,
    "the print from web orders is weird").  Silent, total loss of the only
    lines that matter, so parse defensively here instead of trusting callers.
    """
    if isinstance(raw, str):
        try:
            raw = json.loads(raw)
        except Exception:
            return []
    if isinstance(raw, dict):
        raw = [raw]
    return [it for it in list(raw or []) if isinstance(it, dict)]


def money(v):
    try:
        return f"${float(v):.2f}"
    except Exception:
        return ""

def is_positive(v):
    try:
        return float(v) > 0
    except Exception:
        return False

def auth_headers():
    if not PRINT_BRIDGE_TOKEN:
        raise RuntimeError("PRINT_BRIDGE_TOKEN is required for the print bridge")
    return {"X-Print-Bridge-Token": PRINT_BRIDGE_TOKEN}

# ── WHY is the printer unreachable? (owner 2026-10-04: "our system should
# have triggered the reason") ───────────────────────────────────────────────
# 10/4 12:32–3:50 pm the owner was paged "printer unreachable" seven times
# while the printer was fine: this computer's UniFi Teleport VPN to the store
# had dropped (an expired credential; the watchdog reconnected every 60 s and
# never completed the handshake). The agent can SEE that — the VPN service
# state, which interface the route to the printer leaves on, whether the
# port answers — so it says WHY and WHAT TO DO with every heartbeat, and the
# page the owner gets carries the fix instead of the symptom.
_TELEPORT_SERVICE_NAME = os.getenv("TELEPORT_SERVICE_NAME", "UniFi Teleport")
DIAG_PRINTER_OK = "PRINTER_OK"
DIAG_VPN_DOWN = "VPN_DOWN"
DIAG_PRINTER_OFF_ON_LAN = "PRINTER_OFF_ON_LAN"
DIAG_PRINTER_OFF_BEHIND_VPN = "PRINTER_OFF_BEHIND_VPN"


def _tcp_ok(ip, port, timeout=2.0):
    try:
        with socket.create_connection((ip, int(port)), timeout=timeout):
            return True
    except Exception:
        return False


def _teleport_state():
    """'Connected' / 'Disconnected' / 'Connecting' from scutil (macOS); 'unknown'
    elsewhere or when the service is not configured."""
    try:
        out = subprocess.run(["scutil", "--nc", "status", _TELEPORT_SERVICE_NAME],
                             capture_output=True, text=True, timeout=3).stdout
        first = (out.strip().splitlines() or [""])[0].strip()
        return first or "unknown"
    except Exception:
        return "unknown"


def _route_interface(ip):
    """The interface the route to ``ip`` leaves on ('utun4' = through a VPN,
    'en0' = the local network), '' when unknown."""
    try:
        out = subprocess.run(["route", "-n", "get", ip], capture_output=True, text=True, timeout=3).stdout
        m = re.search(r"interface:\s*(\S+)", out)
        return m.group(1) if m else ""
    except Exception:
        return ""


def _local_addr_for(ip):
    """This computer's own address on the path to ``ip`` (no packet is sent)."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect((ip, 9))
            return s.getsockname()[0]
        finally:
            s.close()
    except Exception:
        return ""


def _same_lan(a, b):
    return bool(a and b and a.rsplit(".", 1)[0] == b.rsplit(".", 1)[0])


def diagnose_printer_path(ip=None, port=None):
    """Why can't we reach the printer — in words a person at the store or the
    owner can act on. Never raises; cheap when the port answers."""
    ip = ip or PRINTER_IP
    port = int(port or PRINTER_PORT)
    if _tcp_ok(ip, port):
        return {"reason_code": DIAG_PRINTER_OK, "reason": "printer answers", "hint": "",
                "tcp_ok": True, "teleport_state": "", "route_interface": ""}
    teleport = _teleport_state()
    iface = _route_interface(ip)
    local = _local_addr_for(ip)
    via_tunnel = iface.startswith("utun")
    if _same_lan(local, ip):
        code = DIAG_PRINTER_OFF_ON_LAN
        reason = (f"this computer is on the store network but the printer at {ip} "
                  f"does not answer on port {port}")
        hint = "Check the printer at the store: power, paper, and its network light. Power-cycle it."
    elif via_tunnel or teleport.lower().startswith("connected"):
        code = DIAG_PRINTER_OFF_BEHIND_VPN
        reason = f"the VPN to the store is up but the printer at {ip} does not answer on port {port}"
        hint = "Check the printer at the store: power, paper, and its network light."
    else:
        code = DIAG_VPN_DOWN
        reason = (f"the VPN from this computer to the store is {teleport} "
                  f"(the route to the printer leaves on {iface or 'the local network'}); "
                  "the printer is probably fine")
        hint = ("On this computer open UniFi Endpoint or WiFiman -> Teleport -> On. "
                "If it will not stay on, sign in again or open a fresh Teleport link "
                "from the UniFi Network app.")
    return {"reason_code": code, "reason": reason, "hint": hint, "tcp_ok": False,
            "teleport_state": teleport, "route_interface": iface}


def post_heartbeat(printer_reachable=False, pending_orders_seen=None, mode="polling"):
    """Tell Railway the in-store print spooler is alive without sending PII."""
    payload = {
        "printer_ip": PRINTER_IP,
        "printer_port": PRINTER_PORT,
        "printer_reachable": bool(printer_reachable),
        "mode": mode,
        "version": SPOOLER_VERSION,
        "last_error": _last_printer_error or None,
        "pending_orders_seen": _last_pending_orders_seen if pending_orders_seen is None else pending_orders_seen,
    }
    try:
        payload["diagnosis"] = (
            {"reason_code": DIAG_PRINTER_OK, "reason": "printer answers", "hint": ""}
            if printer_reachable else diagnose_printer_path()
        )
    except Exception as e:  # never let the diagnosis break the heartbeat
        log(f"[HEARTBEAT WARN] diagnosis failed: {e}")
    try:
        r = requests.post(
            f"{RAILWAY_URL}/api/print/heartbeat",
            json=payload,
            headers=auth_headers(),
            timeout=5,
        )
        if r.status_code != 200:
            log(f"[HEARTBEAT WARN] print bridge returned HTTP {r.status_code}")
            return False
        return True
    except Exception as e:
        log(f"[HEARTBEAT WARN] {e}")
        return False

# ── Pizza size word (owner 2026-10-01) ──────────────────────────────────────
# "callers complain yesterday about ordering large pizza but agent confirming
# right but sending ticket for small" (D-48 / T-34, 2026-09-30). The ticket
# printed "1x CHEESE PIZZA / $14.00" with no size anywhere — the cook had to
# infer the size from the price, and the stored size label ("16-inch round")
# never reached the paper. Every whole pie now prints its size word under the
# item name, in the item name's own emphasis: LARGE 16" / SMALL 12" /
# 16x16 SQUARE / GF 9". Self-contained on purpose — this file runs on the
# store laptop with no backend import; backend/pizza_variant.py is the server
# twin and test_confirmed_size_is_the_variant_20261001 pins both.
_PIZZA_SIZE_SQUARE_RE = re.compile(r"16\s*x\s*16|\bsquare\b|\bsicilian\b", re.IGNORECASE)
_PIZZA_SIZE_GF_RE = re.compile(r"(?<!\d)9(?!\d)\s*(?:inch|in\b|\")|\bgluten\b|\bcauliflower\b", re.IGNORECASE)
_PIZZA_SIZE_LARGE_RE = re.compile(r"(?<!\d)(?:16|24)(?!\d)|\blarge\b|\bsixteen\b|\blg\b", re.IGNORECASE)
_PIZZA_SIZE_SMALL_RE = re.compile(r"(?<!\d)12(?!\d)|\b(?:small|personal|medium|twelve|sm)\b", re.IGNORECASE)
_PIZZA_NAME_SMALL_RE = re.compile(r"(?<!\d)12(?!\d)|\bsmall\b|\btwelve\b", re.IGNORECASE)
_PIZZA_NOT_A_WHOLE_PIE = (
    "slice", "sub", "burger", "croissant", "cookie", "roll", "crust", "dough",
    "quiche", "sashuka", "shakshuka", "breakfast", "flatbread", "calzone",
    "stromboli", "parbaked", "build your own", "bagel", "pretzel", "knots",
    "fries", "sauce", "sticks",
)
_PIZZA_TICKET_SIZE_WORDS = {"small": 'SMALL 12"', "large": 'LARGE 16"',
                            "square": "16x16 SQUARE", "gf": 'GF 9"'}


def _pizza_size_key(text):
    s = str(text or "").strip().lower()
    if not s:
        return ""
    if _PIZZA_SIZE_SQUARE_RE.search(s):
        return "square"
    if _PIZZA_SIZE_GF_RE.search(s):
        return "gf"
    if _PIZZA_SIZE_LARGE_RE.search(s):
        return "large"
    if _PIZZA_SIZE_SMALL_RE.search(s):
        return "small"
    return ""


def _pizza_name_size_key(name):
    n = str(name or "").strip().lower()
    if not n:
        return ""
    if _PIZZA_SIZE_SQUARE_RE.search(n):
        return "square"
    if _PIZZA_SIZE_GF_RE.search(n):
        return "gf"
    if _PIZZA_SIZE_LARGE_RE.search(n):
        return "large"
    if _PIZZA_NAME_SMALL_RE.search(n):
        return "small"
    if "grandma" in n and "personal" in n:
        return "small"      # the phone's own name for the 12" Grandma
    return ""


def _pizza_size_from_modifiers(modifiers):
    if isinstance(modifiers, str):
        modifiers = [modifiers]
    if not isinstance(modifiers, list):
        return ""
    for mod in modifiers:
        text = mod
        if isinstance(mod, dict):
            text = mod.get("text") or mod.get("name") or mod.get("label") or ""
        text = str(text or "").strip().lower()
        if not text:
            continue
        if (text.startswith("size:") or "make it large" in text
                or re.fullmatch(r"(?:large|small|personal|medium)(?:\s*\(.*\))?", text)
                or re.fullmatch(r"(?:12|16)\s*(?:-|\s)?\s*(?:inch|\")?(?:\s*round)?(?:\s*\(.*\))?", text)):
            key = _pizza_size_key(text)
            if key:
                return key
    return ""


def _is_whole_pizza(name, item_type=""):
    n = str(name or "").strip().lower()
    if not n or n.startswith("pizza "):
        return False
    is_gf = "gluten" in n or "cauliflower" in n   # "Cauliflower Crust Pizza" IS a pie
    # "thin-crust pizza" is a finished whole pie, not a pizza-crust product.
    # The old substring exclusion saw "crust" and suppressed the explicit
    # SMALL 12" line on exactly the Famous Buffalo/BBQ tickets.
    is_thin_crust_pizza = "thin-crust pizza" in n or "thin crust pizza" in n
    if not is_gf and not is_thin_crust_pizza and any(k in n for k in _PIZZA_NOT_A_WHOLE_PIE):
        return False
    if str(item_type or "").strip().lower() == "pizza":
        return True
    return bool(re.search(r"\bpizzas?\b|\bpies?\b", n))


def pizza_size_word(item):
    """'LARGE 16"' / 'SMALL 12"' / '16x16 SQUARE' / 'GF 9"' for a whole pie,
    '' for anything else or an unknown size (mirror of
    backend.pizza_variant.ticket_size_word)."""
    if not isinstance(item, dict):
        return ""
    name = str(item.get("name") or item.get("display_name") or "")
    if not _is_whole_pizza(name, item.get("type")):
        return ""
    nl = name.lower()
    if "gluten" in nl or "cauliflower" in nl:
        key = "gf"
    else:
        key = (_pizza_size_key(item.get("size"))
               or _pizza_name_size_key(name)
               or _pizza_size_from_modifiers(item.get("modifiers"))
               or ("square" if "grandma" in nl else ""))   # bare Grandma = the 16x16 row
    return _PIZZA_TICKET_SIZE_WORDS.get(key, "")


def _item_line_total(item):
    """Item line price for the ticket: line_total (qty-aware, includes paid
    modifiers), falling back to unit_total*qty then price*qty. 0 = unpriced."""
    for key, per_unit in (("line_total", False), ("unit_total", True), ("price", True)):
        try:
            v = float(item.get(key))
        except (TypeError, ValueError):
            continue
        if v > 0:
            if per_unit:
                try:
                    v *= float(item.get("quantity") or item.get("qty") or 1)
                except (TypeError, ValueError):
                    pass
            return v
    return 0.0

def _derive_ticket_money(o):
    """Classify total_price and return (food, fee_display, grand, shape).

    total_price is the GRAND total (food+tax+fee) for web/POS orders but
    food-only for phone orders; assuming food-only for everyone printed
    COLLECT $159.68 on the $150.34 web order #78 (2026-07-26). Local copy of
    backend/ticket_money.py's derive_ticket_money — keep the two in lockstep
    (backend/test_kitchen_ticket_money_shapes.py asserts parity). Only used
    when the server payload has no explicit food_subtotal/grand_total.
    Ambiguous payloads (unpriced items) keep the historical food-only math:
    a too-low COLLECT loses money silently. grand INCLUDES the tip.
    """
    def _f(key):
        try:
            return float(o.get(key) or 0)
        except Exception:
            return 0.0
    total, tax, fee, tip = _f("total_price"), _f("tax"), _f("delivery_fee"), _f("tip")
    # Card surcharge column (owner 9/24): stored like the tip — inside the
    # grand shapes' total_price, on top of the phone (food-only) shape.
    surcharge = max(_f("card_surcharge"), 0.0)
    # Garden Fund round-up (owner 2026-09-26): same stored shape as the
    # surcharge. Mirrors backend/ticket_money.py — keep in lockstep.
    surcharge = surcharge + max(_f("garden_fund"), 0.0)
    # Discount column (owner 9/26 auto BUY 3 GET 1 FREE): inside the grand
    # shapes' total_price, subtracted from the phone (food-only) shape; food
    # stays PRE-discount so Food − Discount + … still sums to COLLECT.
    discount = max(_f("discount"), 0.0)
    items = o.get("items") or o.get("order_items") or []
    items_sum = round(sum(_item_line_total(i) for i in items if isinstance(i, dict)), 2)
    charged_items = round(items_sum - discount, 2)
    is_delivery = "deliv" in str(o.get("order_type") or "").lower()

    shape, food, fee_display = "food_only", total, fee
    grand = round(total - discount + tax + fee + tip + surcharge, 2)
    if items_sum > 0.005 and abs(total - items_sum) > 0.02 and total > charged_items + 0.02:
        if abs(total - (charged_items + tax + fee + tip + surcharge)) <= 0.02:
            shape, food, grand = "grand_tip_included", round(total - tax - fee - tip - surcharge + discount, 2), total
        elif abs(total - (charged_items + tax + fee + surcharge)) <= 0.02:
            shape, food = "grand_tip_separate", round(total - tax - fee - surcharge + discount, 2)
            grand = round(total + tip, 2)
        elif fee <= 0.005 and is_delivery:
            # Fee folded into the total while the delivery_fee column is 0
            # (the order #78 family) — recover it so lines sum to COLLECT.
            hidden_tip_in = round(total - charged_items - tax - tip - surcharge, 2)
            hidden_tip_out = round(total - charged_items - tax - surcharge, 2)
            if tip > 0.005 and 0.02 < hidden_tip_in <= 25.0:
                shape = "grand_hidden_fee_tip_included"
                food, fee_display, grand = items_sum, hidden_tip_in, total
            elif 0.02 < hidden_tip_out <= 25.0:
                shape, food, fee_display = "grand_hidden_fee", items_sum, hidden_tip_out
                grand = round(total + tip, 2)
    # Subtotal-column tie-break (#2468, prod 2026-10-11) -- lockstep with
    # backend/ticket_money.py. When the items do not settle the shape (they
    # summed $17 against a $19 subtotal on order ...130) the subtotal COLUMN
    # does, but only when it reconciles to the cent: subtotal + tax + fee + tip
    # == total proves total_price is already the grand total. Without this the
    # BOOKS said $20.26 and the PAPER said $21.52 for the same ticket, because
    # this copy was never told. A subtotal that does not add up is ignored.
    if shape == "food_only":
        sub_v = _f("subtotal")
        if sub_v > 0.005:
            if abs(total - (sub_v + tax + fee + tip + surcharge)) <= 0.02:
                shape, food, fee_display, grand = "grand_subtotal_tip_included", sub_v, fee, total
            elif abs(total - (sub_v + tax + fee + surcharge)) <= 0.02:
                shape, food, fee_display = "grand_subtotal_tip_separate", sub_v, fee
                grand = round(total + tip, 2)
    # Tax-rate tie-break (P-13, 2026-11-01) - mirrors backend/ticket_money.py.
    if shape == "food_only" and tax >= 0.5 and total - tax > 0:
        if abs(round((total - tax - fee - tip - surcharge) * 0.06625, 2) - tax) <= 0.011 \
                and abs(round(total * 0.06625, 2) - tax) > 0.011:
            shape = "grand_tax_rate_reconciled"
            food = round(total - tax - fee - tip - surcharge + discount, 2)
            grand = total
    if food <= 0:
        shape, food, fee_display = "food_only", total, fee
        grand = round(total - discount + tax + fee + tip + surcharge, 2)
    if grand < 0:
        grand = 0.0
    return round(food, 2), round(fee_display, 2), round(grand, 2), shape

_SCHED_TAG_RE = re.compile(r"\[SCHEDULED for ([^\]]+)\]", re.IGNORECASE)
_SCHED_FREE_RE = re.compile(
    r"SCHEDULED[^\d\n]{0,40}?(\d{1,2}:\d{2}\s*[AP]\.?M\.?)", re.IGNORECASE)


def _scheduled_banner_text(o, otype=""):
    """Local-time text for a scheduled order's banner, or '' when not scheduled.

    Checks the three ways a schedule reaches a ticket (see build_ticket).
    Never raises — a malformed value just means no banner, never a lost
    ticket.
    """
    try:
        # 1. Explicit field (tracker's naive-UTC ISO, e.g. 2026-09-05T22:00:00).
        raw = str(o.get("scheduled_for") or o.get("scheduled_time") or "").strip()
        if raw and re.match(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}", raw):
            dt = datetime.datetime.strptime(raw[:16].replace("T", " "), "%Y-%m-%d %H:%M")
            dt = dt.replace(tzinfo=datetime.timezone.utc).astimezone()
            today = datetime.datetime.now(datetime.timezone.utc).astimezone().date()
            t = dt.strftime("%I:%M %p").lstrip("0")
            return t if dt.date() == today else f"{dt.strftime('%a %b %d')} {t}"
        si = str(o.get("special_instructions") or o.get("delivery_instructions") or "")
        # 2. record_order's stamp.
        m = _SCHED_TAG_RE.search(si)
        if m:
            return m.group(1).strip()
        # 3. Free-text note with a clock time.
        m = _SCHED_FREE_RE.search(si)
        if m:
            return m.group(1).upper().replace(".", "").replace("  ", " ").strip()
    except Exception:
        pass
    return ""


def build_ticket(o, include_specs=True):
    """Wawa-style kitchen ticket: big black item names, every modifier on
    its own marked line, accent colour only on lines the kitchen acts on
    (owner ruling 2026-07-28), ***FULFILMENT*** banner, instructions
    impossible to miss. Mirrors the layout language of
    backend/printer_client.py's Wawa ticket, scaled to 32 columns.

    include_specs=False drops the prep recipe lines (owner 2026-08-02:
    'MEASUREMENT PENDING' guidance on a customer's receipt looks bad) — used
    for the duplicate CUSTOMER/STORE/DRIVER copies; the kitchen ticket always
    keeps them. Kitchen recipe components use the same large, one-line visual
    hierarchy as the reference ticket in IMG_0763 (owner 2026-08-30); driver
    and customer copies remain unchanged."""
    b = bytearray()
    b += INIT + ALIGN_CTR + BOLD_ON + DOUBLE_ON
    b += enc("  ZORRO'S  ") + LF + DOUBLE_OFF + BOLD_OFF
    ticket = o.get("ticket_number") or o.get("id","??")
    src = (o.get("source","ONLINE")).upper()
    otype = (o.get("order_type") or "PICKUP").upper()
    # Ticket number BIG (owner 2026-07-26, order #62): the small one-line
    # "ORDER #D-12" was missed at the counter.
    b += BOLD_ON + DOUBLE_ON + enc(f"TICKET #{ticket}") + LF + DOUBLE_OFF + BOLD_OFF
    b += BOLD_ON + enc(f"[{src}]") + LF + BOLD_OFF
    b += enc(datetime.datetime.now().strftime("%m/%d %I:%M %p")) + LF
    # Owner 2026-08-03: precise RECEIVED (order intake, server UTC → local)
    # and PRINTED times on every ticket, so staff can time the make.
    recv_txt = ""
    try:
        raw_created = str(o.get("created_at") or "").strip()
        if raw_created:
            cdt = datetime.datetime.strptime(raw_created[:19], "%Y-%m-%d %H:%M:%S")
            cdt = cdt.replace(tzinfo=datetime.timezone.utc).astimezone()
            recv_txt = cdt.strftime("%I:%M %p").lstrip("0")
    except Exception:
        recv_txt = ""
    printed_txt = (datetime.datetime.now(datetime.timezone.utc)
                   .astimezone().strftime("%I:%M %p").lstrip("0"))
    if recv_txt:
        # "RECEIVED ...  PRINTED ..." was 33 columns and wrapped a stray
        # fragment onto its own line on a 32-column roll.
        b += BOLD_ON + enc(f"RECV {recv_txt}  PRINT {printed_txt}") + LF + BOLD_OFF
    else:
        b += BOLD_ON + enc(f"PRINTED {printed_txt}") + LF + BOLD_OFF
    b += enc("="*32) + LF + ALIGN_LEFT
    # ── WEB order: PICKUP or DELIVERY at the TOP (owner fix list 2026-10-15
    # #6). The fulfilment line (WEB PICKUP / WEB DELIVERY, owner 2026-09-11)
    # sits under the items; on a web ticket the first thing the line must
    # read is whether it goes out the door or waits at the counter.
    if src in ("WEB", "ONLINE", "WEBSITE"):
        b += (ALIGN_CTR + BOLD_ON + COLOR2_ON + DOUBLE_ON
              + enc("DELIVERY" if "DELIVER" in otype else "PICKUP") + LF
              + DOUBLE_OFF + COLOR2_OFF + BOLD_OFF + ALIGN_LEFT)
        b += enc("="*32) + LF
    # ── SCHEDULED banner (owner 2026-09-05 "fix this now") ─────────────────
    # A scheduled order must announce its time at the top of the ticket, or
    # the kitchen fires a 6 PM order at noon. D-6 (9/4), T-4 and D-5 (9/5)
    # all needed a hand-printed banner because nothing rendered the time.
    # Three ways a schedule reaches a ticket — check them all:
    #   1. a scheduled_for / scheduled_time field (naive-UTC ISO from the
    #      tracker → shown in local time);
    #   2. record_order's "[SCHEDULED for Fri Sep 5, 6:00 PM]" stamp;
    #   3. a free-text "SCHEDULED PICKUP 6:00 PM" note.
    sched_txt = _scheduled_banner_text(o, otype)
    if sched_txt:
        # Owner 2026-09-06: "enlarge and bold the line on the printed ticket
        # that shows the scheduled pickup or delivery time." Every line of
        # the block prints DOUBLE width+height, bold, wrapped at the 16
        # double-width columns — the fulfilment word ("PICKUP FOR" /
        # "DELIVERY FOR") and the time each get their own big line so the
        # kitchen can read them from the make station.
        b += ALIGN_CTR + BOLD_ON + DOUBLE_ON
        b += enc("** SCHEDULED **") + LF
        b += enc(f"{otype} FOR"[:16]) + LF
        # Day-form text ("Fri Sep 05, 7:00 PM") breaks at the comma, and the
        # clock time is one unbreakable token — never "7:00" / "PM" split.
        _pieces = [p.strip() for p in sched_txt.split(",") if p.strip()]
        for _piece in _pieces:
            _piece = re.sub(r"(\d{1,2}:\d{2})\s+([AP]M)", r"\1 \2", _piece)
            _words, _line = _piece.split(" "), ""
            for _w in _words:
                if _line and len(_line) + 1 + len(_w) > 16:
                    b += enc(_line.replace(" ", " ")) + LF
                    _line = _w
                else:
                    _line = (_line + " " + _w).strip() if _line else _w
            if _line:
                b += enc(_line.replace(" ", " ")) + LF
        b += DOUBLE_OFF + BOLD_OFF
        b += enc("="*32) + LF + ALIGN_LEFT
    items = _coerce_items(o.get("items") or o.get("order_items") or [])
    # Server payloads sometimes inject payment-banner pseudo-items like
    # "*** PAID (PAYMENT_LINK) ***" / "*** NOT PAID — COLLECT $31.99 ***".
    # They printed as "*HOT* 1x *** PAID ***" item lines (owner photos 8/3,
    # D-7 + T-31). We render real PAID/NOT PAID banners ourselves — drop them.
    def _is_payment_pseudo_item(it):
        n = str(it.get("name") or it.get("display_name") or "")
        nl = n.lower()
        return ("***" in n) or nl.startswith(("paid", "not paid")) or "collect $" in nl
    items = [it for it in items if isinstance(it, dict) and not _is_payment_pseudo_item(it)]
    # Internal build-guidance strings stored as plain modifiers (not "spec:"
    # prefixed) leaked onto customer/driver copies. Treat them as spec-class.
    _INTERNAL_MOD_MARKS = ("measurement pending", "current kitchen standard",
                           "build not verified", "staff confirm",
                           "follow current kitchen")
    for idx, item in enumerate(items):
        name = item.get("name") or item.get("display_name","Item")
        qty  = item.get("quantity") or item.get("qty",1)
        # HOT/COLD mark leads every item line (owner 2026-07-24): HOT in the
        # accent color on two-color paper, COLD inverse.
        temp = item_temperature(item)
        b += TALL_ON + BOLD_ON + temperature_tag(temp)
        # Per-item price on the right (owner 2026-07-26, order #62): the one
        # paper ticket must price every line like any other receipt. The
        # HOT/COLD tag occupies printable columns, so budget for it.
        # The WHAT-TO-MAKE line is the one the kitchen reads from a distance,
        # so it prints double-height and BLACK — the same hierarchy the POS
        # ticket uses (big black items, colour reserved for exceptions).
        # Everything around it stays normal size; if every line is the same
        # weight, none of them are.
        name_txt  = f"{qty}x {name.upper()}"
        price_txt = money(_item_line_total(item)) if _item_line_total(item) > 0 else ""
        # Owner 2026-09-05 ("kitchen prints are not visible for kitchen
        # staff need bold bigger letters"): the WHAT-TO-MAKE line was only
        # double-HEIGHT while the modifiers under it were already double
        # width+height — the main item was the least readable big element.
        # It now prints DOUBLE width AND height, bold, wrapped at the 16
        # double-width columns so a long name never truncates, and the
        # price drops to its own normal-size line so it can't crowd it.
        b += LF + DOUBLE_ON + BOLD_ON
        _half = TICKET_WIDTH // 2
        _words, _line = name_txt.split(" "), ""
        for _w in _words:
            if _line and len(_line) + 1 + len(_w) > _half:
                b += enc(_line) + LF
                _line = _w
            else:
                _line = (_line + " " + _w).strip() if _line else _w
        if _line:
            b += enc(_line) + LF
        # The pizza's SIZE WORD, same emphasis as the name (owner 2026-10-01:
        # D-48 printed "CHEESE PIZZA" for a pie the caller heard as 16-inch;
        # the cook made the 12"). Skipped only when the name already says it.
        _size_word = pizza_size_word(item)
        if _size_word and _size_word not in name_txt:
            b += enc(_size_word) + LF
        b += DOUBLE_OFF + BOLD_OFF + TALL_OFF
        if price_txt:
            b += enc(price_txt.rjust(TICKET_WIDTH)) + LF
        # Kitchen-only standard build (owner 2026-08-30, reference IMG_0763):
        # one ingredient per line, large enough to read from the make station.
        # Customer/store/driver copies pass include_specs=False and preserve
        # their existing clean layout. Customer removals are omitted from the
        # base build because the explicit large red NO line below is the only
        # instruction the cook should see for that ingredient.
        if include_specs:
            build = ticket_build_for(name, item)
            if build:
                removed = _removed_words(item.get("modifiers") or [])
                b += BOLD_ON + enc("MAKE:") + LF + BOLD_OFF
                for comp in build:
                    words = {w for w in re.findall(r"[a-z]+", comp.lower())}
                    if removed and any(words & r for r in removed):
                        continue
                    b += (DOUBLE_ON + BOLD_ON
                          + mod_block("", comp.upper(), MOD_WIDTH)
                          + BOLD_OFF + DOUBLE_OFF)
        # Half-and-half sides print as one grouped, bold block (LEFT:/RIGHT:)
        # so each side reads as a complete build. Removals print inverse,
        # add-ons in the accent color, "side of" lines get a pack checkbox
        # (owner 2026-07-24: highlighted for visibility).
        mods = item.get("modifiers") or []
        half_groups, other_mods = group_half_modifiers(mods)
        # Wawa discipline (owner 2026-08-07, T-23 photo): ONE ingredient per
        # line, no sentences.  A compound modifier like "NO veggies — steak &
        # mozzarella ONLY" wrapped over four double-width lines and the make
        # station couldn't read it.  Split on the dash connectors, drop a
        # clause whose words are already covered by the other modifier lines,
        # and strip trailing filler ("only", "please").
        _expanded = []
        _seen_words = set()
        for _m in other_mods:
            _seen_words.update(re.findall(r"[a-z]+", str(_m).lower()))
        for _m in other_mods:
            for _clause in re.split(r"\s+[—–-]{1,2}\s+", str(_m)):
                _clause = re.sub(r"\s+(?:only|please|pls)\s*$", "", _clause.strip(),
                                 flags=re.I).strip(" ,;")
                if not _clause:
                    continue
                _cw = set(re.findall(r"[a-z]+", _clause.lower()))
                _others = _seen_words - set(re.findall(r"[a-z]+", str(_m).lower())) \
                          | {w for x in _expanded for w in re.findall(r"[a-z]+", x.lower())}
                if _clause is not _m and _cw and _cw <= _others:
                    continue
                if _clause not in _expanded:
                    _expanded.append(_clause)
        other_mods = _expanded
        for side_label, toppings in half_groups.items():
            b += TALL_ON + BOLD_ON + mod_block("   ", side_label + ": " + " + ".join(toppings)) + BOLD_OFF + DOUBLE_OFF
        for mod in other_mods:
            m = str(mod); ml = m.lower()
            if (ml.startswith(("spec:", "build:"))
                    or ml == "build not verified - staff confirm"
                    or any(k in ml for k in _INTERNAL_MOD_MARKS)):
                # Defensive compatibility with orders saved by older servers:
                # internal recipe/spec diagnostics never belong on a live
                # human ticket. The _INTERNAL_MOD_MARKS net also catches
                # un-prefixed guidance strings ("MEASUREMENT PENDING") that
                # leaked onto customer/driver copies (owner photos 8/3).
                continue
            # Owner ruling 2026-07-28: the accent colour marks ONLY the lines
            # the kitchen has to act on differently — removals, sides
            # (dressing/sauce) and extras. Everything else stays black so the
            # marked lines are the ones that catch the eye.
            # Owner 2026-08-06 (POS ticket photo IMG_0591): modifiers on the
            # Shift4 ticket print as large as the item lines — the make station
            # reads them from the same distance.  Ours printed at normal size
            # and got lost.  Double width AND height (DOUBLE_ON) at MOD_WIDTH
            # so nothing truncates ("ITALIAN SOFT ROL" on that very photo).
            if ml.startswith("no ") or ml.startswith("without "):
                b += (DOUBLE_ON + BOLD_ON + COLOR2_ON
                      + mod_block("! ", m.upper(), MOD_WIDTH)
                      + COLOR2_OFF + BOLD_OFF + DOUBLE_OFF)
            elif ml.startswith("side of ") or ml.startswith("side "):
                # The [] glyph already says "side" — repeating it burns 8 of
                # the 16 double-width columns and forces a needless wrap.
                txt = m.upper()
                for lead in ("SIDE OF ", "SIDE "):
                    if txt.startswith(lead):
                        txt = txt[len(lead):]
                        break
                b += DOUBLE_ON + BOLD_ON + mod_block("[]", txt, MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
            elif ml.startswith("extra ") or ml.startswith("add "):
                # Same for "ADD" against the + glyph.  "EXTRA" stays: it means
                # a double portion of something already on the build, which is
                # not the same instruction as adding it.
                txt = m.upper()
                if txt.startswith("ADD "):
                    txt = txt[4:]
                b += DOUBLE_ON + BOLD_ON + mod_block("+ ", txt, MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
            elif ml.startswith("light ") or ml.startswith("easy "):
                b += DOUBLE_ON + BOLD_ON + mod_block("~ ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
            elif ml.startswith("sub ") or ml.startswith("swap "):
                b += DOUBLE_ON + BOLD_ON + mod_block("> ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
            else:
                b += DOUBLE_ON + BOLD_ON + mod_block("+ ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
        # Per-item special instructions (owner order #78: "SUB SPINACH for
        # lettuce" was stored on the item but the kitchen never saw it —
        # only order-level notes printed). Same key precedence as the POS
        # path (pos_submission._pos_item_for). Skip if a modifier already
        # says the same thing.
        special = str(item.get("special_instructions")
                      or item.get("notes") or "").strip()
        if special and not _note_already_said(special, other_mods):
            b += DOUBLE_ON + BOLD_ON + COLOR2_ON + mod_block(">>", special.upper(), MOD_WIDTH)
            b += COLOR2_OFF + BOLD_OFF + DOUBLE_OFF
        if idx < len(items) - 1:
            b += enc("- " * (TICKET_WIDTH // 2)) + LF
    if include_specs:
        _packing = _packing_instructions(o, items)
        if _packing:
            b += enc("!"*32) + LF
            b += ALIGN_CTR + INVERSE_ON + BOLD_ON + DOUBLE_ON
            b += enc(" PACKAGING ") + LF
            b += DOUBLE_OFF + BOLD_OFF + INVERSE_OFF + ALIGN_LEFT
            _has_missing = False
            for _instruction in _packing:
                if isinstance(_instruction, dict):
                    _qty = _instruction.get("quantity")
                    _code = str(_instruction.get("code") or "").strip().upper()
                    _label = str(_instruction.get("instruction") or _instruction.get("label") or "").strip().upper()
                    _status = str(_instruction.get("status") or _instruction.get("inventory_status") or "MEASURED").strip().upper()
                    _has_missing = _has_missing or _status == "MISSING"
                    _qty_text = "" if _qty is None else f"{float(_qty):g}x"
                    _text = " ".join(p for p in (_status, _qty_text, _code, _label) if p)
                else:
                    _text = str(_instruction or "").strip().upper()
                if _text:
                    b += DOUBLE_ON + BOLD_ON + mod_block("[]", _text, MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF
            _footer = "MANAGER CHECK EVERY MISSING LINE" if _has_missing else "PACK EVERY LINE"
            b += BOLD_ON + enc(_footer) + LF + BOLD_OFF
    b += enc("="*32) + LF
    # Fulfilment banner, styled like the POS ticket's ***TO GO*** — accent
    # colour, double height, starred so it is unmissable at a glance. Whether
    # this goes out the door or waits at the counter is the single fact staff
    # most need off the paper.
    # Owner 2026-09-11: web orders say WEB PICKUP / WEB DELIVERY in the
    # banner itself, same tall bold accent style — the small [WEB] tag by the
    # ticket number was not enough for the cooks to tell the channel apart.
    banner = (f"*** WEB {otype} ***" if src in ("WEB", "ONLINE", "WEBSITE")
              else f"***{otype}***")
    b += (ALIGN_CTR + BOLD_ON + COLOR2_ON + TALL_ON
          + enc(banner) + LF
          + TALL_OFF + COLOR2_OFF + BOLD_OFF + ALIGN_LEFT)
    # The name is how the counter matches ticket to customer at handover, so
    # it carries the accent colour too (owner ruling 2026-07-28).
    if o.get("customer_name"):
        b += (BOLD_ON + COLOR2_ON + enc(f"Name:  {o['customer_name']}")
              + COLOR2_OFF + BOLD_OFF + LF)
    if o.get("customer_phone"): b += enc(f"Phone: {o['customer_phone']}") + LF
    if o.get("address") and "DELIVERY" in otype:
        # Delivery address tall + bold, wrapped with continuation indent so a
        # long street name never hard-wraps mid-word at the printer.
        b += TALL_ON + BOLD_ON + mod_block("ADDR:  ", str(o["address"])) + BOLD_OFF + DOUBLE_OFF
    if o.get("payment_method"):
        pay = str(o.get("payment_method") or "").upper()
        status = str(o.get("payment_status") or "").upper()
        b += enc(f"Pay:   {pay}{(' / ' + status) if status else ''}") + LF
    def _f(key):
        try:
            return float(o.get(key) or 0)
        except Exception:
            return 0.0

    tax_v, tip_v = _f("tax"), _f("tip")
    # Prefer the server's explicit money fields (grand_total INCLUDES tip);
    # older servers only send raw columns, where total_price means "grand"
    # for web/POS orders but "food" for phone orders — derive locally then.
    # Naively summing the columns double-counted tax+fee on web orders
    # (order #78 printed COLLECT $159.68 for a $150.34 order).
    food, collect = _f("food_subtotal"), _f("grand_total")
    if collect > 0:
        fee_v = _f("delivery_fee_display") or _f("delivery_fee")
    else:
        food, fee_v, collect, _shape = _derive_ticket_money(o)
    if food > 0:
        b += enc(f"Food:  ${food:.2f}") + LF
    # Owner 2026-09-26: automatic BUY 3 GET 1 FREE — Food is PRE-discount, so
    # the promo prints as its own line and the lines still sum to COLLECT.
    disc_v = _f("discount")
    if disc_v > 0:
        promo_lines = []
        try:
            raw_promos = o.get("auto_promos")
            if isinstance(raw_promos, str):
                raw_promos = json.loads(raw_promos or "[]")
            for pl in raw_promos or []:
                if isinstance(pl, dict) and float(pl.get("amount") or 0) > 0:
                    promo_lines.append((str(pl.get("label") or "PROMO"), float(pl["amount"])))
        except Exception:
            promo_lines = []
        named = round(sum(a for _, a in promo_lines), 2)
        for label, amt in promo_lines:
            b += enc(f"{label[:22]}: -${amt:.2f}") + LF
        if disc_v - named > 0.005:
            b += enc(f"Discount: -${disc_v - named:.2f}") + LF
    if tax_v > 0:
        b += enc(f"Tax:   ${tax_v:.2f}") + LF
    if fee_v > 0:
        b += enc(f"Deliv: ${fee_v:.2f}") + LF
    if tip_v > 0:
        b += BOLD_ON + enc(f"TIP:   ${tip_v:.2f}") + LF + BOLD_OFF
    # Card processing fee (owner 2026-09-24: passed to the customer when the
    # customer chooses card) — its own line on the customer copy so the
    # receipt discloses it; 0 on every cash ticket, so the line is absent.
    cfee_v = _f("card_surcharge")
    if cfee_v > 0:
        b += enc(f"Card fee: ${cfee_v:.2f}") + LF
    # Zorro's Garden Fund round-up (owner 2026-09-26): thanked on the
    # customer copy; a chosen contribution, never a fee.
    gfund_v = _f("garden_fund")
    if gfund_v > 0:
        b += enc(f"Garden Fund: ${gfund_v:.2f} - thank you!") + LF
    # PAID / NOT PAID banner (owner 2026-07-26, order #62): the single paper
    # ticket must say unmistakably whether money is still owed. An order with
    # no payment_status at all (older payloads) keeps the legacy COLLECT line.
    status_l = str(o.get("payment_status") or "").strip().lower()
    if status_l == "paid":
        # Owner 2026-08-03: when paid by card, the ticket must state the exact
        # amount charged and the tip on its own loud line — drivers/cashiers
        # were calling to ask what was collected and whether the tip is theirs.
        paid_total = collect if collect > 0 else round(food + tax_v + fee_v + tip_v, 2)
        b += ALIGN_CTR + BOLD_ON + DOUBLE_ON + enc(" ** PAID ** ") + LF
        b += DOUBLE_OFF + BOLD_OFF
        if paid_total > 0:
            b += BOLD_ON + enc(f"CARD TOTAL ${paid_total:.2f}") + LF + BOLD_OFF
        if tip_v > 0:
            b += BOLD_ON + COLOR2_ON + TALL_ON
            b += enc(f"TIP INCLUDED ${tip_v:.2f}") + LF
            b += TALL_OFF + COLOR2_OFF + BOLD_OFF
        b += enc("COLLECT NOTHING") + LF + ALIGN_LEFT
    elif status_l and collect > 0:
        b += ALIGN_CTR + INVERSE_ON + BOLD_ON + DOUBLE_ON
        b += enc("  NOT PAID  ") + LF
        b += enc(f"COLLECT ${collect:.2f}") + LF
        b += DOUBLE_OFF + BOLD_OFF + INVERSE_OFF + ALIGN_LEFT
    elif collect > 0:
        b += BOLD_ON + DOUBLE_ON + enc(f"COLLECT ${collect:.2f}") + LF + DOUBLE_OFF + BOLD_OFF
    # Owner 2026-08-03: timer targets on the kitchen ticket. Until the owner
    # sets standard quote minutes we print write-in boxes — never invent an
    # ETA the customer wasn't promised. Kitchen tickets only (include_specs);
    # customer/driver copies skip the internal timer boxes.
    if include_specs:
        b += enc("-"*32) + LF
        if "DELIVERY" in otype:
            b += TALL_ON + BOLD_ON
            b += enc("OUT THE DOOR BY: ______") + LF
            b += enc("AT CUSTOMER BY:  ______") + LF
            b += BOLD_OFF + TALL_OFF
        else:
            b += TALL_ON + BOLD_ON
            b += enc("READY BY: ______") + LF
            b += BOLD_OFF + TALL_OFF
            b += enc("[ ] waiting at counter") + LF
    # Order-level notes get a full SPECIAL INSTRUCTIONS section (Wawa-style)
    # instead of one small NOTE line that hard-wrapped past 32 columns.
    # Internal staff annotations (review-hold reasons etc.) must NEVER reach
    # a customer's hands: Anna's CUSTOMER COPY printed "UNCONFIRMED - CALL
    # ENDED EARLY - VERIFY WITH CUSTOMER" (owner photo 8/3). Kitchen tickets
    # keep them; customer/store/driver copies drop them.
    _visible_notes = _ticket_notes_for_surface(o, include_specs)
    if _visible_notes:
        b += enc("-"*32) + LF
        b += ALIGN_CTR + INVERSE_ON + BOLD_ON + enc(" SPECIAL INSTRUCTIONS ") + LF
        b += BOLD_OFF + INVERSE_OFF + ALIGN_LEFT
        b += TALL_ON + BOLD_ON
        for line in wrap_lines(_visible_notes.upper(), TICKET_WIDTH - 1):
            b += enc(" " + line) + LF
        b += BOLD_OFF + DOUBLE_OFF
    b += enc(" "*32) + LF + CUT
    return bytes(b)

def build_test_order():
    return {
        "id": "TEST",
        "ticket_number": "TEST",
        "source": "SETUP",
        "order_type": "pickup",
        "customer_name": "Printer Test",
        "customer_phone": "",
        "notes": "If you can read this, the kitchen printer is connected.",
        "items": [
            {
                "name": "TEST PRINT - ZORRO'S KITCHEN",
                "quantity": 1,
                "modifiers": [
                    "Local printer path works",
                    "Phone orders will print here",
                    "Keep this computer running",
                ],
            }
        ],
    }

def _send_raw(ip, port, data, timeout=5):
    with socket.create_connection((ip, port), timeout=timeout) as s:
        s.sendall(data)


class PartialSend(OSError):
    """The connection died MID-TICKET: `sent` of `total` bytes were handed to
    the printer before the error, so a partial ticket may be on paper."""

    def __init__(self, sent, total, cause):
        OSError.__init__(self, "partial send: %d/%d bytes delivered before %s"
                         % (sent, total, cause or "the link dropped"))
        self.sent = sent
        self.total = total
        self.cause = cause


def _link_key(ip, port):
    return (str(ip), int(port))


def _note_link_up(ip, port):
    """A reachability probe (or a fully delivered ticket) found the link up."""
    key = _link_key(ip, port)
    prev = _link_state.get(key)
    now = _clock()
    _link_state[key] = {"up": True, "at": now, "reason": ""}
    if prev and not prev.get("up") and key in _wait_logged_at:
        log(f"[PRINTER LINK] {ip}:{port} back after {int(now - prev['at'])}s "
            f"down ({prev.get('reason')}) — waiting tickets print now")
    _wait_logged_at.pop(key, None)
    _wait_reported.clear()


def _note_link_down(ip, port, reason):
    """A probe failed, or a ticket died mid-send: nothing goes to this
    printer until a probe succeeds again. Keeps the original 'down since'."""
    key = _link_key(ip, port)
    prev = _link_state.get(key)
    since = prev["at"] if prev and not prev.get("up") else _clock()
    _link_state[key] = {"up": False, "at": since, "reason": str(reason)[:120]}


def _link_down_reason(ip=None, port=None):
    """Why the printer must NOT be sent to right now, else None.

    Only a probe result counts: None when the last probe succeeded or when
    nothing in this process has probed the printer yet (a hand-print script
    keeps its legacy send-and-see behaviour)."""
    st = _link_state.get(_link_key(ip or PRINTER_IP, port or PRINTER_PORT))
    if st and not st.get("up"):
        return st.get("reason") or "unreachable"
    return None


def probe_printer_link(ip=None, port=None):
    """Fresh reachability probe, run right before every ticket.

    Rides check_printer_fault (one connection: TCP connect + DLE EOT, which
    records the link state and the status-capable flag) so the probe and
    the paper/cover gate cost one socket, not two. With PRINTER_STATUS_CHECK
    off the probe is a bare TCP connect — still a reachability probe.

    Returns (down_reason, fault): down_reason is None when the link is up,
    fault is check_printer_fault's answer (None when healthy or unknown)."""
    ip = ip or PRINTER_IP
    port = port or PRINTER_PORT
    if STATUS_CHECK_ENABLED:
        fault = check_printer_fault(ip, port)
        return _link_down_reason(ip, port), fault
    try:
        with socket.create_connection((ip, port), timeout=PROBE_TIMEOUT) as s:
            s.close()
    except Exception as e:
        reason = str(e) or type(e).__name__
        _note_link_down(ip, port, reason)
        return reason, None
    _note_link_up(ip, port)
    return None, None


def _link_gate(ip, port):
    """The reason a send must NOT start now, else None.

    A send needs a probe that SUCCEEDED within PROBE_FRESH_SECONDS. A probe
    that failed, or one older than that, is re-run here — a link that was
    up five polls ago says nothing about the next 8 KB. No entry at all
    (never probed in this process: --test-print, a hand-print script) keeps
    the legacy send-and-see path."""
    st = _link_state.get(_link_key(ip, port))
    if st is None:
        return None
    if st.get("up") and _clock() - st["at"] <= PROBE_FRESH_SECONDS:
        return None
    down, _fault = probe_printer_link(ip, port)
    return down


def _log_link_wait(reason, waiting, what="tickets", ip=None, port=None):
    """One '[PRINTER WAIT]' line per WAIT_LOG_SECONDS while tickets wait for
    the link — the 9/26 log had 139 per-attempt error lines for the same
    fact. Returns True when a line was written."""
    ip = ip or PRINTER_IP
    port = port or PRINTER_PORT
    key = _link_key(ip, port)
    now = _clock()
    last = _wait_logged_at.get(key)
    if last is not None and now - last < WAIT_LOG_SECONDS:
        return False
    _wait_logged_at[key] = now
    st = _link_state.get(key) or {}
    down_for = int(now - st["at"]) if st.get("at") is not None else 0
    log(f"[PRINTER WAIT] link down at {ip}:{port} for {down_for}s "
        f"({reason}) — {waiting} {what} waiting, nothing sent, nothing marked; "
        f"probing every poll, next line in {int(WAIT_LOG_SECONDS)}s")
    return True


def _hold_for_link(down, waiting_ids, what="tickets"):
    """The link is down: hold the batch. One log line a minute; each waiting
    order is reported to the print-job ledger ONCE per outage (the stuck
    report names it) — never once per poll like the 9/26 flood."""
    _log_link_wait(down, len(waiting_ids), what)
    for oid in waiting_ids:
        if oid in _wait_reported:
            continue
        _wait_reported.add(oid)
        report_print_failed(
            oid, f"waiting: printer link down at {PRINTER_IP}:{PRINTER_PORT} "
                 f"({down}) — nothing sent; prints when the link is back")


def _send_timeout_for(nbytes):
    """Socket timeout scaled to the ticket: base + per-KB, capped."""
    t = SEND_TIMEOUT_BASE + SEND_TIMEOUT_PER_KB * (nbytes / 1024.0)
    return max(SEND_TIMEOUT_BASE, min(SEND_TIMEOUT_MAX, t))


def _sendall_tracked(sock, data):
    """sendall() that knows how far it got.

    One socket.sendall() of the whole ticket raises with no word on how many
    bytes the kernel already pushed down the link. Sending SEND_CHUNK-sized
    pieces keeps the count to within one chunk: a mid-ticket error is raised
    as PartialSend(sent, total) when anything left the Mac (a partial page
    may be on the rail), and re-raised as the plain error when nothing did
    (an ordinary retry — no paper to worry about). Returns the byte count."""
    total = len(data)
    sent = 0
    while sent < total:
        chunk = data[sent:sent + SEND_CHUNK]
        try:
            sock.sendall(chunk)
        except (OSError, socket.timeout) as e:
            if sent:
                raise PartialSend(sent, total, e)
            raise
        sent += len(chunk)
    return sent


# ── Partial-send bookkeeping (orders and bridge jobs share it) ───────────────

def _set_last_partial(sent, total, why):
    """Record the partial send the caller (the polling loop) books next."""
    global _last_partial_send, _last_print_failure
    _last_partial_send = (int(sent), int(total))
    _last_print_failure = why


def _partial_retry_due(key):
    """False while a partial-sent ticket is still backing off."""
    not_before = _retry_not_before.get(key)
    return not_before is None or _clock() >= not_before


def _ticket_for_send(key, data):
    """The bytes to send for `key`: the POSSIBLE DUPLICATE banner goes on top
    of the one safe retry of an unconfirmed send (the whole ticket may be on
    the rail); else the REPRINT banner whenever a previous attempt put PART
    of this ticket on paper."""
    if _unconfirmed_sends.get(key):
        return _duplicate_banner() + data
    if _partial_sends.get(key):
        return _reprint_banner() + data
    return data


def _clear_partial(key):
    _partial_sends.pop(key, None)
    _retry_not_before.pop(key, None)
    _partial_paged.discard(key)


def _after_partial_send(key, label, order_id=None, ticket=""):
    """Book a ticket that died mid-send (print_ticket left _last_partial_send).

    Logs the one line that names the fact, schedules the retry with a
    30 → 60 → 120 … s backoff capped at PARTIAL_BACKOFF_MAX (never the next
    10 s poll), and — after PARTIAL_PAGE_AFTER partials of the SAME ticket —
    pages the owner ONCE through the existing alert path: the print-failed
    report (the ledger's FAILED text reaches the owner SMS) plus the fault
    heartbeat. Returns the retry delay in seconds."""
    sent, total = _last_partial_send or (0, 0)
    n = _partial_sends.get(key, 0) + 1
    _partial_sends[key] = n
    delay = min(PARTIAL_BACKOFF_MAX, PARTIAL_BACKOFF_BASE * (2 ** (n - 1)))
    _retry_not_before[key] = _clock() + delay
    who = f"{label} (ticket {ticket})" if ticket else label
    log(f"[PARTIAL SEND] {label} bytes={sent}/{total}{' ticket ' + ticket if ticket else ''} "
        f"— a partial page may be on the rail; re-queued under the REPRINT "
        f"banner, retry in {int(delay)}s (partial #{n})")
    if n >= PARTIAL_PAGE_AFTER and key not in _partial_paged:
        _partial_paged.add(key)
        reason = (f"printer link unstable at {PRINTER_IP}:{PRINTER_PORT} — "
                  f"{who} died mid-send {n}x ({sent}/{total} bytes last time); "
                  f"held, retrying every {int(delay)}s with a REPRINT banner")
        if order_id is not None:
            report_print_failed(order_id, reason)
        _flag_printer_fault(reason)
        log(f"[PAGE] owner paged once for {label}: printer link unstable at "
            f"{PRINTER_IP}:{PRINTER_PORT}")
    return delay


# Copies (PACK / DRIVER / PAID) that died mid-send. A copy is best-effort
# and prints AFTER the order is marked, so the server never re-serves it —
# the retry lives here: {(oid, label): {"data", "to", "label"}} plus the
# same backoff/partial counters keyed by (oid, label). Retried on the SAME
# printer under the REPRINT banner; given up after PARTIAL_PAGE_AFTER tries
# (the POS can reprint a copy on demand — the line must not get a pile).
_copy_retries = {}


COPY_RETRY_MAX_TRIES = 10


def _queue_copy_retry(oid, label, data, to):
    key = (oid, label)
    _copy_retries[key] = {"data": bytes(data), "to": to, "tries": 0}
    _after_partial_send(key, f"{label} #{oid}")


def _drain_copy_retries():
    """Retry queued copies whose backoff has passed. Runs once per poll.

    A retry that dies mid-send again backs off like a ticket; one refused by
    the link gate (nothing sent) simply waits for the next poll. Either way
    the copy is dropped — loudly — after PARTIAL_PAGE_AFTER partials or
    COPY_RETRY_MAX_TRIES attempts: the POS reprints a copy on demand, and
    the line must never get a pile."""
    for key in list(_copy_retries):
        if not _partial_retry_due(key):
            continue
        entry = _copy_retries[key]
        oid, label = key
        data = _reprint_banner() + entry["data"]
        if entry["to"] == "receipt":
            ok = print_receipt_copy(data, label=f"{label} #{oid} (reprint)")
        else:
            ok = print_ticket(data)
        if ok:
            log(f"[COPY] {label} #{oid} reprinted whole after a partial send")
            _copy_retries.pop(key, None)
            _clear_partial(key)
            continue
        entry["tries"] += 1
        if _last_partial_send:
            _after_partial_send(key, f"{label} #{oid}")
        if (_partial_sends.get(key, 0) >= PARTIAL_PAGE_AFTER
                or entry["tries"] >= COPY_RETRY_MAX_TRIES):
            log(f"[COPY ERROR] {label} #{oid} not reprinted after a partial send "
                f"({_last_print_failure or 'send failed'}) — reprint it from the POS")
            _copy_retries.pop(key, None)
            _clear_partial(key)


def _note_status_capable(ip, port, *replies):
    if any(_status_identifies(b) for b in replies):
        _status_capable.add((ip, port))

# ── POST-SEND VERDICT (owner 10/5, Noe: "it printed normally") ───────────────
# 10/5: every one of the day's 40 tickets printed, the Epson answered DLE EOT
# 0x12 (online, no fault bits) right after every send — and every order row
# still read "UNVERIFIED: real-time status is not a completion barrier". The
# audit-9/26 wording treated the status sample as nothing because it is not
# the GS ( H completion echo. Ruling: a ticket whose bytes all left this Mac
# AND whose printer then answered ONLINE on the same connection is PRINT
# CONFIRMED (printed_verified 1). The completion echo stays the stronger
# proof and keeps its own wording. "UNVERIFIED" is reserved for the cases
# that really prove nothing — no status answer (link lost / mute printer /
# check disabled), a byte that is not an ESC/POS status byte — always WITH
# the reason; a status byte with fault bits is an ERROR (never marked).
POST_SEND_CONFIRMED = "confirmed"
POST_SEND_UNVERIFIED = "unverified"
POST_SEND_ERROR = "error"
ONLINE_EVIDENCE = "printed — printer answered online after send"


def classify_post_send(offline, paper, status_capable):
    """(verdict, reason) for what the printer said right after the ticket.

    ``offline``/``paper`` are the DLE EOT n=2 / n=4 reply bytes (None on
    silence); ``status_capable`` says whether this printer has ever answered
    DLE EOT in this process. Pure — no sockets, no globals."""
    if offline is None and paper is None:
        if status_capable:
            return POST_SEND_UNVERIFIED, ("no status reply after send — link lost, "
                                          "delivery unconfirmed")
        return POST_SEND_UNVERIFIED, ("no status reply after send (printer never "
                                      "answers DLE EOT) — no physical acknowledgement")
    if not _status_identifies(offline) and not _status_identifies(paper):
        shown = "0x%02x" % offline if offline is not None else "0x%02x" % paper
        return POST_SEND_UNVERIFIED, (f"status reply {shown} is not an ESC/POS status "
                                      f"byte — printer condition unknown")
    fault = _interpret_status(offline, paper)
    if fault:
        return POST_SEND_ERROR, f"printer reported '{fault}' right after send"
    byte = offline if _status_identifies(offline) else paper
    return POST_SEND_CONFIRMED, f"{ONLINE_EVIDENCE} (DLE EOT 0x{byte:02x})"


def _post_send_status_sample(sock, ip, port):
    """Ask the printer, on the SAME socket and after the ticket bytes, how it
    is — and turn the answer into the ticket's verdict (classify_post_send).

    An identifying status byte with no fault bits means the whole ticket
    reached the printer (the connection outlived the last byte) and the
    printer is online to print it: PRINT CONFIRMED. Silence from a printer
    that is known to answer DLE EOT means the link likely died mid-stream —
    the order-#59 hole — so that case fails the print into the bounded
    sent-unconfirmed retry instead of being marked. Silence from a printer
    that never answers is marked, but UNVERIFIED with the reason.

    Returns None to proceed with mark-printed, else a short failure reason.
    """
    global _last_send_unconfirmed, _last_print_evidence, _last_print_confirmed_online
    offline = _query_status_byte(sock, STATUS_OFFLINE, timeout=SEND_STATUS_TIMEOUT)
    paper = _query_status_byte(sock, STATUS_PAPER) if offline is not None else None
    _note_status_capable(ip, port, offline, paper)
    verdict, reason = classify_post_send(offline, paper, (ip, port) in _status_capable)
    if verdict == POST_SEND_ERROR:
        return f"{reason} — ticket may not have printed"
    if verdict == POST_SEND_CONFIRMED:
        _last_print_confirmed_online = True
        _last_print_evidence = f"{reason} at {ip}:{port}"
        return None
    if offline is None and (ip, port) in _status_capable:
        _last_send_unconfirmed = True
        return reason
    # Never seen this printer speak DLE EOT (or it answered noise): nothing
    # proves paper. Keep legacy send-and-mark rather than reprint forever,
    # but say so once per ticket — and on the order row, with the reason.
    log(f"[SEND CHECK WARN] {ip}:{port}: {reason}; delivery unverified")
    _last_print_evidence = f"sent to {ip}:{port}; {reason} — UNVERIFIED"
    return None

def _next_completion_id():
    global _completion_seq
    _completion_seq = (_completion_seq + 1) % 10000
    return "{:04d}".format(_completion_seq).encode("ascii")

def _confirm_completion(sock, ident, timeout=None):
    """Wait for the GS ( H process-ID echo: a TRUE job-completion signal.

    The request must already have been sent after the ticket bytes. Because
    GS ( H is an ordinary buffered command, the printer replies only after all
    preceding data has been processed — so seeing our ID back means the ticket
    was consumed. Scans a small window for 0x37 <ident> 0x00 to tolerate an
    interleaved ASB/status byte. Returns True on confirmation.
    """
    if timeout is None:
        timeout = COMPLETION_TIMEOUT
    expected = COMPLETION_RSP_HEADER + ident
    buf = b""
    try:
        sock.settimeout(timeout)
        while len(buf) < 64:
            chunk = sock.recv(16)
            if not chunk:
                return False
            buf += chunk
            if expected in buf:
                return True
    except (OSError, socket.timeout):
        return False
    return False

def _send_ticket(ip, port, data, timeout=None):
    """Send a ticket on one connection, then check on it as well as we can.

    With PRINT_COMPLETION_CHECK=1 (requires --probe-completion certification):
    appends a GS ( H process-ID request and waits for the echo — genuine
    confirmation the ticket was processed. Otherwise falls back to the DLE EOT
    status sample (link survival + status, not proof of printing).

    The socket timeout defaults to one scaled to the ticket length
    (_send_timeout_for): the flat 5 s fired mid-ticket on 9/26. Bytes go out
    through _sendall_tracked, so a connection that dies mid-ticket raises
    PartialSend(sent, total) — the caller knows a partial page may be on the
    rail — while a failure before the first byte raises the plain error.

    Returns None on success, else a failure-reason string. Raises on
    connect/send errors like _send_raw did.
    """
    global _last_send_unconfirmed, _last_print_completed, _last_print_evidence
    global _last_print_printer
    if timeout is None:
        timeout = _send_timeout_for(len(data))
    _last_print_printer = "%s:%d" % (ip, int(port))
    with socket.create_connection((ip, port), timeout=timeout) as s:
        if _completion_enabled_for(ip, port):
            ident = _next_completion_id()
            t0 = time.time()
            _sendall_tracked(s, data + COMPLETION_REQ_PREFIX + ident)
            if _confirm_completion(s, ident):
                _last_print_completed = True
                _last_print_evidence = (
                    "completion echo GS ( H id %s from %s:%s in %d ms — the printer "
                    "processed every byte before it" % (
                        ident.decode("ascii"), ip, port, int((time.time() - t0) * 1000)))
                return None
            _last_send_unconfirmed = True
            return ("no completion echo from %s:%s within %ss — printing unconfirmed"
                    % (ip, port, int(COMPLETION_TIMEOUT)))
        _sendall_tracked(s, data)
        if not SEND_STATUS_CHECK_ENABLED:
            _last_print_evidence = (f"sent to {ip}:{port}; post-send check disabled "
                                    f"(PRINT_SEND_STATUS_CHECK=0) — UNVERIFIED")
            return None
        return _post_send_status_sample(s, ip, port)

def print_ticket(data):
    global _last_print_failure, _last_send_unconfirmed, _last_print_completed
    global _last_partial_send, _last_print_evidence, _last_print_confirmed_online
    _last_print_failure = ""
    _last_send_unconfirmed = False
    _last_print_completed = False
    _last_print_confirmed_online = False
    _last_print_evidence = ""
    _last_partial_send = None
    # Kitchen path only — the receipt-station Epson prints black already.
    data = _apply_darkness_boost(data)
    # GATE (D-22, 9/26): not one byte onto a link the last probe found down.
    # No log line per attempt — the polling loop says "[PRINTER WAIT]" once
    # a minute for the whole batch; the backup printer (if any) still gets
    # its chance below, exactly as when the primary connect itself fails.
    down = _link_gate(PRINTER_IP, PRINTER_PORT)
    if down:
        why = f"waiting: printer link down at {PRINTER_IP}:{PRINTER_PORT} ({down}) — nothing sent"
    else:
        try:
            why = _send_ticket(PRINTER_IP, PRINTER_PORT, data)
            if why is None:
                _note_link_up(PRINTER_IP, PRINTER_PORT)
                return True
            log(f"[PRINTER ERROR] {why}")
        except PartialSend as e:
            # PART of the ticket is on paper. Never a backup-printer fallback
            # (a second full copy beside a half page is the D-22 confusion
            # in reverse); the loop re-queues it under a REPRINT banner and
            # writes the one "[PARTIAL SEND]" line that names the order.
            _set_last_partial(e.sent, e.total, str(e))
            _note_link_down(PRINTER_IP, PRINTER_PORT, "partial send: %s" % (e.cause or "link dropped mid-ticket"))
            log(f"[PRINTER ERROR] {e}")
            return False
        except Exception as e:
            why = str(e)
            log(f"[PRINTER ERROR] {e}")
    _last_print_failure = why
    # Primary unreachable — fall back to the backup printer if one is configured.
    # Only ONE of the two ever prints a given ticket in a poll: we return True as
    # soon as either succeeds, so mark-printed acks it and it won't reprint.
    if BACKUP_PRINTER_IP:
        try:
            why = _send_ticket(BACKUP_PRINTER_IP, BACKUP_PRINTER_PORT, data)
            if why is None:
                _last_print_failure = ""
                log(f"[BACKUP OK] Printed on backup {BACKUP_PRINTER_IP}:{BACKUP_PRINTER_PORT} (primary down)")
                return True
            log(f"[BACKUP PRINTER ERROR] {why}")
        except Exception as e:
            why = str(e)
            log(f"[BACKUP PRINTER ERROR] {e}")
        _last_print_failure = f"{_last_print_failure}; backup: {why}"
    return False

def print_receipt_copy(data, label=""):
    """Print a duplicate receipt on the receipt-station printer when configured.

    Falls back to the normal kitchen path (print_ticket, incl. its backup
    chain) if no receipt printer is set or it fails — copies are best-effort
    everywhere, but the fallback keeps them flowing when the Epson is down.
    Every slip logs its destination so a pile of unexpected paper at either
    station can be traced to the order and path that produced it.
    """
    global _last_partial_send, _last_print_failure
    _last_partial_send = None
    if RECEIPT_PRINTER_IP:
        # OWNER REPORT 2026-08-03: every order came out ~4x on the receipt
        # printer AND ~4x in the kitchen. The bytes had already reached the
        # Epson and PRINTED — only the post-send status check came back
        # unanswered (the Epson does not answer like the BTP does) — and this
        # fallback then reprinted the very same slip in the kitchen.
        #
        # So: once the socket send SUCCEEDS the paper is already out, and the
        # copy must NEVER be re-sent to another printer. Only a genuine
        # delivery failure (connect/send raised — nothing left the Mac) may
        # fall back. Duplicate paper at two stations is far worse than one
        # unverified customer copy, which staff can reprint on demand.
        #
        # D-22 (9/26): a copy that died MID-send is the third case — part of
        # it IS on paper, so it neither counts as delivered nor falls back
        # (a whole copy in the kitchen beside a half page at the counter is
        # the same confusion). It returns False with _last_partial_send set;
        # _print_extra_copies re-queues it on the SAME printer under the
        # REPRINT banner. And a link the last probe found down gets nothing
        # (the gate), which is the connect-failed case: fall back.
        down = _link_gate(RECEIPT_PRINTER_IP, RECEIPT_PRINTER_PORT)
        if down:
            log(f"[RECEIPT PRINTER WAIT] {label or 'copy'}: link down at "
                f"{RECEIPT_PRINTER_IP}:{RECEIPT_PRINTER_PORT} ({down}) — nothing "
                f"sent; falling back to kitchen printer")
        else:
            try:
                why = _send_ticket(RECEIPT_PRINTER_IP, RECEIPT_PRINTER_PORT, data)
                if why is None:
                    _note_link_up(RECEIPT_PRINTER_IP, RECEIPT_PRINTER_PORT)
                    log(f"[COPY] {label or 'copy'} -> receipt {RECEIPT_PRINTER_IP}")
                else:
                    log(f"[COPY] {label or 'copy'} -> receipt {RECEIPT_PRINTER_IP} "
                        f"(delivered, unverified: {why}) — NOT reprinting elsewhere")
                return True
            except PartialSend as e:
                _set_last_partial(e.sent, e.total, str(e))
                _note_link_down(RECEIPT_PRINTER_IP, RECEIPT_PRINTER_PORT,
                                "partial send: %s" % (e.cause or "link dropped mid-copy"))
                log(f"[RECEIPT PRINTER ERROR] {label or 'copy'}: {e} — a partial "
                    f"page may be at the counter; NOT reprinting elsewhere")
                return False
            except Exception as e:
                log(f"[RECEIPT PRINTER ERROR] {e} — nothing delivered; "
                    f"falling back to kitchen printer")
    ok = print_ticket(data)
    if ok:
        log(f"[COPY] {label or 'copy'} -> kitchen {PRINTER_IP}")
    return ok

def _status_identifies(byte_val):
    """True if byte_val looks like an ESC/POS real-time status byte.

    The ESC/POS spec fixes bit0=0, bit1=1, bit4=1, bit7=0 in every DLE EOT
    reply — the store BTP-M300A answers exactly that (certified 7/25: 0x16,
    0x12, 0x12, 0x12). This function previously required bit4=0, which no
    spec-conforming reply has — so the REAL printer's bytes never identified
    and a genuine paper-out (0x32) fell through as "unknown printer". Both
    patterns are accepted now: the spec one, plus the legacy bit4=0 shape for
    any emulation that sends it. A printer that does not speak DLE EOT (some
    Star models, a CUPS/lpr relay, a USB bridge) returns nothing or noise;
    those fail this check and are treated as "status unknown" so we never
    wedge the line on an un-queryable printer.
    """
    if byte_val is None:
        return False
    return (byte_val & 0b10010011) == 0b00010010 or (byte_val & 0b10010011) == 0b00000010

def _query_status_byte(sock, command, timeout=1.5):
    """Send one DLE EOT command and read its single-byte reply (None on silence)."""
    try:
        sock.sendall(command)
        sock.settimeout(timeout)
        chunk = sock.recv(1)
    except (OSError, socket.timeout):
        return None
    return chunk[0] if chunk else None

def _interpret_status(offline, paper):
    """Map DLE EOT status bytes to a human fault reason, or None if healthy/unknown.

    Returns None when neither byte identifies as ESC/POS status (unknown printer)
    so the caller falls back to its prior send-and-mark behavior.
    """
    if not _status_identifies(offline) and not _status_identifies(paper):
        return None
    reasons = []
    if _status_identifies(offline):
        if offline & 0x04:            # bit2: cover open
            reasons.append("cover open")
        if offline & 0x20:            # bit5: printing stopped, paper end
            reasons.append("out of paper")
        if offline & 0x40:            # bit6: error occurred
            reasons.append("printer error")
    if _status_identifies(paper) and (paper & 0x60) == 0x60:  # bits5,6: roll paper end
        if "out of paper" not in reasons:
            reasons.append("out of paper")
    return ", ".join(reasons) if reasons else None

def check_printer_fault(ip=None, port=None, timeout=3):
    """Ask the printer whether it can actually print RIGHT NOW.

    Returns a short human reason string ("out of paper", "cover open", ...) when
    the printer reports a definite fault, else None. Pure: no global mutation.

    A None result means "healthy OR can't tell" — including a printer that is
    unreachable (that failure is already handled by print_ticket's retry) or one
    that does not answer DLE EOT. The caller only blocks marking-printed on a
    DEFINITE fault, so an un-queryable printer keeps its prior behavior.
    """
    if not STATUS_CHECK_ENABLED:
        return None
    ip = ip or PRINTER_IP
    port = port or PRINTER_PORT
    try:
        with socket.create_connection((ip, port), timeout=timeout) as s:
            offline = _query_status_byte(s, STATUS_OFFLINE)
            paper = _query_status_byte(s, STATUS_PAPER)
    except Exception as e:
        # The 9/26 hole: this "can't tell" let every poll go on to
        # _send_ticket while the Epson was unreachable. The link state is
        # recorded here so print_ticket's gate refuses the send.
        _note_link_down(ip, port, str(e) or type(e).__name__)
        return None
    if offline is None and paper is None and _link_key(ip, port) in _status_capable:
        # Connected, but a printer that always answers DLE EOT said nothing:
        # the socket is open into a dead link (a flapping Wi-Fi bridge
        # accepts the connect and drops the bytes). Not usable.
        _note_link_down(ip, port, "connected but no status reply — link not usable")
        return None
    # Remember that this printer answers DLE EOT: from then on, post-send
    # silence in _confirm_drained means a dead link, not a mute printer.
    _note_status_capable(ip, port, offline, paper)
    _note_link_up(ip, port)
    return _interpret_status(offline, paper)

def _note_bridge_status(status_code, what):
    """Remember a print-token rejection so the agent stops pretending it is well.

    401/403 means this spooler is locked out of the cloud: it will never see a
    ticket and every report it sends is discarded. That is an outage, not an
    idle store, so it must be loud locally and must not be reported as a
    healthy printer. Returns True when the call was rejected.
    """
    global _auth_rejected
    if status_code not in (401, 403):
        return False
    reason = (f"print bridge rejected our PRINT_BRIDGE_TOKEN (HTTP {status_code} "
              f"on {what}) — this spooler cannot see or acknowledge tickets")
    if _auth_rejected != reason:
        log(f"[AUTH ERROR] {reason}. Fix PRINT_BRIDGE_TOKEN on this computer.")
    _auth_rejected = reason
    return True


def _flag_printer_fault(reason):
    """Record a printer fault and page staff immediately.

    Reports the printer as NOT reachable in an out-of-band heartbeat so the
    Railway spooler watchdog (ops_alerting.check_print_spooler) pages the owner
    with the printer power/Wi-Fi/paper runbook — instead of waiting for the next
    polling heartbeat, which bare TCP reachability would report as healthy.
    """
    global _last_printer_error
    _last_printer_error = reason
    post_heartbeat(printer_reachable=False, mode="fault")

def print_test_ticket():
    ok = print_ticket(build_ticket(build_test_order()))
    if ok:
        print("[OK] Test kitchen ticket sent. Printer spooler path is ready.")
    else:
        print("[WARN] Test kitchen ticket failed. Check printer IP, port, WiFi, and power.")
    return ok

def probe_completion(ip=None, port=None):
    """Certify GS ( H fn=48 process-ID support on the printer (owner-approved).

    PRINTS ONE SMALL SLIP. If the printer does not implement GS ( H it may
    render the trailing request bytes as stray characters on that slip — which
    is exactly the evidence needed, and why this must never run unattended or
    against a printer mid-service. Only after this probe passes is it safe to
    set PRINT_COMPLETION_CHECK=1. Interactive command: plain print().
    """
    ip = ip or PRINTER_IP
    port = port or PRINTER_PORT
    ident = _next_completion_id()
    slip = (INIT + ALIGN_CTR + BOLD_ON + enc("COMPLETION PROBE (GS ( H)") + LF
            + BOLD_OFF + enc(_log_ts()) + LF + enc("ID " + ident.decode("ascii")) + LF + CUT)
    print(f"Probing {ip}:{port} for GS ( H process-ID completion support...")
    try:
        with socket.create_connection((ip, port), timeout=5) as s:
            t0 = time.time()
            s.sendall(slip + COMPLETION_REQ_PREFIX + ident)
            ok = _confirm_completion(s, ident, timeout=10)
            elapsed = time.time() - t0
    except Exception as e:
        print(f"[FAIL] Probe could not run: {e}")
        return False
    if ok:
        note_completion_certified(ip, port, elapsed)
        print(f"[OK] Process-ID echo received {elapsed:.2f}s after send — and GS ( H replies only")
        print("     after all preceding data is processed, so this printer supports the true")
        print(f"     completion barrier. CERTIFIED: recorded in {_CAPS_STORE}")
        print("     (printer_capabilities.json) — the completion check is now ON for this")
        print("     printer by default; tickets to it are marked VERIFIED only on the echo.")
        return True
    print(f"[NO] No process-ID echo within 10s (waited {elapsed:.2f}s).")
    print("     Inspect the probe slip: stray characters after the ID line mean the printer")
    print("     does NOT implement GS ( H. Not certified: tickets to it are confirmed by the")
    print("     DLE EOT status answer after the send (online = printed).")
    return False

def fetch_bridge_test_order():
    try:
        r = requests.get(
            f"{RAILWAY_URL}/api/print/test",
            headers=auth_headers(),
            timeout=8,
        )
        if r.status_code == 200:
            return r.json().get("order") or build_test_order()
        print(f"[TEST FETCH ERROR] print bridge returned HTTP {r.status_code}")
    except Exception as e:
        print(f"[TEST FETCH ERROR] {e}")
    return None

def check_railway_health():
    try:
        r = requests.get(f"{RAILWAY_URL}/api/print/health", timeout=5)
        if r.status_code == 200:
            return True, "Railway print bridge reachable."
        return False, f"Railway print bridge returned HTTP {r.status_code}."
    except Exception as e:
        return False, f"Railway print bridge unreachable: {e}"

def check_pending_print_auth():
    try:
        r = requests.get(
            f"{RAILWAY_URL}/api/orders/pending-print",
            headers=auth_headers(),
            timeout=8,
        )
        if r.status_code == 200:
            orders = r.json().get("orders", [])
            return True, f"Pending-print API accepted token; {len(orders)} order(s) waiting."
        if r.status_code in {401, 403}:
            return False, "Pending-print API rejected PRINT_BRIDGE_TOKEN."
        return False, f"Pending-print API returned HTTP {r.status_code}."
    except Exception as e:
        return False, f"Pending-print API check failed: {e}"

def check_phone_order_readiness():
    """Read the public, non-PII phone-order readiness report from Railway."""
    try:
        r = requests.get(f"{RAILWAY_URL}/api/phone-order/readiness", timeout=8)
        if r.status_code == 200:
            return True, r.json()
        return False, {"error": f"Phone-order readiness returned HTTP {r.status_code}."}
    except Exception as e:
        return False, {"error": f"Phone-order readiness check failed: {e}"}

def _yes_no(value):
    return "yes" if value else "no"

def print_public_status():
    """Print a no-token, no-paper status summary for the store computer."""
    print("Zorro's Print Agent Status")
    print(f"Railway: {RAILWAY_URL}")
    print(f"Printer target: {PRINTER_IP}:{PRINTER_PORT}")
    print("")

    ok, report = check_phone_order_readiness()
    if not ok:
        print(f"[FAIL] Railway readiness: {report.get('error', 'unavailable')}")
        print("Next: check internet connection, then run --status again.")
        return False

    checks = report.get("checks") or {}
    print(f"Status: {report.get('status', 'unknown')}")
    print(f"Readiness: {report.get('readiness_pct', '?')}%")
    print(f"Menu items: {report.get('menu_item_count', '?')}")
    print("")
    for key, label in (
        ("menu_brain_loaded", "Menu brain loaded"),
        ("live_order_enabled", "Live orders enabled"),
        ("phone_auto_activation_enabled", "Phone auto POS/KDS activation"),
        ("orders_schema_ready", "Order schema ready"),
        ("print_bridge_token_configured", "Railway print token configured"),
        ("store_spooler_auto_print_ready", "Store spooler + printer ready"),
    ):
        print(f"[{_yes_no(checks.get(key)).upper()}] {label}")

    blockers = report.get("blockers") or []
    if blockers:
        print("")
        print("Blockers:")
        for blocker in blockers:
            print(f"  - {blocker}")

    print("")
    if report.get("ready_for_phone_orders"):
        print("[READY] Phone orders can enter POS/KDS and print automatically.")
    elif report.get("software_ready_without_physical_printer"):
        print("[WAITING] Software is ready. Start this spooler on the store Wi-Fi.")
    else:
        print("[NOT READY] Fix the failed software checks before taking live phone orders.")
    print(f"Next: {report.get('next_action', 'Run --doctor on the store computer.')}")
    return True

def print_doctor():
    """Run no-paper setup diagnostics for the store computer."""
    print("Zorro's Print Agent Doctor")
    print(f"Railway: {RAILWAY_URL}")
    print(f"Printer: {PRINTER_IP}:{PRINTER_PORT}")
    print("")

    checks = []
    token_ready = bool(PRINT_BRIDGE_TOKEN)
    checks.append(("PRINT_BRIDGE_TOKEN", token_ready, "Set" if token_ready else "Missing"))
    if not token_ready:
        for label, ok, detail in checks:
            print(f"[{'OK' if ok else 'FAIL'}] {label}: {detail}")
        print("")
        print("Next: set PRINT_BRIDGE_TOKEN, then run --doctor again.")
        return False

    railway_ok, railway_detail = check_railway_health()
    checks.append(("Railway bridge", railway_ok, railway_detail))

    pending_ok, pending_detail = check_pending_print_auth()
    checks.append(("Pending-print auth", pending_ok, pending_detail))

    printer_ok = ping_printer(timeout=3)
    printer_detail = "Printer TCP port reachable." if printer_ok else (_last_printer_error or "Printer not reachable.")
    checks.append(("Printer network", printer_ok, printer_detail))

    # Reachable over TCP isn't enough — a powered printer can be out of paper or
    # cover-open. Ask its real status so setup catches a no-paper printer here.
    paper_fault = check_printer_fault() if printer_ok else None
    if printer_ok:
        checks.append(("Printer status", paper_fault is None,
                       f"Printer reports: {paper_fault}." if paper_fault else "Printer reports ready (or does not support status)."))
    printer_ok = printer_ok and not paper_fault

    heartbeat_ok = post_heartbeat(printer_reachable=printer_ok, mode="doctor")
    checks.append(("Heartbeat", heartbeat_ok, "Railway received heartbeat." if heartbeat_ok else "Heartbeat failed."))

    for label, ok, detail in checks:
        print(f"[{'OK' if ok else 'FAIL'}] {label}: {detail}")

    ready = all(ok for _, ok, _ in checks)
    print("")
    if ready:
        print("[READY] Spooler prerequisites are ready. Start the live spooler or run --test-bridge-print.")
    else:
        print("[NOT READY] Fix the FAIL item(s), then run --doctor again.")
    return ready

def print_bridge_test_ticket():
    order = fetch_bridge_test_order()
    if not order:
        print("[WARN] Bridge test ticket failed. Check RAILWAY_URL and PRINT_BRIDGE_TOKEN.")
        return False
    ok = print_ticket(build_ticket(order))
    if ok:
        print("[OK] Bridge test kitchen ticket sent. Railway, token, and printer path are ready.")
    else:
        print("[WARN] Bridge test print failed. Check printer IP, port, WiFi, and power.")
    return ok

def ping_printer(timeout=8, attempts=3):
    """Is the kitchen printer answering?  Retries before declaring it down.

    Owner report 2026-08-07: a stream of printer-offline SMS alerts while the
    printer was in fact printing.  Measured LAN round-trip to the BTP that
    morning was 0.7-1.3 SECONDS (a degraded wireless link, not a dead
    printer), and the polling loop pinged with timeout=1 — so the check could
    not succeed even when the printer was healthy.  One slow answer must not
    page anyone; only a printer that misses every attempt is treated as down.
    """
    global _last_printer_error
    last = ""
    for attempt in range(max(1, attempts)):
        try:
            with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=timeout) as s:
                s.close()
            _last_printer_error = ""
            if attempt:
                log(f"[PRINTER CHECK] Reachable on attempt {attempt + 1} — slow link, not down.")
            return True
        except Exception as e:
            last = str(e)
            if attempt + 1 < max(1, attempts):
                time.sleep(1.0)
    _last_printer_error = last
    log(f"[PRINTER CHECK] Not reachable at {PRINTER_IP}:{PRINTER_PORT} "
        f"after {max(1, attempts)} tries — {last}")
    return False

def fetch_orders():
    global _last_pending_orders_seen
    try:
        r = requests.get(
            f"{RAILWAY_URL}/api/orders/pending-print",
            headers=auth_headers(),
            timeout=8,
        )
        if r.status_code == 200:
            orders = r.json().get("orders",[])
            _last_pending_orders_seen = len(orders)
            return orders
        if not _note_bridge_status(r.status_code, "pending-print"):
            log(f"[FETCH ERROR] print bridge returned HTTP {r.status_code}")
    except Exception as e:
        log(f"[FETCH ERROR] {e}")
    _last_pending_orders_seen = 0
    return []

def _note_print_evidence(order_id, sends=1):
    """print_ticket just succeeded for this order: keep what it proved, for
    the mark-printed payload and printed_state.json. `sends` counts every
    attempt whose bytes left this Mac — on the one safe retry the evidence
    says so, because the first send may already be on the rail."""
    text = _last_print_evidence or "sent; no post-send evidence recorded — UNVERIFIED"
    if sends > 1:
        text = (f"confirmed on send {sends}; {sends - 1} earlier send(s) unconfirmed — "
                f"POSSIBLE DUPLICATE on the rail; {text}")
    _evidence_by_id[str(order_id)] = {
        "verified": bool(_last_print_completed or _last_print_confirmed_online),
        "evidence": text[:300],
        "sends": int(sends),
        "printer": _last_print_printer or "%s:%d" % (PRINTER_IP, PRINTER_PORT),
    }


def _print_verdict():
    """The suffix of the 'Printed #…' log line: verified, or loudly not."""
    if _last_print_completed:
        return " (verified: completion echo)"
    if _last_print_confirmed_online:
        return " (verified: printer answered online after send)"
    return " (UNVERIFIED: %s)" % (_last_print_evidence or "no post-send evidence")


# The wording the pre-10/5 build wrote for a ticket the printer answered
# ONLINE for. An entry carrying it is upgraded wherever it is read back
# (a heal, a paid-copy re-mark): the proof was on the ticket all along —
# ticket-based, never by who paid or how old the entry is.
_OLD_ONLINE_WORDING_RE = re.compile(
    r"DLE EOT status answered after send \(0x([0-9a-fA-F]{2})\)")


def _upgrade_legacy_evidence(verified, text):
    """(verified, text) with the old 'answered … — UNVERIFIED' wording turned
    into the online verdict when the byte it recorded had no fault bits."""
    text = str(text or "")
    if verified or "UNVERIFIED" not in text:
        return bool(verified), text
    m = _OLD_ONLINE_WORDING_RE.search(text)
    if not m:
        return False, text
    byte = int(m.group(1), 16)
    if not _status_identifies(byte) or _interpret_status(byte, None):
        return False, text
    head = text[:m.start()]
    sent_at = ""
    mm = re.search(r"sent to ([0-9A-Za-z.:\-]+);\s*$", head)
    if mm:
        sent_at = f" at {mm.group(1)}"
        head = head[:mm.start()]
    return True, (f"{head}{ONLINE_EVIDENCE} (DLE EOT 0x{byte:02x}){sent_at} "
                  f"[reclassified from the pre-10/5 wording]")[:300]


def _mark_payload(order_id):
    """Evidence the existing mark-printed call carries: printed_verified 1 on
    the GS ( H completion echo OR an online status answer after the send
    (owner 10/5), else 0, and the evidence text. A re-mark (a heal after a
    lost mark, a paid-copy relist) carries what the print proved at the
    time, from printed_state.json — upgraded if that build's wording was
    the old one."""
    key = str(order_id)
    ev = _evidence_by_id.get(key)
    if ev is None:
        st = _print_state.get(key) or {}
        if st.get("evidence"):
            verified, text = _upgrade_legacy_evidence(st.get("verified"), st["evidence"])
            ev = {"verified": verified,
                  "evidence": "healed mark (re-marked from the local print registry) — " + text,
                  "sends": int(st.get("sends") or 0)}
        else:
            ev = {"verified": False, "sends": 0,
                  "evidence": ("no print evidence recorded for this order on this Mac "
                               "(printed before the physical-ack build, or by hand) — UNVERIFIED")}
    return {
        "printed_verified": 1 if ev.get("verified") else 0,
        "print_evidence": str(ev.get("evidence") or "")[:300],
        "printer": str(ev.get("printer") or "%s:%d" % (PRINTER_IP, PRINTER_PORT)),
        "spooler_version": SPOOLER_VERSION,
        "sends": int(ev.get("sends") or 0),
    }


def mark_printed(order_id):
    return requests.post(
        f"{RAILWAY_URL}/api/orders/{order_id}/mark-printed",
        headers=auth_headers(),
        json=_mark_payload(order_id),
        timeout=5,
    )

def mark_printed_or_warn(order_id):
    try:
        response = mark_printed(order_id)
        if response.status_code == 200:
            return True
        if not _note_bridge_status(response.status_code, "mark-printed"):
            log(f"[MARK ERROR] order #{order_id} returned HTTP {response.status_code}")
    except Exception as e:
        log(f"[MARK ERROR] order #{order_id}: {e}")
    return False

def fetch_print_jobs():
    """Generic print jobs (Digital Fridge labels): opaque ESC/POS payloads
    the cloud composed — push verbatim, never rendered here."""
    try:
        r = requests.get(
            f"{RAILWAY_URL}/api/print/pending-jobs",
            headers=auth_headers(),
            timeout=8,
        )
        if r.status_code == 200:
            return r.json().get("jobs", [])
        if not _note_bridge_status(r.status_code, "pending-jobs"):
            print(f"[JOB FETCH ERROR] print bridge returned HTTP {r.status_code}")
    except Exception as e:
        print(f"[JOB FETCH ERROR] {e}")
    return []


#: Bridge job types that print at the COUNTER'S RECEIPT STATION instead of the
#: kitchen BTP. Everything not listed here (fridge labels, void slips, requeued
#: kitchen tickets) keeps the kitchen path.
#: driver_slip (2026-10-05): the DRIVER SLIP queued when a run is assigned —
#: the driver's paper, same station the order-time DRIVER COPY used.
#: bottle_label (2026-10-11): the drink BOTTLE LABELS the RECIPES tile queues
#: beside a recipe card — same cook, same counter, same paper as the card.
RECEIPT_STATION_JOB_TYPES = ("receipt_copy", "recipe_card", "driver_slip",
                             "bottle_label")


def process_pending_jobs_once():
    import base64
    printed_count = 0
    jobs = fetch_print_jobs()
    for idx, job in enumerate(jobs):
        jid = job.get("id")
        if not jid or jid in _printed_jobs:
            continue
        # D-22 (9/26): a job that died mid-send backs off (30 s → 5 min),
        # never the next 10 s poll; its retry carries the REPRINT banner.
        jkey = ("job", jid)
        if not _partial_retry_due(jkey):
            continue
        # 2026-09-09: the register's "reprint receipt" button queues a
        # receipt_copy job. That is customer paper and belongs on the
        # RECEIPT station (RECEIPT_PRINTER_IP), never the kitchen BTP —
        # print_receipt_copy() sends there and only falls back to the
        # kitchen when nothing left the Mac. Every other job type (fridge
        # labels, void slips, requeued tickets) keeps the kitchen path.
        #
        # 2026-09-21: the POS RECIPES button queues a recipe_card job and
        # rides the same lane. A recipe card is paper a cook reads standing
        # at the bench, so it belongs at the counter — and it must never
        # shoulder into the kitchen BTP's ticket stream mid-rush, nor be
        # held back by a kitchen-printer fault that has nothing to do with
        # it.
        job_type = str(job.get("job_type") or "").strip().lower()
        to_receipt_station = job_type in RECEIPT_STATION_JOB_TYPES
        if not to_receipt_station:
            # The fault probe is the KITCHEN printer's status; a receipt
            # copy never touches it, so a kitchen fault must not hold one.
            fault = check_printer_fault()
            if fault:
                _flag_printer_fault(fault)
                print(f"[PRINTER FAULT] {fault} — holding job #{jid}; will retry")
                break
            # D-22 (9/26): the probe could not reach the kitchen printer —
            # not one byte goes out; the jobs wait, one log line a minute.
            down = _link_down_reason()
            if down:
                waiting = [j.get("id") for j in jobs[idx:]
                           if j.get("id") and j.get("id") not in _printed_jobs]
                _log_link_wait(down, len(waiting), what="jobs")
                break
        try:
            payload = base64.b64decode(job.get("payload_b64") or "")
        except Exception as e:
            requests.post(
                f"{RAILWAY_URL}/api/print/jobs/{jid}/mark-failed",
                headers=auth_headers(), json={"error": f"bad payload: {e}"},
                timeout=5)
            continue
        if to_receipt_station:
            sent = bool(payload) and print_receipt_copy(
                _ticket_for_send(jkey, payload),
                label=f"{job_type} job #{jid} {job.get('title', '')}".strip())
        else:
            sent = bool(payload) and print_ticket(_ticket_for_send(jkey, payload))
        if sent:
            _clear_partial(jkey)
            try:
                requests.post(
                    f"{RAILWAY_URL}/api/print/jobs/{jid}/mark-printed",
                    headers=auth_headers(), timeout=5)
                _printed_jobs.add(jid)
                printed_count += 1
                print(f"Printed job #{jid} — {job.get('title', '')}")
            except Exception as e:
                print(f"[JOB MARK ERROR] #{jid}: {e} — may reprint")
        elif _last_partial_send:
            # Part of this job is on paper: book it, and hold the rest of
            # the batch for this poll — the link just proved unstable.
            _after_partial_send(jkey, f"job #{jid}",
                                ticket=str(job.get("title") or job_type))
            break
        else:
            print(f"Failed job #{jid} — will retry")
    return printed_count


_printed_jobs = set()



def maybe_kick_drawer(order):
    """Pop the cash drawer for a COMPLETED cash sale (best-effort).

    Only a cash-method ticket already marked paid kicks — an unpaid
    pay-at-counter order must not open the drawer before money changes
    hands, and card sales never do. Fires after the ticket printed and was
    acked, so a print failure never leaves the drawer open with no ticket.
    """
    if not DRAWER_KICK_ENABLED:
        return False
    method = str(order.get("payment_method") or "").strip().lower()
    status = str(order.get("payment_status") or "").strip().lower()
    if method != "cash" or status != "paid":
        return False
    try:
        _send_raw(PRINTER_IP, PRINTER_PORT, OPEN_DRAWER)
        log(f"[DRAWER] kicked for cash order #{order.get('id')}")
        return True
    except Exception as e:
        log(f"[DRAWER ERROR] {e}")
        return False


def report_print_failed(order_id, error):
    """Tell the bridge this ticket could NOT print (W5 print-job ledger).

    Best-effort: a failure to report must never affect the retry loop —
    the ticket stays pending and retries either way.

    Skipped once the bridge has rejected our token: re-POSTing a report it
    will 401 every poll forever just floods the request log with a symptom
    while hiding the cause (seen in prod 2026-08-03 — the same order reported
    every cycle for hours, all 401)."""
    if _auth_rejected:
        return
    body = {"error": str(error)[:300]}
    extra = _bytes_out_evidence(order_id, str(error))
    if extra:
        body.update(extra)
    try:
        r = requests.post(
            f"{RAILWAY_URL}/api/orders/{order_id}/print-failed",
            headers=auth_headers(),
            json=body,
            timeout=5,
        )
        _note_bridge_status(r.status_code, "print-failed")
    except Exception as e:
        log(f"[REPORT ERROR] order #{order_id}: {e}")


def _bytes_out_evidence(order_id, error):
    """Extra print-failed fields when ticket bytes DID leave this Mac for
    this order — an unconfirmed send (the one safe retry, a hold) or a
    partial page. The server then records the row as 'sent (unverified)',
    not 'failed'. None when nothing was ever sent (a link-down wait)."""
    sends = int(_unconfirmed_sends.get(order_id, 0))
    if not sends:
        st = _print_state.get(str(order_id)) or {}
        if st.get("held"):
            sends = int(st.get("sends") or 0)
    partial = int(_partial_sends.get(order_id, 0))
    if not sends and not partial:
        return None
    if sends:
        text = f"sent {sends}x to {PRINTER_IP}:{PRINTER_PORT}, unconfirmed — {error}"
    else:
        text = (f"partial send to {PRINTER_IP}:{PRINTER_PORT} ({partial}x) — a partial "
                f"page may be on the rail; {error}")
    return {"sent_unconfirmed": True, "sends": sends, "print_evidence": text[:300]}

def _copy_banner(label):
    """Centered, emphasized copy label prepended to duplicate receipts."""
    return (b"\x1b@" + b"\x1ba\x01" + b"\x1d!\x11"
            + ("*** %s ***" % label).encode("ascii", "replace")
            + b"\n" + b"\x1d!\x00" + b"\x1ba\x00")


def _print_extra_copies(o):
    """Owner directive 2026-07-31: three receipts per order, four on delivery.

    Copy 1 (kitchen) is the unlabeled ticket printed on the critical path
    above — its success is what marks the order printed. Copies 2..N print
    here BEST-EFFORT with a banner naming who each is for; a failed extra
    copy logs loudly but never fails the order or triggers reprint holds
    (a duplicate-confused kitchen is worse than a missing customer copy).
    """
    # Owner directive 2026-09-04: "just 2 kitchen 1 drivers copy" — the pile
    # of CUSTOMER + STORE + DRIVER receipts was confusing the line. Now:
    #   • the kitchen gets exactly TWO — the main ticket printed on the
    #     critical path above (prep), plus ONE "PACK COPY" here, both on the
    #     kitchen printer;
    #   • a DELIVERY adds ONE driver copy on the receipt/delivery printer;
    #   • no customer or store copies auto-print — the POS reprints on demand.
    oid = o.get("id") or o.get("order_id")
    is_delivery = "deliver" in str(o.get("order_type", "")).lower()
    kitchen_full = build_ticket(o)
    clean_full = build_ticket(o, include_specs=False)

    # 2nd kitchen copy (PACK) — same kitchen printer as the prep ticket.
    # A PACK copy that dies mid-send (D-22, 9/26) is re-queued on the same
    # printer under the REPRINT banner once the backoff passes — never a
    # second full copy beside a half page.
    # Owner 2026-09-29: "only a total of 1 kitchen print" — the PACK copy is OFF
    # unless PRINT_PACK_COPY=1.
    if _PACK_COPY_ON:
        pack = _copy_banner("PACK COPY") + kitchen_full
        try:
            why = _send_ticket(PRINTER_IP, PRINTER_PORT, pack)
            if why:
                log("[COPY ERROR] PACK COPY #%s did not print (%s)" % (oid, why))
            else:
                log("[COPY] PACK COPY #%s -> kitchen %s" % (oid, PRINTER_IP))
        except PartialSend as e:
            _note_link_down(PRINTER_IP, PRINTER_PORT,
                            "partial send: %s" % (e.cause or "link dropped mid-copy"))
            _set_last_partial(e.sent, e.total, str(e))
            _queue_copy_retry(oid, "PACK COPY", pack, to="kitchen")
        except Exception as e:
            log("[COPY ERROR] PACK COPY #%s: %s" % (oid, e))

    # 1 driver copy, delivery only, on the delivery/receipt printer. This is
    # the page that reached the driver incomplete on 9/26: a partial send is
    # re-queued on the same printer under the REPRINT banner.
    # DRIVER_COPY_AT=assign moves this sheet to assignment time: the server's
    # driver_slip bridge job prints then, and nothing prints here.
    if is_delivery and DRIVER_COPY_AT != "assign":
        driver = _copy_banner("DRIVER COPY") + clean_full
        try:
            if not print_receipt_copy(driver, label=f"DRIVER COPY #{oid}"):
                if _last_partial_send:
                    _queue_copy_retry(oid, "DRIVER COPY", driver, to="receipt")
                else:
                    log("[COPY ERROR] DRIVER COPY #%s did not print (%s)" % (
                        oid, _last_print_failure))
        except Exception as e:
            log("[COPY ERROR] DRIVER COPY #%s: %s" % (oid, e))

    # Owner 2026-09-06: "if it's paid, just one, to confirm a payment. So
    # maximum four in total." ONE PAID-confirmation copy on the receipt
    # printer when the order is already paid at print time — never more.
    if str(o.get("payment_status", "")).lower() == "paid":
        paid = _copy_banner("PAID - CONFIRMATION") + clean_full
        try:
            if not print_receipt_copy(paid, label=f"PAID COPY #{oid}"):
                if _last_partial_send:
                    _queue_copy_retry(oid, "PAID COPY", paid, to="receipt")
                else:
                    log("[COPY ERROR] PAID COPY #%s did not print (%s)" % (
                        oid, _last_print_failure))
        except Exception as e:
            log("[COPY ERROR] PAID COPY #%s: %s" % (oid, e))


def process_pending_orders_once():
    """Fetch, print, and mark each currently pending order once.

    Extracted from the infinite loop so setup tests can prove the store-computer
    spooler path without blocking forever.
    """
    printed_count = 0
    # Owner 2026-09-06 ("stop printing million tickets for one order"): an
    # operator hand-print registers the id in printed_ids.json from ANOTHER
    # process, but this running spooler only loaded that file at boot — so
    # in the seconds between a hand-print and the server's kitchen_printed
    # mark, this poll printed the same order again. Re-read the registry
    # every poll (it is tiny) so a hand-printed order is never re-printed.
    try:
        _printed.update(_load_printed())
        _merge_print_state_from_disk()
    except Exception:
        pass
    seen_this_poll = set()
    orders = fetch_orders()
    for idx, o in enumerate(orders):
        oid = o.get("id") or o.get("order_id")
        if not oid:
            log("[ORDER ERROR] Pending order missing id/order_id; skipping")
            continue
        if oid in seen_this_poll:
            # The same id twice in ONE response is a duplicate row, not a
            # requeue — nothing can have cleared the flag between them.
            continue
        seen_this_poll.add(oid)
        if oid in _printed:
            # Already on paper once. Two very different reasons the server
            # can serve it again — see the REQUEUE block near the top.
            decision, why = _classify_relisted(oid, o)
            if decision == "paidcopy":
                # Owner 2026-09-29 ("only a total of 1 kitchen print and 1
                # customer and 1 delivery"): a payment landing after the
                # kitchen ticket prints ONLY the customer's PAID confirmation
                # copy on the receipt printer — the kitchen ticket is never
                # reprinted for a payment.
                try:
                    paidc = _copy_banner("PAID - CONFIRMATION") + build_ticket(o)
                    okc = print_receipt_copy(paidc, label=f"PAID COPY #{oid}")
                    log("[COPY] PAID COPY (now paid, no kitchen reprint) #%s -> %s" % (oid, "ok" if okc else "FAILED"))
                except Exception as e:
                    log("[COPY ERROR] PAID COPY #%s: %s" % (oid, e))
                ok = mark_printed_or_warn(oid)
                _record_print_state(oid, o, marked=ok, requeue=True)
                _healed_marks.discard(oid)
                _persist_printed()
                continue
            if decision == "requeue":
                if oid in _held_unconfirmed:
                    continue
                if not _partial_retry_due(oid):
                    continue   # died mid-send earlier; backing off (D-22, 9/26)
                fault = check_printer_fault()
                if fault:
                    _flag_printer_fault(fault)
                    report_print_failed(oid, fault)
                    log(f"[PRINTER FAULT] {fault} — holding update ticket for #{oid} and the rest of this batch; will retry")
                    break
                down = _link_down_reason()
                if down:
                    _hold_for_link(down, _waiting_ids(orders[idx:]))
                    break
                ticket = o.get("ticket_number") or oid
                if print_ticket(_ticket_for_send(oid, _requeue_banner(why) + build_ticket(o))):
                    _clear_partial(oid)
                    _unconfirmed_sends.pop(oid, None)
                    _note_print_evidence(oid)
                    ok = mark_printed_or_warn(oid)
                    # ONE kitchen ticket. No PACK / DRIVER / PAID copies: the
                    # owner's 9/6 maximum was spent on the first print, and a
                    # correction the line cannot tell from a second order is
                    # exactly what the banner exists to prevent.
                    _record_print_state(oid, o, marked=ok, requeue=True)
                    _healed_marks.discard(oid)   # a failed re-mark may heal once more
                    _persist_printed()
                    printed_count += 1
                    tail = "" if ok else "; Railway did not mark it — server flag heals next poll"
                    log(f"[REQUEUE] #{ticket} {why} — update ticket printed once for {o.get('customer_name', '')}{_print_verdict()}{tail}")
                elif _last_partial_send:
                    # Part of the update ticket is on paper (D-22, 9/26):
                    # book it, retry under the REPRINT banner after the
                    # backoff, and hold the rest of this batch for now.
                    _after_partial_send(oid, f"#{oid}", order_id=oid, ticket=str(ticket))
                    break
                elif _last_send_unconfirmed:
                    # Bytes left the Mac; the update MAY be on paper. Never
                    # spray a correction: count it printed, let the heal path
                    # repair the flag, and say so loudly — on the row too.
                    _record_print_state(
                        oid, o, marked=False, requeue=True, verified=False,
                        evidence=f"update ticket sent, unconfirmed ({_last_print_failure}) — UNVERIFIED"[:300])
                    _healed_marks.discard(oid)
                    _unconfirmed_sends.pop(oid, None)
                    log(f"[REQUEUE] #{ticket} {why} — sent without confirmation, counted as printed; check the rail ({_last_print_failure})")
                else:
                    report_print_failed(oid, _last_print_failure or "print_ticket failed (send error)")
                    log(f"[REQUEUE] Failed update ticket for #{ticket} — will retry")
                continue
            # Owner 2026-09-06 (T-24 #999893): this spooler printed the ticket
            # at 15:49, but the mark-printed POST to the server failed on a
            # network blip, so the server kept serving it as unprinted — the
            # watchdog paged "NOT PRINTED" and an operator hand-printed a
            # DUPLICATE. The local registry is the truth here: heal the
            # server flag (idempotent) instead of leaving it to drift. Once
            # per id per process — a dead server must not turn into a storm.
            if oid not in _healed_marks:
                _healed_marks.add(oid)
                try:
                    if mark_printed_or_warn(oid):
                        if str(oid) in _print_state:
                            _note_state_marked(oid)
                        else:
                            # Printed before this build: remember what the
                            # server is serving now, so the NEXT change to
                            # this order is recognised as a requeue.
                            _record_print_state(oid, o, marked=True)
                        log(f"[HEAL] server flag for #{oid} repaired ({why})")
                except Exception as e:
                    log(f"[HEAL WARN] could not repair mark for #{oid}: {e}")
            continue
        # Belt-and-suspenders: if the server itself says this order is
        # already printed (an operator hand-printed and marked it), never
        # print it again regardless of what the pending list returned.
        if o.get("kitchen_printed") in (1, True, "1", "true"):
            _printed.add(oid)
            continue
        # Held after MAX_UNCONFIRMED_SENDS sent-but-unconfirmed attempts (the
        # original + the one safe retry): do NOT keep reprinting (each attempt
        # may have physically printed) and do NOT mark printed (it may never
        # have). Staff were paged at hold time; the order stays visibly
        # pending for a human to resolve. The only release is that human's
        # deliberate requeue — a POS correction/reprint carries the bridge's
        # UPDATED ORDER marker — and it prints once more, still bannered.
        if oid in _held_unconfirmed:
            if not _requeue_marker_text(o):
                continue
            _release_hold(oid, "staff requeued it from the POS")
        # D-22 (9/26): a ticket that died mid-send retries after its backoff
        # (30 s → 60 → 120 … capped at 5 min), never on the next 10 s poll.
        if not _partial_retry_due(oid):
            continue
        # Confirm the printer can actually print BEFORE we send + mark-printed.
        # A powered printer that is out of paper / cover-open accepts the bytes
        # and silently drops the ticket; marking it printed would lose the order.
        # One fault holds this whole batch (all orders share the printer) and
        # pages staff; the held orders stay pending and retry next poll.
        fault = check_printer_fault()
        if fault:
            _flag_printer_fault(fault)
            report_print_failed(oid, fault)
            log(f"[PRINTER FAULT] {fault} — holding #{oid} and the rest of this batch; will retry")
            break
        # D-22 (9/26): the SAME probe just tried to reach the printer. If it
        # could not, NOTHING is sent — the 9/26 loop went on to _send_ticket
        # 139 times into a dead link and put partial tickets on paper. The
        # batch waits; one "[PRINTER WAIT]" line a minute; nothing is marked.
        down = _link_down_reason()
        if down:
            _hold_for_link(down, _waiting_ids(orders[idx:]))
            break
        if print_ticket(_ticket_for_send(oid, build_ticket(o))):
            sends = _unconfirmed_sends.pop(oid, 0) + 1
            _clear_partial(oid)
            _note_print_evidence(oid, sends)
            # The verdict belongs to THIS ticket: the copies below reuse the
            # same send path and overwrite the last-print globals.
            verdict = _print_verdict()
            if mark_printed_or_warn(oid):
                _printed.add(oid)
                _persist_printed()   # survive a restart — never re-print
                _record_print_state(oid, o, marked=True, sends=sends, held=False)
                printed_count += 1
                # Extra copies come AFTER the order is marked printed. Before
                # this, a failed mark-printed left the order retryable and
                # every retry sprayed a fresh set of customer/store copies at
                # the counter.
                _print_extra_copies(o)
                maybe_kick_drawer(o)
                # "Printed" = dispatched to the printer and acked by Railway
                # (the mark-printed state). Only the completion barrier proves
                # the paper actually moved; the line says which it was, and
                # an unverified mark is LOUD (audit 9/26 item 5).
                log(f"Printed #{o.get('ticket_number','?')} — {o.get('customer_name','')}{verdict}")
            else:
                # The PAPER already printed. Owner 2026-09-08 ("no more stuck
                # orders"): T-22 Dan printed at 00:00:27, the mark-printed
                # POST timed out on a Railway blip, the id was NOT put in the
                # local registry, and the next poll printed the same ticket
                # AGAIN at 00:03:13. The local registry is the truth
                # (T-24 lesson): register it now so no poll re-prints it,
                # and let the [HEAL] path repair the server flag on the next
                # poll. Extra copies wait for the heal, so a duplicate never
                # sprays the counter either.
                _printed.add(oid)
                _persist_printed()
                _record_print_state(oid, o, marked=False, sends=sends, held=False)
                printed_count += 1
                log(f"Printed #{oid}{verdict}, but Railway did not mark it printed — registered locally; server flag will be healed next poll")
        else:
            if _last_partial_send:
                # PART of the ticket is on paper (D-22, 9/26). Never marked,
                # never re-sent on the next poll: booked with a backoff, the
                # retry carries the REPRINT banner, and after
                # PARTIAL_PAGE_AFTER partials the owner is paged once. The
                # rest of this batch waits — the link just proved unstable.
                _after_partial_send(oid, f"#{oid}", order_id=oid,
                                    ticket=str(o.get("ticket_number") or ""))
                break
            if _last_send_unconfirmed:
                # Ticket bytes went out but nothing confirmed them: the ticket
                # MAY have printed. ONE safe retry after the backoff, under
                # the POSSIBLE DUPLICATE banner (a reprint beats a silently
                # lost order, and the banner beats a confused line); a second
                # unconfirmed send holds and pages a human ONCE — never a
                # third, the count survives a restart (printed_state.json).
                n = _unconfirmed_sends.get(oid, 0) + 1
                _unconfirmed_sends[oid] = n
                if n >= MAX_UNCONFIRMED_SENDS:
                    _held_unconfirmed.add(oid)
                    _record_print_state(
                        oid, o, marked=False, sends=n, held=True, verified=False,
                        evidence=f"sent {n}x unconfirmed, held ({_last_print_failure}) — UNVERIFIED"[:300])
                    reason = (f"held after {n} unconfirmed sends (one safe retry used) — may "
                              f"have printed {n}x or 0x; check the rail; if it is NOT there, "
                              f"reprint it from the POS ({_last_print_failure})")
                    report_print_failed(oid, reason)
                    _flag_printer_fault(f"order #{oid} {reason}")
                    log(f"[HOLD] #{oid} sent {n}x without confirmation — no more auto-reprints; "
                        f"staff paged; a POS reprint releases it")
                    continue
                _record_print_state(
                    oid, o, marked=False, sends=n, held=False, verified=False,
                    evidence=f"sent {n}x unconfirmed ({_last_print_failure}) — retry pending"[:300])
                _retry_not_before[oid] = _clock() + UNCONFIRMED_RETRY_BACKOFF
                report_print_failed(
                    oid, f"{_last_print_failure or 'sent, unconfirmed'}; one safe retry in "
                         f"{int(UNCONFIRMED_RETRY_BACKOFF)}s under the POSSIBLE DUPLICATE banner, then hold")
                log(f"[UNCONFIRMED] #{oid} sent once without confirmation ({_last_print_failure}) — "
                    f"one safe retry in {int(UNCONFIRMED_RETRY_BACKOFF)}s under the POSSIBLE DUPLICATE "
                    f"banner, then hold")
                continue
            report_print_failed(oid, _last_print_failure or "print_ticket failed (send error)")
            log(f"Failed #{oid} — will retry")
    # Copies that died mid-send earlier (PACK / DRIVER / PAID) get their
    # bannered retry here, once their backoff has passed.
    try:
        _drain_copy_retries()
    except Exception as e:
        log(f"[COPY ERROR] copy retry pass failed: {e}")
    return printed_count


def _release_hold(oid, why):
    """A human requeued a held order (the bridge marker is on the payload):
    let it print ONCE more, still under the POSSIBLE DUPLICATE banner — the
    earlier sends may be on the rail. The send count is KEPT (the banner
    and the evidence come from it, and it must read the same as after a
    restart): a send that is unconfirmed again holds straight away."""
    _held_unconfirmed.discard(oid)
    _retry_not_before.pop(oid, None)
    st = _print_state.get(str(oid)) or {}
    _unconfirmed_sends[oid] = max(int(_unconfirmed_sends.get(oid, 0)),
                                  int(st.get("sends") or 0))
    _update_print_state(oid, held=False)
    log(f"[HOLD RELEASED] #{oid} — {why}; printing once under the POSSIBLE DUPLICATE banner")


def _waiting_ids(batch):
    """Ids in `batch` that would print if the link were up — the count the
    [PRINTER WAIT] line names and the orders the ledger hears about once."""
    ids = []
    for o in batch:
        oid = o.get("id") or o.get("order_id")
        if not oid or oid in _held_unconfirmed:
            continue
        if o.get("kitchen_printed") in (1, True, "1", "true"):
            continue
        if oid in _printed and _classify_relisted(oid, o)[0] != "requeue":
            continue
        ids.append(oid)
    return ids

def run():
    global _last_printer_error
    if not acquire_spooler_lock():
        return False
    backup = f" | Backup: {BACKUP_PRINTER_IP}:{BACKUP_PRINTER_PORT}" if BACKUP_PRINTER_IP else ""
    receipt = f" | Receipts: {RECEIPT_PRINTER_IP}:{RECEIPT_PRINTER_PORT}" if RECEIPT_PRINTER_IP else ""
    log(f"Zorro's Print Agent | Printer: {PRINTER_IP}:{PRINTER_PORT}{backup}{receipt} | Railway: {RAILWAY_URL}")
    if COMPLETION_CHECK_MODE == "off":
        log("[COMPLETION] PRINT_COMPLETION_CHECK=0 — no completion barrier; a ticket is "
            "marked VERIFIED when the printer answers online after the send")
    elif _completion_enabled_for(PRINTER_IP, PRINTER_PORT):
        log(f"[COMPLETION] {PRINTER_IP}:{PRINTER_PORT} certified for GS ( H completion — "
            f"a ticket is marked VERIFIED on the echo; one safe retry, then hold")
    else:
        log(f"[COMPLETION] {PRINTER_IP}:{PRINTER_PORT} not certified for GS ( H — a ticket is "
            f"marked VERIFIED when the printer answers online after the send (DLE EOT); "
            f"--probe-completion adds the stronger completion echo")
    auth_headers()
    printer_ready = ping_printer()
    post_heartbeat(printer_reachable=printer_ready, mode="startup")
    if printer_ready:
        log("[OK] Printer reachable. Spooler is ready.")
    else:
        log("[WARN] Printer is not reachable yet. Orders will retry until the printer is online.")
    log("Press Ctrl+C to stop\n")
    while True:
        try:
            process_pending_orders_once()
            process_pending_jobs_once()
            # A printer that is reachable over TCP but out of paper / cover-open is
            # NOT ready — report it as not-reachable so the watchdog keeps paging.
            reachable = ping_printer()
            fault = check_printer_fault() if reachable else None
            if fault:
                _last_printer_error = fault
            # D-22 (9/26): a link that accepts a bare TCP connect but drops
            # the bytes (or answers no status) is NOT a reachable printer —
            # tickets are waiting on it. Say so, so the spooler watchdog
            # pages the owner the same way it does for a dead printer.
            link_down = _link_down_reason()
            if link_down and not fault:
                _last_printer_error = f"printer link down: {link_down}"
            # A locked-out spooler is an outage even when the printer itself is
            # fine — it will never receive a ticket. Never report it as healthy.
            if _auth_rejected:
                _last_printer_error = _auth_rejected
            post_heartbeat(
                printer_reachable=(reachable and not fault and not link_down
                                   and not _auth_rejected),
                mode="polling")
        except Exception as e:
            log(f"[ERROR] {e}")
        time.sleep(POLL_INTERVAL)
    return True

def main(argv=None):
    args = list(sys.argv[1:] if argv is None else argv)
    if "--status" in args or "status" in args:
        return 0 if print_public_status() else 1
    if "--test-print" in args or "test-print" in args:
        return 0 if print_test_ticket() else 1
    if "--test-bridge-print" in args or "test-bridge-print" in args:
        auth_headers()
        return 0 if print_bridge_test_ticket() else 1
    if "--doctor" in args or "doctor" in args:
        return 0 if print_doctor() else 1
    if "--probe-completion" in args or "probe-completion" in args:
        return 0 if probe_completion() else 1
    if "--check" in args or "check" in args:
        auth_headers()
        printer_ready = ping_printer()
        post_heartbeat(printer_reachable=printer_ready, mode="check")
        return 0 if printer_ready else 1
    if "--once" in args or "once" in args:
        auth_headers()
        processed = process_pending_orders_once()
        post_heartbeat(printer_reachable=ping_printer(), mode="once")
        return 0 if processed >= 0 else 1
    return 0 if run() else 1

if __name__ == "__main__":
    raise SystemExit(main())