"""NEOva2TruckAnchorBackend — **LAB / EXPERIMENTAL**. Track-A laboratory bridge only.

  ⚠️  THIS IS NOT A PRODUCTION ANCHOR AND NEVER PRODUCES AN A2 ANCHOR.  ⚠️

Track A (this file) is the throwaway lane that lets the AI Arena and the apps be tested
end-to-end against a real NEOva2 today: apps → the EXISTING CL → this adapter → NEOva2.
It is a laboratory: deleted at will, no production promises, no evidence anyone may rely on.

Track B — the real mission ledger (apps → NEOfortress → CL2 → NEOcomp/Shredder → NEOva2) —
is a separate architecture and does NOT inherit anything from this file. When that lands,
the wire code below is at most the outbound adapter of a future Shredder; the semantics
here are lab semantics and must not be carried across.

What this file honestly is: a proven-compatible piece of WIRE. It showed that CL's 32-byte
batch_root reaches the real NEOva2 door and lands durably on two real shelves, with no
translator service and no change to NEOva2. That is a transport result, not an evidence
result — see "WHAT THIS DELIBERATELY DOES NOT DO" below.

Construction REQUIRES `CL_NEOVA2_LAB_MODE=1`, so it cannot be wired into anything that
matters by accident.

WHY A SIXTH BACKEND AND NOT A TRANSLATOR SERVICE
------------------------------------------------
`NEOvaAnchorBackend` (neova.py) speaks the NEOtrust v0.6/0.7 HTTP seam
(`POST /v2/submit`, `GET /v2/era|/v2/bundle`). The NEOva2 fleet speaks NO HTTP at
all — its door is a raw TCP stream of self-describing binary frames. Two different
engines, two different protocols; the shape (send ONE opaque 32-byte commitment) is
identical on both.

`AnchorBackend` is already the parametrized seam for exactly this, so the bridge is a
backend, not a new daemon: no extra service, no extra failure domain, no change to
NEOva2 (which is scope-frozen). Grunf's law — the simplest path that still carries
the load.

WHAT THE WIRE ACTUALLY IS
-------------------------
One frame = 40 B header (`>4sHH8sQQII`: MAGIC "NVT1", version, flags, gateway_id[8],
epoch, batch_seq, count, reserved) + count*32 B commitments + 32 B SHA-256 digest over
(header || commitments) + 64 B Ed25519 signature over that digest. ONE signature per
batch; NEOva2 never inspects the 32 bytes. We send count=1: CL's `batch_root` IS the
commitment, so only the opaque root crosses (guardrail #2 holds byte-for-byte).

The door replies ONLY after the sender half-closes, with one `>QQ`
(durable_batches, durable_objects) — and it counts a batch durable only once BOTH
independent shelves have fsync'd it. So one connection per anchor round: connect →
send → shutdown(SHUT_WR) → read 16 B → close. At CL's anchor cadence that is free.

THE TLS CONFIRMATION GAP, AND THE READBACK LEG THAT CLOSES IT
-------------------------------------------------------------
The door's `>QQ` summary is **plaintext-only**. The mechanism (not just the observation):
Python's `ssl` has no TLS half-close — after `unwrap()` sends close_notify the client's
TLS state is gone, so the door's reply arrives as undecryptable raw records (0x17
application-data), exactly what the fence documented. Not fixable sender-side; we stop
trying. neova2's own sender says the same in `bench/rate_sender.py`.

The confirmation channel is therefore OUT-OF-BAND: the fleet's own proof path
(`neova/proofs.py`). The brain's time-triggered closing ladder witnesses every pending
position range at the next hour close (60 s at the live fleet's test cadence), after
which a PROOF BUNDLE exists for our commitment: its canonical POSITION, the carrying
batch identity (gateway_id, epoch, batch_seq — ours), the covering range record, BOTH
qualified-witness attestations (base + witness-about-witness), a Merkle membership
proof, and the closed-tier ladder. `neova.proofs.verify_bundle` is a pure function a
relying party runs against a PINNED registry — written, in its own words, for exactly
this ledger.

So over TLS this backend now: submits the truck, closes with close_notify (never reads
the in-band summary), then POLLS a configured readback for the bundle and attests ONLY
after the bundle passes BOTH gates. First the structural/identity gate: commitment ==
our root (independently re-observed, not an echo), batch identity == the one we signed,
position inside the batch's range, both attestation kinds present, embedded registry
byte-identical to the pinned one. Then the CRYPTOGRAPHIC gate: the pure verifier
(`neova2_bundle_verify.py`, a frozen verbatim copy of neova2's own `verify_bundle` —
kept as a copy because CL's CI has no neova2 checkout to import) judges the Merkle
membership proof, both witness signatures against the PINNED registry, and every tier
ladder step. A readback REQUIRES a non-empty pinned registry (enforced at construction)
and the verifier is never called with trusted_registry=None on the attestation path —
a bundle judged by its own embedded keys would be a self-signed proof. Any failure on
either gate leaves the submission UNCONFIRMED (AnchorError), never an attestation.
The e2e verify step additionally judges the same bundle with neova2's own copy of the
verifier, so the frozen copy is cross-checked against upstream (plus the live-fleet
golden-bundle test in tests/fixtures/).

With TLS on and NO readback configured, the #31 fence is unchanged: refuse before
spending a sequence. A submission we cannot confirm is A0/A1, never an attestation.

ASSURANCE — WHAT THIS CAN AND CANNOT HONESTLY CLAIM
---------------------------------------------------
The door's reply is a DURABILITY confirmation, not an address. The live NEOva2 fleet
exposes no readback: `bench/proof_server.py` (GET /proof/<commitment>) exists in the
neova2 repo but is NOT deployed on the fleet. Consequences, stated rather than hidden:

  * `assurance_level` is `lab_neova2_two_shelf_durable_NOT_A2` — exactly what happened: the
    door verified structure + digest + one signature, and both shelves confirmed durable.
    It is NOT a canonical position, NOT a segment commitment, NOT crypto-final, NOT A2.
  * `opaque_root` is necessarily an ECHO of what we submitted. The ABC intends it to be
    the independently re-observed root; with no readback there is nothing to re-observe,
    so `verify_anchor`'s root check is VACUOUS on this backend. `independently_verified`
    in `host_anchor_status()` says so out loud.
  * `anchored_at_height` carries the sender-allocated `batch_seq`. It is OUR address for
    the batch (the identity the shelves store under), never a NEOva2 canonical position.
  * `finality_certificate_hash` carries the door-verified `batch_digest`. It is a real,
    door-checked binding over the exact bytes received — it is NOT a finality certificate
    (the ABC fixes the field name; the semantics are declared here, not faked).

WHAT THIS DELIBERATELY DOES NOT DO (the honest gap list)
--------------------------------------------------------
Review found that a truthful docstring does not repair a wrong OBJECT: `AnchorAttestation`
means A2 by contract, so returning one for a durability confirmation is a category error
however well it is commented. This file does not pretend otherwise — it is fenced instead:

  1. **The A3 name carries the truth.** `assurance_level` is
     `lab_neova2_two_shelf_durable_NOT_A2`. A relying party reading only that field cannot
     mistake this for an anchor. It is deliberately ugly.
  2. **Construction requires `CL_NEOVA2_LAB_MODE=1`.** No accidental production wiring.
  3. **mTLS submission + readback confirmation EXIST now** (see the section above), but
     only when a readback is configured; TLS without one still refuses. The confirmed
     assurance string is `neova2_witnessed_position_not_era_sealed` — witnessed by two
     qualified masters and hour-closed, honestly NOT era-sealed and NOT crypto-final
     (no EraSeal ceremony has ever occurred; `Anchored != crypto-final` stands).
  4. **Cryptographic verification gates the attestation IN this backend** — via the
     frozen verbatim copy in `neova2_bundle_verify.py` (see the section above), always
     against a REQUIRED pinned registry, never against the bundle's embedded keys. The
     drift risk of a copy is guarded twice: the e2e verify step re-judges with neova2's
     own verifier, and CI judges the real 2026-07-25 live-fleet bundle with the copy.
  5. **The sequence allocator is process-local.** It is atomic and fsync'd (below), which is
     enough for one lab CL, but it is NOT safe for several CL cells sharing one gateway
     identity. Track B needs a real distributed allocator.

Track B (NEOfortress → CL2 → NEOcomp/Shredder) is where 5 gets solved properly, with a
clean design rather than by patching this. Do not grow this file toward production.

SEQUENCE SAFETY
---------------
Batch identity is (gateway_id, gateway_epoch, batch_seq). Reusing a sequence with different
content is equivocation at the shelf, so the allocator persists BEFORE use — atomically
(temp file + fsync + rename + directory fsync), so a crash or a full disk skips a sequence
and can never resurrect an old one from a half-written file. Process-local lock only; see
limitation 5 above.
"""

from __future__ import annotations

import hashlib
import json
import socket
import struct
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Optional

from cl.anchor.backend import AnchorAttestation, AnchorBackend, AnchorError

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

# --- the wire (vendored constants; must match neova2 `neova/truck.py`) ---------------
MAGIC = b"NVT1"
VERSION = 1
HEADER = struct.Struct(">4sHH8sQQII")   # 40 bytes
SUMMARY = struct.Struct(">QQ")          # door reply: durable_batches, durable_objects
OBJ = 32
# The A3 field a relying party actually reads. It must be impossible to mistake for an A2
# anchor, because this backend cannot produce one: nothing independent is re-observed. The
# name carries that, not a comment — an honest docstring does not fix a wrong object.
ASSURANCE = "lab_neova2_two_shelf_durable_NOT_A2"
# The readback-confirmed A3: root independently re-observed from the fleet's proof bundle,
# position canonical, TWO qualified-witness attestations over the covering range — and
# still honestly NOT era-sealed (no EraSeal ceremony exists; Anchored != crypto-final).
ASSURANCE_CONFIRMED = "neova2_witnessed_position_not_era_sealed"

# --- the readback (vendored bundle contract; must match neova2 `neova/proofs.py`) -----
BUNDLE_FORMAT = "neova2-proof-bundle-v1"
_RECORD_DOMAIN = b"NVREC1"


def range_record_hash(range_start: int, count: int, object_root: bytes,
                      prev_record_hash: bytes) -> bytes:
    """The identity hash of a NEOva2 closed range record. Vendored formula; must match
    neova2 `neova/witness.py::ClosedRecord.record_hash` byte-for-byte — it is what the
    attestation's `finality_certificate_hash` carries, so a third party can bind our
    anchor to the exact witnessed record the fleet's book chains."""
    return hashlib.sha256(_RECORD_DOMAIN
                          + struct.pack(">QQ", range_start, count)
                          + object_root + prev_record_hash).digest()

# Explicit opt-in. Constructing this backend without it fails, so the lab bridge can never be
# wired into anything that matters by accident or by a copied env file.
LAB_ACK_ENV = "CL_NEOVA2_LAB_MODE"

# A socket factory so tests can drive a fake door; the host binds the real one.
Connector = Callable[[], socket.socket]


def build_truck(gateway_id: bytes, gateway_epoch: int, batch_seq: int,
                commitments: bytes, signer: Ed25519PrivateKey) -> bytes:
    """One signed NVT1 frame. Mirrors neova2's `truck.pack` exactly — the door recomputes
    this digest over the bytes it received and refuses the WHOLE batch on any mismatch."""
    if len(gateway_id) != 8:
        raise ValueError("gateway_id must be exactly 8 bytes")
    count, remainder = divmod(len(commitments), OBJ)
    if remainder or count < 1:
        raise ValueError("commitments must be 32*N bytes, N >= 1")
    header = HEADER.pack(MAGIC, VERSION, 0, gateway_id, gateway_epoch, batch_seq, count, 0)
    digest = hashlib.sha256(header + commitments).digest()
    return header + commitments + digest + signer.sign(digest)


class _SequenceAllocator:
    """Persist-before-use monotonic batch sequences. A crash must skip, never reuse."""

    def __init__(self, path: Optional[Path]) -> None:
        self._path = path
        self._lock = threading.Lock()
        self._next = 1
        if path is not None and path.exists():
            try:
                self._next = int(path.read_text().strip()) + 1
            except (ValueError, OSError):
                raise AnchorError(f"unreadable NEOva2 sequence file {path} — refusing to guess")

    def take(self) -> int:
        with self._lock:
            seq = self._next
            if self._path is not None:
                self._persist(seq)                # durable BEFORE the frame goes out
            self._next = seq + 1
            return seq

    def _persist(self, seq: int) -> None:
        """Atomic + fsync'd. A plain write_text can leave a torn or unflushed file, and a torn
        file that parses as a SMALLER number would resurrect a spent sequence — reuse with
        different content is equivocation at the shelf, the one thing this must never allow."""
        import os

        directory = self._path.parent
        directory.mkdir(parents=True, exist_ok=True)
        temporary = self._path.with_suffix(self._path.suffix + ".tmp")
        with open(temporary, "w") as handle:
            handle.write(str(seq))
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, self._path)         # atomic within the directory
        directory_fd = os.open(directory, os.O_RDONLY)
        try:
            os.fsync(directory_fd)                # the rename itself must survive a crash
        finally:
            os.close(directory_fd)


class _IdentityWatermark:
    """The chain's identity memory: (gateway_epoch, last_seq) in ONE JSON file per chain,
    persisted BEFORE use — the file IS the chain, and the same crash law as the bare
    sequence file holds: a crash skips, never reuses.

    Why the epoch lives here and not in the environment: batch identity is
    (gateway, epoch, seq), and a hand-picked epoch is how one equivocates by accident —
    re-running an old epoch resubmits an already-anchored commitment under a new identity
    (observed live, 2026-07-28, refused by the fleet's own guard). So the watermark is
    authoritative: config seeds it exactly once, and a config epoch that later DISAGREES
    with the chain's memory is refused loudly — an identity fork must be an explicit act
    (a new watermark file), never a stale environment variable.

    After a confirmed readback the watermark also records (last_batch_root,
    last_position) — the per-chain progress mark the LAB rig previously kept by hand.
    """

    SCHEMA = "NEOVA2_CHAIN_WATERMARK_V1"

    def __init__(self, path: Path, *, seed_epoch: int,
                 legacy_seq_path: Optional[Path] = None) -> None:
        self._path = Path(path)
        self._lock = threading.Lock()
        if self._path.exists():
            try:
                data = json.loads(self._path.read_text())
                epoch, last_seq = int(data["gateway_epoch"]), int(data["last_seq"])
            except (ValueError, KeyError, OSError) as error:
                raise AnchorError(
                    f"unreadable NEOva2 chain watermark {self._path} — refusing to guess: {error}")
            if data.get("watermark_schema") != self.SCHEMA:
                raise AnchorError(
                    f"{self._path}: unknown watermark schema {data.get('watermark_schema')!r}")
            if seed_epoch != epoch:
                raise AnchorError(
                    f"gateway epoch mismatch: config says {seed_epoch}, the chain watermark "
                    f"{self._path} holds {epoch} — the watermark is the chain's memory; match "
                    f"it, or fork the identity EXPLICITLY with a new watermark file")
            self._state = data
        else:
            last_seq = 0
            if legacy_seq_path is not None and Path(legacy_seq_path).exists():
                try:
                    last_seq = int(Path(legacy_seq_path).read_text().strip())
                except (ValueError, OSError) as error:
                    raise AnchorError(
                        f"unreadable legacy sequence file {legacy_seq_path} — refusing to "
                        f"seed a watermark from a guess: {error}")
            self._state = {"watermark_schema": self.SCHEMA, "gateway_epoch": seed_epoch,
                           "last_seq": last_seq, "last_batch_root": None,
                           "last_position": None}
            self._persist()                       # the identity exists before any use of it
        self._next = int(self._state["last_seq"]) + 1

    @property
    def epoch(self) -> int:
        return int(self._state["gateway_epoch"])

    def take(self) -> "tuple[int, int]":
        with self._lock:
            seq = self._next
            self._state["last_seq"] = seq
            self._persist()                        # durable BEFORE the frame goes out
            self._next = seq + 1
            return self.epoch, seq

    def confirm(self, batch_root_hex: str, position: int) -> None:
        with self._lock:
            self._state["last_batch_root"] = batch_root_hex
            self._state["last_position"] = position
            self._persist()

    def snapshot(self) -> dict:
        return dict(self._state)

    def _persist(self) -> None:
        """Atomic + fsync'd, same dance as the bare sequence file and for the same reason:
        a torn file that parses as a SMALLER number would resurrect a spent sequence."""
        import os

        directory = self._path.parent
        directory.mkdir(parents=True, exist_ok=True)
        temporary = self._path.with_suffix(self._path.suffix + ".tmp")
        with open(temporary, "w") as handle:
            handle.write(json.dumps(self._state, sort_keys=True))
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, self._path)
        directory_fd = os.open(directory, os.O_RDONLY)
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)


@dataclass
class NEOva2TruckConfig:
    host: str
    port: int
    gateway_id: bytes                       # exactly 8 bytes
    gateway_epoch: int = 1
    signer: Optional[Ed25519PrivateKey] = field(default=None, repr=False)
    sequence_path: Optional[Path] = None
    # The chain watermark (epoch + seq + last confirmed anchor, one file per chain).
    # When set it supersedes sequence_path for allocation; a legacy sequence file is
    # read ONCE to seed the watermark, then left alone.
    watermark_path: Optional[Path] = None
    timeout_s: float = 30.0
    # Mutual TLS 1.3, pinned-CA model — the live fleet's door REQUIRES a client certificate
    # (`netguard.accept_hardened(srv, srv_tls)` with `CERT_REQUIRED`). neova2's own tlsutil
    # docstring names this link explicitly: "Every link (Ledger->door, door->shelf)". Hostname
    # checking is off by design there (IP endpoints, pinned CA), so we mirror that exactly.
    # All three empty = plaintext, which only a local test door accepts.
    tls_cert: str = ""
    tls_key: str = ""
    tls_ca: str = ""
    # The readback leg. `readback_fetch(commitment)` returns the fleet's proof bundle dict
    # for that 32-byte commitment, or None while it is not yet witnessed (the poll keeps
    # going until `readback_poll_s` expires). Sources: an HTTP proof server
    # (`dual_anchor.neova2_truck_backend_from_env` builds that fetcher from
    # CL_NEOVA2_PROOF_URL), a watched file (the scripted live round), or a test stub.
    readback_fetch: Optional[Callable[[bytes], Optional[dict]]] = None
    readback_poll_s: float = 120.0
    readback_interval_s: float = 3.0
    # PINNED qualified-witness registry (witness_id -> 64-char hex Ed25519 pub), obtained
    # out of band — never trusted from the bundle itself. REQUIRED whenever a readback is
    # configured (enforced in __post_init__): without a pinned trust anchor the verifier
    # would judge signatures against the bundle's own embedded keys, which any forger
    # controls — a self-signed proof is no proof. The bundle's embedded registry must be
    # byte-identical to this or the anchor is refused (impostor set).
    pinned_witnesses: Optional[dict] = None

    def __post_init__(self) -> None:
        if len(self.gateway_id) != 8:
            raise ValueError("gateway_id must be exactly 8 bytes")
        if self.signer is None:
            raise ValueError("a gateway signing key is required (owner-set; never defaulted)")
        if any((self.tls_cert, self.tls_key, self.tls_ca)) and not all(
                (self.tls_cert, self.tls_key, self.tls_ca)):
            raise ValueError("mutual TLS needs all three of cert/key/ca, or none at all")
        if self.readback_fetch is not None and not self.pinned_witnesses:
            raise ValueError(
                "a readback requires a non-empty pinned_witnesses registry: without a "
                "PINNED trust anchor the bundle would be judged against its own embedded "
                "keys, which any forger controls — a self-signed proof is no proof "
                "(fail-closed at construction, before any sequence is spent)")

    def tls_context(self):
        """The pinned-CA mutual-TLS client context, or None for a plaintext local test door."""
        if not self.tls_cert:
            return None
        import ssl

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
        ctx.load_cert_chain(self.tls_cert, self.tls_key)
        ctx.load_verify_locations(self.tls_ca)
        ctx.check_hostname = False          # pinned-CA, IP endpoints (mirrors neova2 tlsutil)
        ctx.verify_mode = ssl.CERT_REQUIRED
        return ctx


class NEOva2TruckAnchorBackend(AnchorBackend):
    """Submit CL's 32-byte batch_root to the live NEOva2 door as a one-object truck."""

    def __init__(self, config: NEOva2TruckConfig,
                 connector: Optional[Connector] = None,
                 lab_mode_ack: bool = False) -> None:
        import os

        if not (lab_mode_ack or os.environ.get(LAB_ACK_ENV, "").strip() == "1"):
            raise ValueError(
                f"NEOva2TruckAnchorBackend is a LAB/EXPERIMENTAL Track-A bridge and never "
                f"produces an A2 anchor — set {LAB_ACK_ENV}=1 to acknowledge, or pass "
                f"lab_mode_ack=True in tests. Track B (NEOfortress -> CL2 -> NEOcomp/Shredder) "
                f"is where production anchoring belongs.")
        self._cfg = config
        if config.watermark_path is not None:
            self._wm: Optional[_IdentityWatermark] = _IdentityWatermark(
                config.watermark_path, seed_epoch=config.gateway_epoch,
                legacy_seq_path=config.sequence_path)
            self._seq = None
        else:
            self._wm = None
            self._seq = _SequenceAllocator(config.sequence_path)
        self._connect = connector or self._default_connector
        self._last: dict = {}

    def _default_connector(self) -> socket.socket:
        raw = socket.create_connection((self._cfg.host, self._cfg.port),
                                       timeout=self._cfg.timeout_s)
        ctx = self._cfg.tls_context()
        if ctx is None:
            return raw
        try:
            return ctx.wrap_socket(raw, server_hostname=None)
        except Exception:
            raw.close()
            raise

    def anchor(self, batch_root: bytes) -> AnchorAttestation:
        if len(batch_root) != 32:
            raise AnchorError("NEOva2 truck anchor requires a 32-byte batch_root")
        over_tls = self._cfg.tls_context() is not None
        if over_tls and self._cfg.readback_fetch is None:
            # Fail-closed BEFORE spending a sequence: over TLS the door's `>QQ` summary is
            # unreliable by neova2's own account, so with no readback there is no
            # confirmation channel and a submission is A0/A1, never A2 (guardrail #1).
            raise AnchorError(
                "NEOva2 anchoring over mutual TLS cannot be confirmed: the door's durability "
                "summary is plaintext-only (neova2 bench/rate_sender.py says so, and the bytes "
                "read back over TLS are raw record framing). A submission we cannot confirm is "
                "A0/A1, not A2 — refusing to attest. Fix: configure the readback leg "
                "(readback_fetch / CL_NEOVA2_PROOF_URL, backed by neova2's proof path — "
                "bench/proof_server.py or the fleet's durable files).")
        if self._wm is not None:
            epoch, seq = self._wm.take()          # the chain's memory, not the environment
        else:
            epoch, seq = self._cfg.gateway_epoch, self._seq.take()
        frame = build_truck(self._cfg.gateway_id, epoch, seq,
                            batch_root, self._cfg.signer)
        digest = frame[HEADER.size + OBJ:HEADER.size + OBJ + 32]

        try:
            sock = self._connect()
        except OSError as error:
            raise AnchorError(f"NEOva2 door unreachable at "
                              f"{self._cfg.host}:{self._cfg.port}: {error}") from error
        summary = None
        try:
            sock.settimeout(self._cfg.timeout_s)
            sock.sendall(frame)
            if over_tls:
                # No in-band read over TLS (see docstring): end the stream with a proper
                # close_notify so the door sees clean EOF at a frame boundary, and let the
                # READBACK below be the only confirmation channel.
                try:
                    sock = sock.unwrap()
                except (OSError, ValueError):
                    pass                       # best-effort; the door tolerates abortive close
            else:
                sock.shutdown(socket.SHUT_WR)  # the door answers only after half-close
                summary = self._recv_exact(sock, SUMMARY.size)
        except OSError as error:
            raise AnchorError(f"NEOva2 truck submission failed: {error}") from error
        finally:
            try:
                sock.close()
            except OSError:
                pass

        if not over_tls:
            if summary is None:
                raise AnchorError("NEOva2 door closed without a durability summary "
                                  "(frame refused: bad magic/version/digest/signature)")
            durable_batches, durable_objects = SUMMARY.unpack(summary)
            # Fail-closed: anything but exactly our one batch on two shelves is NOT durable.
            if durable_batches != 1 or durable_objects != 1:
                raise AnchorError(
                    f"NEOva2 door did not confirm our batch durable "
                    f"(batches={durable_batches}, objects={durable_objects})")
            self._last = {
                "gateway_id": self._cfg.gateway_id.hex(),
                "gateway_epoch": epoch,
                "batch_seq": seq,
                "batch_digest": digest.hex(),
                "durable_batches": durable_batches,
                "durable_objects": durable_objects,
            }
        else:
            self._last = {
                "gateway_id": self._cfg.gateway_id.hex(),
                "gateway_epoch": epoch,
                "batch_seq": seq,
                "batch_digest": digest.hex(),
                "summary": "skipped (TLS: confirmation is the readback, in-band is dead)",
            }

        if self._cfg.readback_fetch is None:
            return AnchorAttestation(
                opaque_root=batch_root,            # ECHO — no readback exists to re-observe it
                anchored_at_height=seq,            # OUR batch sequence, not a NEOva2 position
                finality_certificate_hash=digest,  # door-verified digest, NOT a certificate
                assurance_level=ASSURANCE,
                block_hash=b"",                    # NEOva2 has no block concept
            )
        return self._confirm_via_readback(batch_root, epoch, seq)

    def _confirm_via_readback(self, batch_root: bytes, epoch: int, seq: int) -> AnchorAttestation:
        """Poll the readback until the fleet can PROVE our commitment, then attest from the
        proof — never from our own submission. The wait is real: the brain witnesses the
        pending range at its next hour close (60 s at the live fleet's test cadence)."""
        deadline = time.monotonic() + self._cfg.readback_poll_s
        bundle = None
        while time.monotonic() < deadline:
            try:
                bundle = self._cfg.readback_fetch(batch_root)
            except Exception as error:  # a broken readback is "not yet", not an attestation
                bundle = None
                last_fetch_error = error  # noqa: F841 — kept for debuggability
            if bundle is not None:
                break
            time.sleep(self._cfg.readback_interval_s)
        if bundle is None:
            raise AnchorError(
                f"NEOva2 readback did not surface a proof bundle for our commitment within "
                f"{self._cfg.readback_poll_s:.0f}s — the submission may be durable but is "
                f"UNCONFIRMED (A0/A1); refusing to attest")
        position, record_hash = self._validate_bundle(bundle, batch_root, epoch, seq)
        # The CRYPTOGRAPHIC judgment — NEOva2's own pure verifier (frozen verbatim copy):
        # membership proof, both witness signatures against the PINNED registry, every
        # ladder step. A bundle that fails here leaves the submission UNCONFIRMED; an
        # attestation gated only on structure would accept a forged bundle.
        from cl.anchor.neova2_bundle_verify import verify_bundle

        # pinned_witnesses is guaranteed non-empty here (__post_init__ refuses a readback
        # without one) — the verifier is NEVER called with trusted_registry=None on the
        # attestation path; a self-signed bundle judged by its own keys is no proof.
        if not self._cfg.pinned_witnesses:
            raise AnchorError("readback confirmation without a pinned witness registry — "
                              "refusing to attest (fail-closed)")
        pinned = {wid.encode("utf-8"): bytes.fromhex(hx)
                  for wid, hx in self._cfg.pinned_witnesses.items()}
        judgment = verify_bundle(bundle, trusted_registry=pinned)
        if not judgment.get("valid"):
            raise AnchorError(
                f"NEOva2 readback bundle failed cryptographic verification "
                f"({judgment.get('reason')}) — the submission stays UNCONFIRMED (A0/A1); "
                f"refusing to attest")
        if self._wm is not None:
            self._wm.confirm(batch_root.hex(), position)
        self._last.update(position=position, record_hash=record_hash.hex(),
                          confirmed="readback",
                          crypto_verified_by=judgment.get("confirmed_by"))
        return AnchorAttestation(
            opaque_root=bytes.fromhex(bundle["commitment"]),  # RE-OBSERVED from the proof
            anchored_at_height=position,                      # canonical NEOva2 position
            finality_certificate_hash=record_hash,            # witnessed range record hash
            assurance_level=ASSURANCE_CONFIRMED,
            block_hash=b"",
        )

    def _validate_bundle(self, bundle: dict, batch_root: bytes, epoch: int, seq: int):
        """Structural + identity checks CL can make without neova2 imports. Signatures,
        Merkle membership and the tier ladder are the pure verifier's job
        (`neova.proofs.verify_bundle`) in the verify step — deliberately not duplicated."""
        def refuse(reason: str):
            raise AnchorError(f"NEOva2 readback bundle refused: {reason}")

        try:
            if bundle.get("format") != BUNDLE_FORMAT:
                refuse(f"unknown format {bundle.get('format')!r}")
            if bundle.get("commitment") != batch_root.hex():
                refuse("bundle commitment != our batch_root — wrong proof")
            batch = bundle.get("batch") or {}
            ours = (self._cfg.gateway_id.hex(), epoch, seq)
            got = (batch.get("gateway_id"), batch.get("gateway_epoch"),
                   batch.get("batch_seq"))
            if got != ours:
                refuse(f"bundle batch identity {got} != ours {ours} — "
                       f"someone else's batch carries this commitment?")
            position = bundle["position"]
            start, count = batch["position_start"], batch["count"]
            if not (isinstance(position, int) and start <= position < start + count):
                refuse(f"position {position} outside our batch range "
                       f"[{start}, {start + count})")
            kinds = sorted(a.get("kind") for a in bundle.get("attestations", []))
            if kinds != ["BASE", "WAW"]:
                refuse(f"need exactly one BASE + one WAW attestation, got {kinds}")
            ids = {a.get("witness_id") for a in bundle["attestations"]}
            if len(ids) != 2:
                refuse("both attestations from the SAME witness — WaW must be independent")
            if self._cfg.pinned_witnesses is not None and \
                    bundle.get("qualified") != self._cfg.pinned_witnesses:
                refuse("embedded qualified registry != the PINNED registry "
                       "(possible impostor witness set)")
            rr = bundle["range_record"]
            if not rr["range_start"] <= position < rr["range_start"] + rr["count"]:
                refuse("covering range record does not cover the position")
            record_hash = range_record_hash(rr["range_start"], rr["count"],
                                            bytes.fromhex(rr["object_root"]),
                                            bytes.fromhex(rr["prev_record_hash"]))
        except AnchorError:
            raise
        except (KeyError, TypeError, ValueError) as error:
            raise AnchorError(f"NEOva2 readback bundle refused: malformed "
                              f"({type(error).__name__}: {error})") from error
        return position, record_hash

    @staticmethod
    def _recv_exact(sock: socket.socket, n: int) -> Optional[bytes]:
        buf = bytearray()
        while len(buf) < n:
            chunk = sock.recv(n - len(buf))
            if not chunk:
                return None
            buf.extend(chunk)
        return bytes(buf)

    def host_anchor_status(self) -> dict:
        """Reported through `/v1/anchor-status`. Says what this backend is AND is not."""
        tls = self._cfg.tls_context() is not None
        readback = self._cfg.readback_fetch is not None
        return {
            "mode": "neova2_truck",
            "door": f"{self._cfg.host}:{self._cfg.port}",
            "tls": tls,
            "readback": readback,
            "chain_watermark": self._wm.snapshot() if self._wm is not None else None,
            "pinned_witnesses": sorted(self._cfg.pinned_witnesses)
                                if self._cfg.pinned_witnesses else None,
            "assurance": ASSURANCE_CONFIRMED if readback else ASSURANCE,
            "independently_verified": readback,   # root re-observed from the proof bundle
            "can_anchor": readback or not tls,
            "detail": ("submits the 32B batch_root as a one-object NVT1 truck; the door "
                       "confirms two-shelf durability."
                       + ("  Readback ON: attests only after the fleet's proof bundle "
                          "re-observes the root at a canonical position under two "
                          "qualified-witness attestations (still NOT era-sealed)."
                          if readback else
                          "  No readback: the root is an echo and there is no canonical "
                          "position — durable != positioned.")
                       + ("  BLOCKED over TLS: the in-band durability summary is "
                          "plaintext-only, so a submission cannot be confirmed and is never "
                          "attested (guardrail #1). Configure the readback leg "
                          "(proof_server.py or a durable-file bundle) to unblock."
                          if tls and not readback else "")),
            "last_submission": self._last or None,
        }
