#!/usr/bin/env python3
"""Small iSCSI PDU bridge/target surface for the GDR/XDVD host bridge.

v234 adds READ CD probe handling for benign Windows/DISCInfo media sniffing. v232 added standard MMC probe handlers for READ TRACK INFORMATION/SET STREAMING/extra READ DVD STRUCTURE probes. v229 adds explicit legacy-vs-effective media-window delta trace fields. v213 adds a non-authoritative firmware-shadow observer/report for the 8050L MODE SENSE/SELECT 0x3E auth state. v208 adds probe-clean regression trace fields and known-good reporting. v205 adds stateful eject/relock handling and fills remaining benign MMC/DISCInfo probe responses. v204 adds a stateful auth/view mode for live iSCSI: it starts in Windows/video direct view, tracks the 8050L MODE SELECT unlock/sticky sequence, then switches READ CAPACITY/MODE SENSE and READ(10/12) to the proven unlocked Xbox game window.

v202 preserves the v201 single-LUN attach baseline and adds Windows MMC media-recognition responses for READ TOC, READ DISC INFORMATION, GET CONFIGURATION, GET EVENT STATUS, MODE SENSE page 2A, and EVPD INQUIRY.  The new backend uses the v192-v198 proven media window rather than pre-enumerated static READ payloads.
Earlier builds treated byte 4 plus bytes 5..6 as a 24-bit DataSegmentLength;
in iSCSI byte 4 is TotalAHSLength and bytes 5..7 are DataSegmentLength.
Real Windows sent an 84-byte first Login text segment, but the bridge read
zero bytes and wrote malformed response lengths. v154 reads/skips AHS correctly and v156 answers Windows discovery LoginOperationalNegotiation keys conservatively, and v158 restores the SendTargets portal-group tag in TargetAddress because Windows rejected a tagless TargetAddress as invalid response text. It also keeps the cooperative stop path and adds a login-PDU guard. It handles:

* Login Request/Response for normal and discovery sessions
* Text Request/Response, including a minimal SendTargets response
* NOP-Out/NOP-In keepalives
* SCSI Command/Data-In/Response PDUs
* SCSI Data-Out PDUs, including a simple R2T path for write/data-out commands
* Logout Request/Response
* Windows attach-oriented negotiation defaults and VPD/REPORT LUNS/READ FORMAT CAPACITIES probes

The SCSI/MMC semantics remain delegated to gdr_host_jsonl_bridge.HostBridge and
GDRScsiParser, so this file is a transport/protocol layer rather than a second
implementation of drive behavior.
"""
from __future__ import annotations

import argparse, hashlib, json, queue, socket, struct, sys, threading, time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from gdr_scsi_command_parser import GDRScsiParser, parse_hex_bytes, hx
from gdr_media_model import GDRMediaModel
from hlds_profiles import apply_session_to_namespace
from gdr_host_jsonl_bridge import Catalog, HostBridge, load_requests

from hlds_gdr_live_media_window_responder import (
    CANONICAL_GDR8050L_0L23_BASENAME,
    build_catalog as dynamic_build_catalog,
    bytes_to_hex as dynamic_bytes_to_hex,
    decode_read_cdb as dynamic_decode_read_cdb,
    enforce_config_provenance as dynamic_enforce_config_provenance,
    find_catalog_match as dynamic_find_catalog_match,
    get_session_model as dynamic_get_session_model,
    hex_to_bytes as dynamic_hex_to_bytes,
    is_read_cdb as dynamic_is_read_cdb,
    load_json as dynamic_load_json,
    make_sense_illegal_request as dynamic_make_sense_illegal_request,
    maybe_truncate_to_allocation as dynamic_maybe_truncate_to_allocation,
    read_capacity_of_file as dynamic_read_capacity_of_file,
    read_media_window as dynamic_read_media_window,
    resolve_media_lba_delta as dynamic_resolve_media_lba_delta,
    resolve_media_path as dynamic_resolve_media_path,
)

OP_NOP_OUT = 0x00
OP_SCSI_CMD = 0x01
OP_TASK_MGMT_REQ = 0x02
OP_LOGIN_REQ = 0x03
OP_TEXT_REQ = 0x04
OP_SCSI_DATA_OUT = 0x05
OP_LOGOUT_REQ = 0x06
OP_SNACK_REQ = 0x10

OP_NOP_IN = 0x20
OP_SCSI_RSP = 0x21
OP_TASK_MGMT_RSP = 0x22
OP_LOGIN_RSP = 0x23
OP_TEXT_RSP = 0x24
OP_SCSI_DATA_IN = 0x25
OP_LOGOUT_RSP = 0x26
OP_R2T = 0x31
OP_REJECT = 0x3F

SCSI_STATUS_GOOD = 0x00
SCSI_STATUS_CHECK_CONDITION = 0x02


def pad4_len(n: int) -> int:
    return (4 - (n & 3)) & 3


def u24(n: int) -> bytes:
    return int(n).to_bytes(3, 'big')


def read_exact(sock: socket.socket, n: int) -> bytes:
    out = bytearray()
    while len(out) < n:
        chunk = sock.recv(n - len(out))
        if not chunk:
            raise EOFError(f'socket closed while reading {n} bytes, got {len(out)}')
        out += chunk
    return bytes(out)


def bhs_ahs_len(bhs: bytes) -> int:
    """Return the Additional Header Segment length in bytes.

    iSCSI BHS byte 4 is TotalAHSLength in 4-byte words; bytes 5..7 are the
    24-bit DataSegmentLength. Earlier bridge versions accidentally read bytes
    4..6 as the data length, which makes a Windows Login Request with
    DataSegmentLength=0x54 look like an empty request.
    """
    return int(bhs[4]) * 4


def bhs_data_len(bhs: bytes) -> int:
    return int.from_bytes(bhs[5:8], 'big')


def recv_pdu(sock: socket.socket) -> tuple[bytes, bytes]:
    bhs = read_exact(sock, 48)
    ahs_len = bhs_ahs_len(bhs)
    if ahs_len:
        _ = read_exact(sock, ahs_len)
    data_len = bhs_data_len(bhs)
    data = read_exact(sock, data_len) if data_len else b''
    pad = pad4_len(data_len)
    if pad:
        _ = read_exact(sock, pad)
    return bhs, data


def prepare_pdu_bhs(bhs: bytes, data: bytes = b'') -> bytes:
    if len(bhs) != 48:
        raise ValueError('iSCSI BHS must be 48 bytes')
    b = bytearray(bhs)
    b[4] = 0
    b[5:8] = u24(len(data))
    return bytes(b)


def send_pdu(sock: socket.socket, bhs: bytes, data: bytes = b'') -> None:
    sock.sendall(prepare_pdu_bhs(bhs, data) + data + (b'\x00' * pad4_len(len(data))))


def parse_text_kv(data: bytes) -> dict[str, str]:
    out: dict[str, str] = {}
    for part in data.split(b'\x00'):
        if not part or b'=' not in part:
            continue
        k, v = part.split(b'=', 1)
        out[k.decode('utf-8', errors='replace')] = v.decode('utf-8', errors='replace')
    return out


def text_kv_bytes(items: list[tuple[str, str]] | dict[str, str]) -> bytes:
    if isinstance(items, dict):
        iterable = list(items.items())
    else:
        iterable = items
    return b''.join((str(k) + '=' + str(v)).encode('utf-8') + b'\x00' for k, v in iterable)


def cdb_len_from_opcode(op: int) -> int:
    # SCSI CDB group-code length table plus MMC/SBC opcodes used here.
    if op in (0x00, 0x03, 0x12, 0x15, 0x1A, 0x1B, 0x1E, 0x23):
        return 6
    if op in (0x25, 0x28, 0x43, 0x46, 0x4A, 0x51, 0x55, 0x5A):
        return 10
    if op in (0xAD, 0xA0, 0xA8):
        return 12
    group = (op >> 5) & 0x7
    if group == 0:
        return 6
    if group in (1, 2):
        return 10
    if group == 5:
        return 12
    return 16


def expected_xfer_from_cdb(cdb: bytes) -> int:
    if not cdb:
        return 0
    op = cdb[0]
    if op in (0x03, 0x12, 0x1A) and len(cdb) >= 5:
        return cdb[4]
    if op in (0x43, 0x46, 0x4A, 0x51, 0x52, 0x5A) and len(cdb) >= 9:
        return int.from_bytes(cdb[7:9], 'big')
    if op == 0x25:
        return 8
    if op == 0x23 and len(cdb) >= 9:
        return int.from_bytes(cdb[7:9], 'big')
    if op == 0xA0 and len(cdb) >= 10:
        return int.from_bytes(cdb[6:10], 'big')
    if op == 0x28 and len(cdb) >= 9:
        return int.from_bytes(cdb[7:9], 'big') * 2048
    if op == 0xA8 and len(cdb) >= 10:
        return int.from_bytes(cdb[6:10], 'big') * 2048
    if op == 0xAD and len(cdb) >= 10:
        return int.from_bytes(cdb[8:10], 'big')
    if op == 0xBE and len(cdb) >= 9:
        # READ CD transfer length is a 24-bit block count at bytes 6..8.
        # Windows/DISCInfo usually supplies ExpectedDataTransferLength in the
        # iSCSI BHS; this fallback uses conservative raw-sector sizing for
        # local smoke PDUs. The Data-In builder will still truncate to the
        # actual initiator expected length when present.
        blocks = int.from_bytes(cdb[6:9], 'big')
        return blocks * 2352
    if op == 0xB6 and len(cdb) >= 11:
        return int.from_bytes(cdb[9:11], 'big')
    if op in (0x15, 0x55):
        if op == 0x15 and len(cdb) >= 5:
            return cdb[4]
        if op == 0x55 and len(cdb) >= 9:
            return int.from_bytes(cdb[7:9], 'big')
    return 0


@dataclass
class IscsiSession:
    target_name: str
    host: str
    port: int
    portal_group_tag: int = 1
    stat_sn: int = 1
    exp_cmd_sn: int = 1
    max_cmd_sn: int = 64
    next_ttt: int = 1
    tsih: int = 1
    command_index: int = 0
    pending_data_out: dict[int, dict[str, Any]] = field(default_factory=dict)
    login_stage: int = 0
    session_type: str = 'Normal'

    def common_response_bhs(self, opcode: int, req_bhs: bytes | None = None) -> bytearray:
        b = bytearray(48)
        b[0] = opcode
        if req_bhs is not None:
            b[16:20] = req_bhs[16:20]  # Initiator Task Tag
            try:
                cmd_sn = int.from_bytes(req_bhs[24:28], 'big')
                if cmd_sn != 0xFFFFFFFF:
                    self.exp_cmd_sn = max(self.exp_cmd_sn, (cmd_sn + 1) & 0xFFFFFFFF)
                    self.max_cmd_sn = max(self.max_cmd_sn, (self.exp_cmd_sn + 63) & 0xFFFFFFFF)
            except Exception:
                pass
        b[24:28] = int(self.stat_sn).to_bytes(4, 'big')
        b[28:32] = int(self.exp_cmd_sn).to_bytes(4, 'big')
        b[32:36] = int(self.max_cmd_sn).to_bytes(4, 'big')
        self.stat_sn = (self.stat_sn + 1) & 0xFFFFFFFF
        return b

    def alloc_command_index(self) -> int:
        idx = self.command_index
        self.command_index += 1
        return idx

    def alloc_ttt(self) -> int:
        ttt = self.next_ttt
        self.next_ttt = (self.next_ttt + 1) & 0xFFFFFFFF
        if self.next_ttt == 0xFFFFFFFF:
            self.next_ttt = 1
        return ttt


def make_login_request(itt: int = 1, cid: int = 0, *, session_type: str = 'Normal', csg: int = 0, nsg: int = 1, cmd_sn: int = 1) -> tuple[bytes, bytes]:
    """Build a Login Request for local smoke tests.

    v149 used 0x87 for a single-step login.  That accidentally encoded
    CSG=1/NSG=3 while the request text was still first-login style.  Windows is
    stricter, so v150's smoke client uses the same two-stage progression a real
    initiator normally uses: security negotiation (0 -> 1), then operational
    negotiation (1 -> 3).
    """
    text_items = [
        ('InitiatorName', 'iqn.2026-06.local.mn103s-smoke'),
        ('SessionType', session_type),
        ('HeaderDigest', 'None'),
        ('DataDigest', 'None'),
        ('MaxRecvDataSegmentLength', '262144'),
        ('DefaultTime2Wait', '0'),
        ('DefaultTime2Retain', '0'),
        ('InitialR2T', 'No'),
        ('ImmediateData', 'Yes'),
    ]
    if csg == 0:
        text_items.insert(0, ('AuthMethod', 'None'))
    text = text_kv_bytes(text_items)
    b = bytearray(48)
    b[0] = OP_LOGIN_REQ | 0x40  # immediate
    b[1] = 0x80 | ((int(csg) & 3) << 2) | (int(nsg) & 3)  # transit + CSG/NSG
    b[2] = 0x00  # max version
    b[3] = 0x00  # min version
    b[8:14] = b'GDRSMK'
    b[14:16] = int(cid).to_bytes(2, 'big')
    b[16:20] = int(itt).to_bytes(4, 'big')
    b[24:28] = int(cmd_sn).to_bytes(4, 'big')
    b[28:32] = (0).to_bytes(4, 'big')
    return bytes(b), text

def login_response_value(key: str, value: str, sess: IscsiSession | None, session_type: str) -> str | None:
    """Return a conservative login-negotiation response value for one key.

    The real Windows discovery trace from v154 showed the initiator sending a
    LoginOperationalNegotiation request with values such as
    MaxBurstLength=Irrelevant.  Answering those with numeric values caused
    Windows to repeat the same Login Request hundreds of thousands of times.
    For Irrelevant-valued keys, answer Irrelevant.  For normal values, answer
    only the key being negotiated rather than an unsolicited full key set.
    """
    k = str(key)
    v = str(value)
    vl = v.lower()
    if vl == 'irrelevant':
        return 'Irrelevant'
    if k == 'AuthMethod':
        return 'None' if 'none' in vl or not v else 'Reject'
    if k in ('HeaderDigest', 'DataDigest'):
        return 'None'
    if k == 'MaxRecvDataSegmentLength':
        return '262144'
    if k == 'MaxBurstLength':
        return '262144'
    if k == 'FirstBurstLength':
        return '262144'
    if k == 'MaxOutstandingR2T':
        return '1'
    if k == 'DefaultTime2Wait':
        return '0'
    if k == 'DefaultTime2Retain':
        return '0'
    if k == 'InitialR2T':
        return 'No'
    if k == 'ImmediateData':
        return 'Yes'
    if k == 'DataPDUInOrder':
        return 'Yes'
    if k == 'DataSequenceInOrder':
        return 'Yes'
    if k == 'ErrorRecoveryLevel':
        return '0'
    if k == 'MaxConnections':
        return '1'
    if k == 'TargetName' and sess and session_type.lower() == 'normal':
        return sess.target_name
    if k == 'TargetPortalGroupTag' and sess and session_type.lower() == 'normal':
        return str(sess.portal_group_tag)
    # InitiatorName and SessionType are declarative initiator keys; do not echo
    # them. Unknown extension keys are also omitted instead of being rejected.
    return None


def make_login_response(req_bhs: bytes, req_data: bytes = b'', sess: IscsiSession | None = None) -> tuple[bytes, bytes]:
    kv = parse_text_kv(req_data)
    session_type = kv.get('SessionType') or (sess.session_type if sess else 'Normal')
    if sess:
        sess.session_type = session_type

    req_flags = int(req_bhs[1])
    req_transit = bool(req_flags & 0x80)
    req_continue = bool(req_flags & 0x40)
    csg = (req_flags >> 2) & 0x03
    nsg = req_flags & 0x03

    # Mirror the initiator's CSG/NSG transition rather than forcing 0x87.
    if req_transit:
        if nsg not in (0, 1, 3):
            nsg = 1 if csg == 0 else 3
        resp_flags = 0x80 | ((csg & 3) << 2) | (nsg & 3)
        if sess:
            sess.login_stage = nsg
    else:
        resp_flags = (0x40 if req_continue else 0x00) | ((csg & 3) << 2)

    text_items: list[tuple[str, str]] = []
    security_stage = (csg == 0 and nsg != 3)

    if security_stage:
        # Only negotiate authentication during SecurityNegotiation.  Windows may
        # send AuthMethod=None or it may send no text on this first PDU; either
        # way, do not include TargetPortalGroupTag or operational keys here.
        text_items.append(('AuthMethod', 'None'))
    else:
        # In operational negotiation, answer requested keys only.  This matters
        # for discovery sessions because Windows sends Irrelevant-valued keys
        # after the basic digest/time keys; returning numeric values there makes
        # it loop forever in login instead of sending SendTargets.
        for k, v in kv.items():
            rv = login_response_value(k, v, sess, session_type)
            if rv is not None:
                text_items.append((k, rv))

        # Smoke clients and simple initiators may send an operational transition
        # with no text.  For a normal session, provide the minimum defaults; for
        # discovery, keep it empty so SendTargets can follow.
        if not text_items and session_type.lower() == 'normal':
            for k, v in (
                ('HeaderDigest', 'None'),
                ('DataDigest', 'None'),
                ('MaxRecvDataSegmentLength', '262144'),
                ('InitialR2T', 'No'),
                ('ImmediateData', 'Yes'),
                ('ErrorRecoveryLevel', '0'),
                ('TargetPortalGroupTag', str(sess.portal_group_tag if sess else 1)),
            ):
                text_items.append((k, v))

        if session_type.lower() == 'normal' and sess and 'TargetPortalGroupTag' not in kv and 'TargetName' in kv:
            text_items.append(('TargetPortalGroupTag', str(sess.portal_group_tag)))

    data = text_kv_bytes(text_items)
    b = sess.common_response_bhs(OP_LOGIN_RSP, req_bhs) if sess else bytearray(48)
    b[0] = OP_LOGIN_RSP
    b[1] = resp_flags & 0xFF
    b[2] = 0x00  # active version
    b[3] = 0x00
    b[8:14] = req_bhs[8:14]
    # Discovery sessions and pre-operational logins should not claim a non-zero
    # target session handle.  Assign the non-zero handle only after a normal
    # session is negotiating/entering full feature.
    tsih = 0
    if session_type.lower() == 'normal' and not security_stage:
        tsih = int(sess.tsih if sess else 1)
    b[14:16] = tsih.to_bytes(2, 'big')
    # Echo CID when present. Some initiators validate this field during login.
    b[20:22] = req_bhs[20:22]
    b[36] = 0x00  # status class: success
    b[37] = 0x00  # status detail: success
    if sess is None:
        itt = int.from_bytes(req_bhs[16:20], 'big')
        b[16:20] = itt.to_bytes(4, 'big')
        b[24:28] = (1).to_bytes(4, 'big')
        b[28:32] = (1).to_bytes(4, 'big')
        b[32:36] = (32).to_bytes(4, 'big')
    return bytes(b), data

def make_text_response(req_bhs: bytes, req_data: bytes, sess: IscsiSession) -> tuple[bytes, bytes]:
    kv = parse_text_kv(req_data)
    send_targets = kv.get('SendTargets')
    if send_targets is not None:
        host = sess.host
        port = sess.port
        # v156 proved Windows accepts the login sequence and reaches SendTargets,
        # but Windows rejects a tagless TargetAddress as invalid SendTargets
        # response text.  Use the RFC-style address form with an explicit
        # portal-group tag while keeping the record minimal: no TargetAlias and
        # a single TargetName/TargetAddress pair.
        data = text_kv_bytes([
            ('TargetName', sess.target_name),
            ('TargetAddress', f'{host}:{port},{sess.portal_group_tag}'),
        ])
    else:
        data = b''
    b = sess.common_response_bhs(OP_TEXT_RSP, req_bhs)
    b[1] = 0x80  # final
    b[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')  # Target Transfer Tag
    return bytes(b), data


def make_nop_in(req_bhs: bytes, req_data: bytes, sess: IscsiSession) -> tuple[bytes, bytes]:
    b = sess.common_response_bhs(OP_NOP_IN, req_bhs)
    b[1] = 0x80
    b[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')
    return bytes(b), req_data


def make_task_mgmt_response(req_bhs: bytes, sess: IscsiSession, response: int = 0x00) -> tuple[bytes, bytes]:
    b = sess.common_response_bhs(OP_TASK_MGMT_RSP, req_bhs)
    b[1] = 0x80
    b[2] = response & 0xFF
    return bytes(b), b''


def make_reject(req_bhs: bytes, sess: IscsiSession, reason: int = 0x04) -> tuple[bytes, bytes]:
    b = sess.common_response_bhs(OP_REJECT, req_bhs)
    b[1] = 0x80
    b[2] = reason & 0xFF
    return bytes(b), bytes(req_bhs)


def make_scsi_command(cdb: bytes, *, data_out: bytes = b'', itt: int = 2, exp_xfer_len: int = 0, force_write: bool = False) -> tuple[bytes, bytes]:
    if len(cdb) > 16:
        raise ValueError('this smoke bridge only supports CDBs up to 16 bytes')
    b = bytearray(48)
    b[0] = OP_SCSI_CMD
    b[1] = 0x80  # final, untagged/simple attribute for this smoke
    if data_out or force_write:
        b[1] |= 0x20  # W bit
    else:
        b[1] |= 0x40  # R bit; harmless for no-data commands in this smoke
    b[16:20] = int(itt).to_bytes(4, 'big')
    b[20:24] = int(exp_xfer_len).to_bytes(4, 'big')
    b[24:28] = int(itt).to_bytes(4, 'big')
    b[28:32] = (1).to_bytes(4, 'big')
    b[32:32+len(cdb)] = cdb
    return bytes(b), data_out


def make_scsi_data_out(itt: int, ttt: int, payload: bytes, *, final: bool = True, offset: int = 0, data_sn: int = 0) -> tuple[bytes, bytes]:
    b = bytearray(48)
    b[0] = OP_SCSI_DATA_OUT
    b[1] = 0x80 if final else 0x00
    b[16:20] = int(itt).to_bytes(4, 'big')
    b[20:24] = int(ttt).to_bytes(4, 'big')
    b[36:40] = int(data_sn).to_bytes(4, 'big')
    b[40:44] = int(offset).to_bytes(4, 'big')
    return bytes(b), payload


def make_r2t(req_bhs: bytes, sess: IscsiSession, *, ttt: int, offset: int, length: int) -> tuple[bytes, bytes]:
    b = bytearray(48)
    b[0] = OP_R2T
    b[16:20] = req_bhs[16:20]
    b[20:24] = int(ttt).to_bytes(4, 'big')
    b[24:28] = int(sess.stat_sn).to_bytes(4, 'big')
    b[28:32] = int(sess.exp_cmd_sn).to_bytes(4, 'big')
    b[32:36] = int(sess.max_cmd_sn).to_bytes(4, 'big')
    b[36:40] = (0).to_bytes(4, 'big')  # R2TSN
    b[40:44] = int(offset).to_bytes(4, 'big')
    b[44:48] = int(length).to_bytes(4, 'big')
    sess.stat_sn = (sess.stat_sn + 1) & 0xFFFFFFFF
    return bytes(b), b''


def make_scsi_data_in(req_bhs: bytes, payload: bytes, *, status: int = SCSI_STATUS_GOOD, sess: IscsiSession | None = None) -> tuple[bytes, bytes]:
    itt = int.from_bytes(req_bhs[16:20], 'big')
    exp_len = int.from_bytes(req_bhs[20:24], 'big')
    sent = payload if not exp_len else payload[:exp_len]
    residual = max(0, exp_len - len(sent)) if exp_len else 0
    overflow = bool(exp_len and len(payload) > exp_len)
    b = sess.common_response_bhs(OP_SCSI_DATA_IN, req_bhs) if sess else bytearray(48)
    b[0] = OP_SCSI_DATA_IN
    b[1] = 0x81  # final + status-present
    if residual:
        b[1] |= 0x02  # underflow
    if overflow:
        b[1] |= 0x04  # overflow
    b[3] = status & 0xFF
    b[16:20] = itt.to_bytes(4, 'big')
    if sess is None:
        b[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')
        b[24:28] = (1).to_bytes(4, 'big')
        b[28:32] = (1).to_bytes(4, 'big')
        b[32:36] = (32).to_bytes(4, 'big')
    else:
        b[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')
    b[36:40] = (0).to_bytes(4, 'big')
    b[40:44] = (0).to_bytes(4, 'big')
    b[44:48] = residual.to_bytes(4, 'big')
    return bytes(b), sent


def make_scsi_response(req_bhs: bytes, *, status: int = SCSI_STATUS_GOOD, sense: bytes = b'', sess: IscsiSession | None = None) -> tuple[bytes, bytes]:
    itt = int.from_bytes(req_bhs[16:20], 'big')
    data = (len(sense).to_bytes(2, 'big') + sense) if sense else b''
    b = sess.common_response_bhs(OP_SCSI_RSP, req_bhs) if sess else bytearray(48)
    b[0] = OP_SCSI_RSP
    b[1] = 0x80
    b[2] = 0x00  # response: command completed at target
    b[3] = status & 0xFF
    b[16:20] = itt.to_bytes(4, 'big')
    if sess is None:
        b[24:28] = (1).to_bytes(4, 'big')
        b[28:32] = (1).to_bytes(4, 'big')
        b[32:36] = (32).to_bytes(4, 'big')
    return bytes(b), data



def scsi_lun_bytes_from_bhs(req_bhs: bytes) -> bytes:
    # iSCSI SCSI Command BHS bytes 8..15 contain the SAM-2 encoded LUN.
    return bytes(req_bhs[8:16]) if len(req_bhs) >= 16 else b'\x00' * 8


def scsi_lun_is_zero(lun: bytes) -> bool:
    return not lun or all(b == 0 for b in lun)


def fixed_sense(*, sense_key: int = 0x05, asc: int = 0x20, ascq: int = 0x00) -> bytes:
    # Fixed-format current sense data, 18 bytes.
    b = bytearray.fromhex('70 00 00 00 00 00 00 0A 00 00 00 00 00 00 00 00 00 00')
    b[2] = int(sense_key) & 0x0F
    b[12] = int(asc) & 0xFF
    b[13] = int(ascq) & 0xFF
    return bytes(b)


def inquiry_no_lun_payload(cdb: bytes) -> bytes:
    alloc = cdb[4] if len(cdb) >= 5 else 36
    b = bytearray(36)
    # Peripheral Qualifier=3 (not supported), Peripheral Device Type=0x1F (no device).
    b[0] = 0x7F
    b[2] = 0x05
    b[3] = 0x02
    b[4] = 31
    return bytes(b[:alloc or 36])


def report_luns_payload(cdb: bytes) -> bytes:
    # One and only one exposed logical unit: LUN 0.  Windows was probing/creating
    # LUN 0..7 because earlier versions ignored the LUN field and answered every
    # INQUIRY as if it were LUN 0.  REPORT LUNS must make the singleton explicit.
    alloc = int.from_bytes(cdb[6:10], 'big') if len(cdb) >= 10 else 16
    payload = (8).to_bytes(4, 'big') + b'\x00' * 4 + b'\x00' * 8
    return payload[:alloc] if alloc else payload




def _u16(n: int) -> bytes:
    return int(n).to_bytes(2, 'big')


def _u32(n: int) -> bytes:
    return int(n).to_bytes(4, 'big')


def _alloc_len_6(cdb: bytes, default: int = 0) -> int:
    return int(cdb[4]) if len(cdb) >= 5 else default


def _alloc_len_10(cdb: bytes, default: int = 0) -> int:
    return int.from_bytes(cdb[7:9], 'big') if len(cdb) >= 9 else default


def _trunc(payload: bytes, alloc: int) -> bytes:
    return payload[:alloc] if alloc else payload


def _lba_to_msf(lba: int) -> bytes:
    # MMC MSF addresses are normally LBA+150 frames, packed as 00:MM:SS:FF
    v = max(0, int(lba) + 150)
    m, rem = divmod(v, 75 * 60)
    s2, f = divmod(rem, 75)
    return bytes([0, m & 0xFF, s2 & 0xFF, f & 0xFF])


def _toc_desc(ctrl_adr: int, track: int, lba: int, *, msf: bool) -> bytes:
    addr = _lba_to_msf(lba) if msf else _u32(lba)
    return bytes([0x00, ctrl_adr & 0xFF, track & 0xFF, 0x00]) + addr


def mmc_read_toc_payload(cdb: bytes, *, leadout_lba: int = 0x00345B60) -> bytes:
    """Conservative DVD-ROM READ TOC/PMA/ATIP response for Windows mount probes.

    v201 proved Windows probes READ TOC repeatedly. Returning CHECK CONDITION
    leaves the class driver with media present but unrecognizable.  These
    responses model a single-session, single-track DVD-ROM enough for mount
    discovery while READ payloads still come from the proven ISO window.
    """
    alloc = _alloc_len_10(cdb, 0)
    msf = bool(len(cdb) > 1 and (cdb[1] & 0x02))
    fmt = int(cdb[2] & 0x0F) if len(cdb) > 2 else 0
    # Format 1: multi-session information.  One session, first track 1.
    if fmt == 0x01:
        body = bytes([0x01, 0x01]) + _toc_desc(0x14, 0x01, 0, msf=msf)
        payload = _u16(len(body)) + body
        return _trunc(payload, alloc)
    # Format 2 raw TOC is not expected in the current Windows trace, but answer
    # with a small lead-in/current-session shape instead of CHECK CONDITION.
    if fmt == 0x02:
        body = bytes([0x01, 0x01]) + _toc_desc(0x14, 0x01, 0, msf=msf) + _toc_desc(0x14, 0xAA, leadout_lba, msf=msf)
        payload = _u16(len(body)) + body
        return _trunc(payload, alloc)
    # Format 0: formatted TOC.  The existing v165/v197 catalog used these same
    # first/last track values and lead-out LBA; v202 generalizes to allocation
    # lengths and the MSF bit used by Windows.
    body = bytes([0x01, 0x01]) + _toc_desc(0x14, 0x01, 0, msf=msf) + _toc_desc(0x14, 0xAA, leadout_lba, msf=msf)
    payload = _u16(len(body)) + body
    return _trunc(payload, alloc)


def mmc_read_disc_information_payload(cdb: bytes) -> bytes:
    # 34-byte read-only finalized disc information response retained from the
    # proven Windows attach catalog, generalized to the caller allocation.
    alloc = _alloc_len_10(cdb, 0)
    payload = bytes.fromhex('00 20 0E 01 01 01 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00')
    return _trunc(payload, alloc)


def mmc_get_event_status_payload(cdb: bytes) -> bytes:
    alloc = _alloc_len_10(cdb, 0)
    # Header says six bytes after the length field; media event class supported,
    # no outstanding event. Matches the v160-v197 catalog shape.
    payload = bytes.fromhex('00 06 00 10 00 00 00 00')
    return _trunc(payload, alloc)


def mmc_get_configuration_payload(cdb: bytes) -> bytes:
    alloc = _alloc_len_10(cdb, 0)
    # Current profile DVD-ROM (0x0010) with minimal feature list.  This is the
    # catalog response Windows already accepted when it requested RT=2/64 bytes;
    # v202 also answers RT=0/8-byte probes.
    payload = bytes.fromhex('00 00 00 28 00 00 00 10 00 00 03 08 00 08 00 00 00 10 01 00 00 01 03 00 00 03 03 04 29 00 00 00 00 10 03 08 00 00 08 00 00 00 00 00')
    return _trunc(payload, alloc)


def mmc_get_performance_payload(cdb: bytes) -> bytes:
    # GET PERFORMANCE is often used by DISCInfo/Windows as a capability probe.
    # We are not modelling rotational speed yet, so return the legal minimal
    # response header and no descriptors instead of CHECK CONDITION.  The
    # allocation length is bytes 8..9 in the 12-byte CDB.
    alloc = int.from_bytes(cdb[8:10], 'big') if len(cdb) >= 10 else 8
    payload = bytes.fromhex('00 00 00 04 00 00 00 00')
    return _trunc(payload, alloc)


def mmc_report_key_payload(cdb: bytes) -> bytes:
    """Minimal REPORT KEY response for DISCInfo/Windows CSS/RPC probing.

    The v207 live trace showed a single A4 probe:
    A4 00 00 00 00 00 00 00 00 08 08 00.  Byte 10 low bits are the
    report-key format; format 0x08 is the RPC-state query used by host tools.
    This emulator intentionally reports a region-free/read-only DVD-ROM surface
    for the GDR-8050L profile, so return a legal 8-byte, no-restriction shape
    instead of CHECK CONDITION.  Other key formats are answered with the same
    short no-CSS/no-RPC payload for now; firmware-backed CSS key semantics are
    a future target.
    """
    alloc = int.from_bytes(cdb[8:10], 'big') if len(cdb) >= 10 else 8
    key_format = cdb[10] & 0x3F if len(cdb) >= 11 else 0
    # Data length 6, followed by a conservative zeroed descriptor.  For the
    # observed format 0x08 this makes DISCInfo report RPC2 scheme NO, matching
    # the current emulator/GDR8050L behavior.
    payload = bytes.fromhex('00 06 00 00 00 00 00 00')
    return _trunc(payload, alloc)


def mmc_read_dvd_structure_probe_payload(cdb: bytes) -> bytes | None:
    """Answer benign standard READ DVD STRUCTURE probes not covered by catalog.

    Vendor/auth READ DVD STRUCTURE format C0 must continue to come from the
    session catalog/firmware-auth evidence.  On GDR-8050L auth CDBs the vendor
    selector is carried in byte 11 while byte 7 may still be 0x00, so this
    helper refuses CDBs with byte 11 == 0xC0 and lets the catalog/auth path
    answer them.
    """
    if len(cdb) < 10 or cdb[0] != 0xAD:
        return None
    fmt = cdb[7] & 0xFF
    vendor_selector = cdb[11] & 0xFF if len(cdb) >= 12 else 0
    alloc = int.from_bytes(cdb[8:10], 'big')
    if vendor_selector == 0xC0:
        return None
    if fmt == 0x01:
        # Copyright/RPC status probe: data length 6 + conservative zeroed body.
        payload = bytes.fromhex('00 06 00 00 00 00 00 00')
        return _trunc(payload, alloc)
    if fmt in (0x00, 0x04):
        # PFI/DMI standard probes. Prefer exact media-profile payloads in the
        # bridge method; this fallback is only for attach/probe tools when a
        # reduced profile is in use.  Use the normal 4-byte DVD-structure header
        # plus a zero-filled structure to avoid a spurious unsupported-command
        # failure without claiming firmware-derived semantics.
        body_len = 2048
        payload = int(body_len + 2).to_bytes(2, 'big') + b'\x00\x00' + (b'\x00' * body_len)
        return _trunc(payload, alloc)
    if fmt == 0x11:
        # DISCInfo/Windows may ask this as a benign protection/content probe.
        payload = bytes.fromhex('00 06 00 00 00 00 00 00')
        return _trunc(payload, alloc)
    return None


def _read_cd_raw_sector_from_user_data(user_data: bytes, lba: int) -> bytes:
    """Build a conservative 2352-byte raw CD-style sector wrapper.

    READ CD is being used here only as a host media-sniffing probe against a
    DVD-ROM/XDVD image. The backing ISO contains 2048-byte user sectors, not
    raw CD frames, so the wrapper supplies a normal sync/header shape, inserts
    the 2048 user bytes, and zero-fills the remaining subfields. This avoids a
    noisy unsupported-command CHECK CONDITION without claiming firmware-derived
    raw-CD behavior.
    """
    user = bytes(user_data[:2048]).ljust(2048, b'\x00')
    # 12-byte sync + 4-byte minute/second/frame/mode header.
    sync = b'\x00' + (b'\xFF' * 10) + b'\x00'
    msf = _lba_to_msf(lba)
    header = bytes([msf[1], msf[2], msf[3], 0x01])
    raw = sync + header + user
    return raw + (b'\x00' * max(0, 2352 - len(raw)))


def mmc_read_cd_probe_payload(cdb: bytes, *, read_user_sector) -> tuple[bytes, dict[str, Any]] | None:
    """Answer benign READ CD (0xBE) probe CDBs from the current visible view.

    This is deliberately a compatibility shim. It is not used for Xbox DVD
    authentication and does not promote any firmware READ-window evidence. The
    common observed CDB is `BE 00 00 00 00 00 00 00 01 F8 00 00`, a single-block
    LBA-0 probe while the drive is still locked/video-visible.
    """
    if len(cdb) < 12 or cdb[0] != 0xBE:
        return None
    host_lba = int.from_bytes(cdb[2:6], 'big')
    blocks = int.from_bytes(cdb[6:9], 'big')
    flags = cdb[9] & 0xFF
    if blocks <= 0:
        return b'', {'read_cd_host_lba': host_lba, 'read_cd_blocks': blocks, 'read_cd_flags': f'0x{flags:02X}', 'read_cd_probe_zero_blocks': True}
    # Keep this bounded; real READ payloads still use READ(10/12).  Host sniffers
    # use one or a few blocks. Returning too much here only increases attach noise.
    bounded_blocks = min(blocks, 16)
    out = bytearray()
    raw_like = bool(flags & 0xD8)  # sync/header/user/edc-ecc style request bits.
    for i in range(bounded_blocks):
        user_payload, media_info = read_user_sector(host_lba + i, 1)
        user_sector = bytes(user_payload[:2048]).ljust(2048, b'\x00')
        if raw_like:
            out += _read_cd_raw_sector_from_user_data(user_sector, host_lba + i)
        else:
            out += user_sector
    info = {
        'read_cd_host_lba': host_lba,
        'read_cd_blocks': blocks,
        'read_cd_bounded_blocks': bounded_blocks,
        'read_cd_flags': f'0x{flags:02X}',
        'read_cd_raw_like': raw_like,
        'read_cd_compatibility_probe': True,
        'firmware_shadow_lba_map_observed': False,
        'firmware_shadow_lba_map_source': 'read_cd_probe_not_firmware_window_proof',
    }
    if blocks > bounded_blocks:
        info['read_cd_blocks_truncated_for_probe_safety'] = True
    return bytes(out), info


def mmc_read_track_information_payload(cdb: bytes) -> bytes:
    """Minimal READ TRACK INFORMATION response for host probe compatibility."""
    alloc = int.from_bytes(cdb[7:9], 'big') if len(cdb) >= 9 else 36
    # 36-byte MMC-style Track Information Data response.  This says track 1,
    # session 1, data mode, no writable/free-block semantics. It is intentionally
    # conservative; writable-media behavior remains a future device-profile task.
    payload = bytearray(36)
    payload[0:2] = (34).to_bytes(2, 'big')
    payload[2] = 0x01   # track number lsb
    payload[3] = 0x01   # session number lsb
    payload[5] = 0x04   # data mode 1-ish/non-audio data track
    return _trunc(bytes(payload), alloc)


def mmc_set_streaming_good_payload(cdb: bytes, data_out: bytes = b'') -> bytes:
    """SET STREAMING is a no-data status command; accept benign tuning probes."""
    return b''

def mmc_mode_page_2a() -> bytes:
    return bytes.fromhex('2A 14 3F 00 75 7F 29 00 16 00 01 00 08 00 10 00 00 00 00 00 00 00')


def mmc_mode_sense10_payload(cdb: bytes) -> bytes:
    alloc = _alloc_len_10(cdb, 0)
    page_code = cdb[2] & 0x3F if len(cdb) > 2 else 0
    # Page 0x2A: MMC capabilities and mechanical status.  Windows asks for a
    # short 16-byte allocation during recognition.  Return a normal 10-byte
    # mode header and truncate the page as requested.
    if page_code == 0x2A:
        page = mmc_mode_page_2a()
        total = 8 + len(page)
        hdr = bytearray(8)
        hdr[0:2] = _u16(total - 2)
        payload = bytes(hdr) + page
        return _trunc(payload, alloc)
    if page_code == 0x3F:
        # MODE SENSE all-pages: return the safe, read-only MMC capabilities page
        # rather than failing the broad DISCInfo probe.  Firmware-backed page
        # synthesis remains a future target; this only prevents host tools from
        # treating the emulated device as broken during enumeration.
        page = mmc_mode_page_2a()
        total = 8 + len(page)
        hdr = bytearray(8)
        hdr[0:2] = _u16(total - 2)
        payload = bytes(hdr) + page
        return _trunc(payload, alloc)
    # Page 0x3E is already handled by the stateful auth/catalog path. Unknown
    # mode pages are best handled by the catalog matcher or CHECK CONDITION.
    return b''


def scsi_inquiry_payload_vpd(cdb: bytes) -> bytes:
    alloc = _alloc_len_6(cdb, 36)
    evpd = bool(len(cdb) > 1 and (cdb[1] & 0x01))
    page = int(cdb[2]) if len(cdb) > 2 else 0
    if not evpd:
        # Standard inquiry, same identity as the proven catalog.
        payload = bytes.fromhex('05 80 05 32 1F 00 00 00 48 4C 2D 44 54 2D 53 54 44 56 44 2D 52 4F 4D 20 47 44 52 38 30 35 30 4C 30 4C 32 33')
        return _trunc(payload, alloc)
    if page == 0x00:
        payload = bytes([0x05, 0x00, 0x00, 0x03, 0x00, 0x80, 0x83])
        return _trunc(payload, alloc)
    if page == 0x80:
        serial = b'MN103S-GDR8050L-EMU'
        payload = bytes([0x05, 0x80, 0x00, len(serial)]) + serial
        return _trunc(payload, alloc)
    if page == 0x83:
        ident = b'iqn.2026-06.local.mn103s:gdr8050l'
        desc = bytes([0x02, 0x01, 0x00, len(ident)]) + ident
        payload = bytes([0x05, 0x83]) + _u16(len(desc)) + desc
        return _trunc(payload, alloc)
    return b''

def make_lun_not_supported_row(index: int, tag: str, cdb: bytes, lun: bytes) -> dict[str, Any]:
    sense = fixed_sense(sense_key=0x05, asc=0x25, ascq=0x00)
    return {
        'ok': False,
        'index': index,
        'tag': tag,
        'scsi_status': 'CHECK CONDITION',
        'sense_key': 'ILLEGAL REQUEST',
        'sense_key_hex': '0x05',
        'asc': '0x25',
        'ascq': '0x00',
        'direction': 'none',
        'source': 'lun_guard',
        'cdb_hex': hx(cdb),
        'lun_hex': hx(lun),
        'data_len': 0,
        'data_sha256': '',
        'data_hex': '',
        'data_preview_hex': '',
        'sense_data_hex': hx(sense),
        'sense_data_sha256': hashlib.sha256(sense).hexdigest(),
        'error': 'logical unit not supported; v202 target exposes only LUN 0',
    }


def make_logout_request(itt: int = 0xFFFFFFFE) -> tuple[bytes, bytes]:
    b = bytearray(48)
    b[0] = OP_LOGOUT_REQ | 0x40
    b[1] = 0x80
    b[16:20] = int(itt).to_bytes(4, 'big')
    return bytes(b), b''


def make_logout_response(req_bhs: bytes, sess: IscsiSession | None = None) -> tuple[bytes, bytes]:
    itt = int.from_bytes(req_bhs[16:20], 'big')
    b = sess.common_response_bhs(OP_LOGOUT_RSP, req_bhs) if sess else bytearray(48)
    b[0] = OP_LOGOUT_RSP
    b[1] = 0x80
    b[2] = 0x00
    b[16:20] = itt.to_bytes(4, 'big')
    if sess is None:
        b[24:28] = (2).to_bytes(4, 'big')
        b[28:32] = (2).to_bytes(4, 'big')
        b[32:36] = (32).to_bytes(4, 'big')
    return bytes(b), b''


def pdu_name(op: int) -> str:
    return {
        OP_NOP_IN: 'NOP_IN', OP_SCSI_RSP: 'SCSI_RESPONSE', OP_TASK_MGMT_RSP: 'TASK_MGMT_RESPONSE',
        OP_LOGIN_RSP: 'LOGIN_RESPONSE', OP_TEXT_RSP: 'TEXT_RESPONSE', OP_LOGOUT_RSP: 'LOGOUT_RESPONSE',
        OP_SCSI_DATA_IN: 'SCSI_DATA_IN', OP_R2T: 'R2T', OP_REJECT: 'REJECT',
    }.get(op, f'0x{op:02X}')


def script_to_host_req(item: dict[str, Any], index: int) -> dict[str, Any]:
    out = {'tag': item.get('tag') or item.get('title') or f'iscsi_cmd_{index:03d}', 'cdb_hex': hx(parse_hex_bytes(item.get('cdb_hex', item.get('cdb'))))}
    dout = parse_hex_bytes(item.get('data_out_hex', item.get('data_out')))
    if dout:
        out['data_out_hex'] = hx(dout)
    return out


def row_to_scsi_pdus(req_bhs: bytes, row: dict[str, Any], sess: IscsiSession | None = None) -> tuple[bytes, bytes]:
    if row.get('scsi_status') == 'GOOD':
        direction = str(row.get('direction','none')).lower()
        payload = parse_hex_bytes(row.get('data_hex','')) if row.get('data_hex') else b''
        if direction == 'in' and payload:
            return make_scsi_data_in(req_bhs, payload, status=SCSI_STATUS_GOOD, sess=sess)
        return make_scsi_response(req_bhs, status=SCSI_STATUS_GOOD, sess=sess)
    sense = parse_hex_bytes(row.get('sense_data_hex','')) if row.get('sense_data_hex') else b''
    return make_scsi_response(req_bhs, status=SCSI_STATUS_CHECK_CONDITION, sense=sense, sess=sess)


def process_scsi_command(bridge: HostBridge, req_bhs: bytes, cdb: bytes, data_out: bytes, index: int, rows_sink: list[dict[str, Any]], tag_prefix: str = 'iscsi') -> tuple[bytes, bytes]:
    tag = f'{tag_prefix}_{index:03d}_{cdb[0]:02x}' if cdb else f'{tag_prefix}_{index:03d}'
    req = {'tag': tag, 'cdb_hex': hx(cdb)}
    if data_out:
        req['data_out_hex'] = hx(data_out)
    row = bridge.response_for(req, index)
    rows_sink.append(row)
    return row_to_scsi_pdus(req_bhs, row, None)


def append_protocol_row(rows_sink: list[dict[str, Any]], req_bhs: bytes, req_data: bytes, rsp_bhs: bytes, rsp_data: bytes, tag: str, extra: dict[str, Any] | None = None) -> None:
    final_rsp_bhs = prepare_pdu_bhs(rsp_bhs, rsp_data)
    row: dict[str, Any] = {
        'index': -1,
        'tag': tag,
        'request_opcode': int(req_bhs[0] & 0x3F),
        'response_opcode': int(final_rsp_bhs[0] & 0x3F),
        'pdu_opcode': pdu_name(final_rsp_bhs[0] & 0x3F),
        'request_flags_hex': f'{req_bhs[1]:02x}',
        'response_flags_hex': f'{final_rsp_bhs[1]:02x}',
        'request_ahs_len': bhs_ahs_len(req_bhs),
        'request_bhs_data_len': bhs_data_len(req_bhs),
        'request_data_len': len(req_data),
        'response_ahs_len': bhs_ahs_len(final_rsp_bhs),
        'response_bhs_data_len': bhs_data_len(final_rsp_bhs),
        'response_data_len': len(rsp_data),
        'ok': (final_rsp_bhs[0] & 0x3F) != OP_REJECT,
        'request_bhs_hex': hx(req_bhs),
        'response_bhs_hex': hx(final_rsp_bhs),
    }
    if req_data:
        row['request_text'] = req_data.decode('utf-8', errors='replace').replace('\x00', '|')
    if rsp_data:
        row['response_text'] = rsp_data.decode('utf-8', errors='replace').replace('\x00', '|') if (final_rsp_bhs[0] & 0x3F) in (OP_LOGIN_RSP, OP_TEXT_RSP) else ''
        row['response_data_sha256'] = hashlib.sha256(rsp_data).hexdigest()
    if extra:
        row.update(extra)
    rows_sink.append(row)


def serve_client(conn: socket.socket, bridge: HostBridge, rows_sink: list[dict[str, Any]], *, sess: IscsiSession | None = None, lock: threading.Lock | None = None) -> None:
    sess = sess or IscsiSession('iqn.2026-06.local.mn103s:gdr8050l', '127.0.0.1', 3260)
    lock = lock or threading.Lock()
    login_pdus = 0
    with conn:
        while True:
            try:
                bhs, data = recv_pdu(conn)
            except EOFError:
                break
            op = bhs[0] & 0x3F
            if op == OP_LOGIN_REQ:
                login_pdus += 1
                if login_pdus > 256:
                    rows_sink.append({
                        'index': -1,
                        'tag': 'iscsi_login_guard_close',
                        'ok': False,
                        'error': 'more than 256 Login Request PDUs on one client session; closing to prevent unbounded Windows login loop',
                        'last_request_flags_hex': f'{bhs[1]:02x}',
                        'last_request_text': data.decode('utf-8', errors='replace').replace('\x00', '|') if data else '',
                    })
                    break
                rb, rd = make_login_response(bhs, data, sess)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_login', {'session_type': sess.session_type, 'login_stage': sess.login_stage})
                send_pdu(conn, rb, rd)
                continue
            if op == OP_TEXT_REQ:
                rb, rd = make_text_response(bhs, data, sess)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_text')
                send_pdu(conn, rb, rd)
                continue
            if op == OP_NOP_OUT:
                rb, rd = make_nop_in(bhs, data, sess)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_nop')
                send_pdu(conn, rb, rd)
                continue
            if op == OP_TASK_MGMT_REQ:
                rb, rd = make_task_mgmt_response(bhs, sess)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_task_mgmt')
                send_pdu(conn, rb, rd)
                continue
            if op == OP_LOGOUT_REQ:
                rb, rd = make_logout_response(bhs, sess)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_logout')
                send_pdu(conn, rb, rd)
                break
            if op == OP_SCSI_DATA_OUT:
                itt = int.from_bytes(bhs[16:20], 'big')
                pending = sess.pending_data_out.get(itt)
                if pending is None:
                    rb, rd = make_reject(bhs, sess, reason=0x09)
                    append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_reject_unexpected_data_out')
                    send_pdu(conn, rb, rd)
                    continue
                offset = int.from_bytes(bhs[40:44], 'big')
                buf = bytearray(pending['buf'])
                if offset > len(buf):
                    buf += b'\x00' * (offset - len(buf))
                end = offset + len(data)
                if end > len(buf):
                    buf += b'\x00' * (end - len(buf))
                buf[offset:end] = data
                pending['buf'] = bytes(buf)
                final = bool(bhs[1] & 0x80)
                expected = int(pending.get('expected') or 0)
                if final or (expected and len(buf) >= expected):
                    with lock:
                        row = bridge.response_for({'tag': pending['tag'], 'cdb_hex': hx(pending['cdb']), 'data_out_hex': hx(bytes(buf))}, int(pending['index']))
                    rows_sink.append(row)
                    rb, rd = row_to_scsi_pdus(pending['req_bhs'], row, sess)
                    sess.pending_data_out.pop(itt, None)
                    send_pdu(conn, rb, rd)
                continue
            if op != OP_SCSI_CMD:
                rb, rd = make_reject(bhs, sess, reason=0x04)
                append_protocol_row(rows_sink, bhs, data, rb, rd, 'iscsi_reject_unsupported')
                send_pdu(conn, rb, rd)
                continue

            op0 = int(bhs[32])
            clen = cdb_len_from_opcode(op0)
            cdb = bytes(bhs[32:32+clen])
            itt = int.from_bytes(bhs[16:20], 'big')
            expected = int.from_bytes(bhs[20:24], 'big') or expected_xfer_from_cdb(cdb)
            write_bit = bool(bhs[1] & 0x20)
            cmd_index = sess.alloc_command_index()
            tag = f'iscsi_{cmd_index:03d}_{cdb[0]:02x}' if cdb else f'iscsi_{cmd_index:03d}'
            lun = scsi_lun_bytes_from_bhs(bhs)
            if cdb and cdb[0] == 0xA0 and scsi_lun_is_zero(lun):
                row = {
                    'ok': True,
                    'index': cmd_index,
                    'tag': tag,
                    'scsi_status': 'GOOD',
                    'sense_key': 'NO SENSE',
                    'sense_key_hex': '0x00',
                    'asc': '0x00',
                    'ascq': '0x00',
                    'direction': 'in',
                    'source': 'v202_report_luns_singleton',
                    'cdb_hex': hx(cdb),
                    'lun_hex': hx(lun),
                    'data_hex': hx(report_luns_payload(cdb)),
                    'data_len': len(report_luns_payload(cdb)),
                    'data_preview_hex': hx(report_luns_payload(cdb)),
                    'data_sha256': hashlib.sha256(report_luns_payload(cdb)).hexdigest(),
                    'reported_luns': 1,
                }
                rows_sink.append(row)
                rb, rd = row_to_scsi_pdus(bhs, row, sess)
                send_pdu(conn, rb, rd)
                continue
            if not scsi_lun_is_zero(lun):
                if cdb and cdb[0] == 0x12:
                    payload = inquiry_no_lun_payload(cdb)
                    row = {
                        'ok': True,
                        'index': cmd_index,
                        'tag': tag,
                        'scsi_status': 'GOOD',
                        'sense_key': 'NO SENSE',
                        'sense_key_hex': '0x00',
                        'asc': '0x00',
                        'ascq': '0x00',
                        'direction': 'in',
                        'source': 'v202_inquiry_no_device_for_nonzero_lun',
                        'cdb_hex': hx(cdb),
                        'lun_hex': hx(lun),
                        'data_hex': hx(payload),
                        'data_len': len(payload),
                        'data_preview_hex': hx(payload),
                        'data_sha256': hashlib.sha256(payload).hexdigest(),
                        'reported_no_device': True,
                    }
                else:
                    row = make_lun_not_supported_row(cmd_index, tag, cdb, lun)
                rows_sink.append(row)
                rb, rd = row_to_scsi_pdus(bhs, row, sess)
                send_pdu(conn, rb, rd)
                continue
            if write_bit and not data and expected:
                ttt = sess.alloc_ttt()
                sess.pending_data_out[itt] = {'req_bhs': bhs, 'cdb': cdb, 'buf': b'', 'expected': expected, 'index': cmd_index, 'tag': tag, 'ttt': ttt}
                rb, rd = make_r2t(bhs, sess, ttt=ttt, offset=0, length=expected)
                send_pdu(conn, rb, rd)
                continue
            with lock:
                row = bridge.response_for({'tag': tag, 'cdb_hex': hx(cdb), **({'data_out_hex': hx(data)} if data else {})}, cmd_index)
            rows_sink.append(row)
            rb, rd = row_to_scsi_pdus(bhs, row, sess)
            send_pdu(conn, rb, rd)


def client_roundtrip(host: str, port: int, items: list[dict[str, Any]], timeout: float, *, exercise_r2t: bool = False) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    with socket.create_connection((host, port), timeout=timeout) as sock:
        sock.settimeout(timeout)
        # Two-stage login smoke: SecurityNegotiation -> LoginOperationalNegotiation -> FullFeature.
        lb, ld = make_login_request(itt=1, session_type='Normal', csg=0, nsg=1, cmd_sn=1)
        send_pdu(sock, lb, ld)
        rb, rd = recv_pdu(sock)
        rows.append({'index': -1, 'tag': 'iscsi_login_security', 'pdu_opcode': pdu_name(rb[0] & 0x3F), 'flags_hex': f'{rb[1]:02x}', 'ok': (rb[0] & 0x3F) == OP_LOGIN_RSP and (rb[1] & 0x8f) == 0x81, 'data_len': len(rd), 'data_text': rd.decode('utf-8', errors='replace').replace('\x00','|')})
        lb, ld = make_login_request(itt=1, session_type='Normal', csg=1, nsg=3, cmd_sn=2)
        send_pdu(sock, lb, ld)
        rb, rd = recv_pdu(sock)
        rows.append({'index': -1, 'tag': 'iscsi_login_operational', 'pdu_opcode': pdu_name(rb[0] & 0x3F), 'flags_hex': f'{rb[1]:02x}', 'ok': (rb[0] & 0x3F) == OP_LOGIN_RSP and (rb[1] & 0x8f) == 0x87, 'data_len': len(rd), 'data_text': rd.decode('utf-8', errors='replace').replace('\x00','|')})
        # Discovery Text smoke proves the server has the SendTargets surface the
        # Windows initiator will ask for before a normal session.
        tb = bytearray(48)
        tb[0] = OP_TEXT_REQ | 0x40
        tb[1] = 0x80
        tb[16:20] = (0xABCDEF01).to_bytes(4, 'big')
        tb[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')
        send_pdu(sock, bytes(tb), text_kv_bytes({'SendTargets': 'All'}))
        rb, rd = recv_pdu(sock)
        rows.append({'index': -1, 'tag': 'iscsi_sendtargets', 'pdu_opcode': pdu_name(rb[0] & 0x3F), 'ok': (rb[0] & 0x3F) == OP_TEXT_RSP and b'TargetName=' in rd, 'data_len': len(rd), 'data_text': rd.decode('utf-8', errors='replace').replace('\x00','|')})
        nb = bytearray(48)
        nb[0] = OP_NOP_OUT | 0x40
        nb[1] = 0x80
        nb[16:20] = (0xABCDEF02).to_bytes(4, 'big')
        nb[20:24] = (0xFFFFFFFF).to_bytes(4, 'big')
        send_pdu(sock, bytes(nb), b'ping')
        rb, rd = recv_pdu(sock)
        rows.append({'index': -1, 'tag': 'iscsi_nop', 'pdu_opcode': pdu_name(rb[0] & 0x3F), 'ok': (rb[0] & 0x3F) == OP_NOP_IN and rd == b'ping', 'data_len': len(rd)})
        for i, item in enumerate(items):
            cdb = parse_hex_bytes(item.get('cdb_hex', item.get('cdb')))
            dout = parse_hex_bytes(item.get('data_out_hex', item.get('data_out')))
            exp = expected_xfer_from_cdb(cdb)
            use_r2t = bool(exercise_r2t and dout and i == 3)
            sb, sd = make_scsi_command(cdb, data_out=(b'' if use_r2t else dout), itt=i+2, exp_xfer_len=exp, force_write=use_r2t)
            send_pdu(sock, sb, sd)
            rb, rd = recv_pdu(sock)
            if (rb[0] & 0x3F) == OP_R2T:
                ttt = int.from_bytes(rb[20:24], 'big')
                db, dd = make_scsi_data_out(i+2, ttt, dout, final=True, offset=0, data_sn=0)
                send_pdu(sock, db, dd)
                rb, rd = recv_pdu(sock)
            rop = rb[0] & 0x3F
            status = rb[3]
            row: dict[str, Any] = {
                'index': i,
                'tag': item.get('tag') or item.get('title') or f'iscsi_cmd_{i:03d}',
                'cdb_hex': hx(cdb),
                'data_out_len': len(dout),
                'pdu_opcode': pdu_name(rop),
                'scsi_status': 'GOOD' if status == 0 else 'CHECK CONDITION',
                'scsi_status_byte': status,
                'ok': status == 0,
                'data_len': 0,
                'data_sha256': '',
                'sense_len': 0,
                'sense_sha256': '',
                'used_r2t': use_r2t,
            }
            if rop == OP_SCSI_DATA_IN:
                row['data_len'] = len(rd)
                row['data_sha256'] = hashlib.sha256(rd).hexdigest() if rd else ''
                row['data_preview_hex'] = hx(rd[:64])
            elif rop == OP_SCSI_RSP and status == SCSI_STATUS_CHECK_CONDITION and len(rd) >= 2:
                slen = int.from_bytes(rd[:2], 'big')
                sense = rd[2:2+slen]
                row['sense_len'] = len(sense)
                row['sense_sha256'] = hashlib.sha256(sense).hexdigest() if sense else ''
                row['sense_preview_hex'] = hx(sense)
            rows.append(row)
        ob, od = make_logout_request()
        send_pdu(sock, ob, od)
        rb, rd = recv_pdu(sock)
        rows.append({'index': len(items), 'tag': 'iscsi_logout', 'pdu_opcode': pdu_name(rb[0] & 0x3F), 'ok': (rb[0] & 0x3F) == OP_LOGOUT_RSP, 'data_len': len(rd)})
    return rows


def safe_tag_name(tag: Any, index: int) -> str:
    text = str(tag or f'row_{index:03d}')
    out = ''.join(ch if ch.isalnum() or ch in ('-', '_', '.') else '_' for ch in text)
    return out[:120] or f'row_{index:03d}'


def write_outputs(rows: list[dict[str, Any]], out_path: Path | None, response_dir: Path | None, payload_rows: list[dict[str, Any]] | None = None, *, compact_data_hex: bool = False) -> None:
    payload_rows = payload_rows or rows
    if response_dir:
        response_dir.mkdir(parents=True, exist_ok=True)
        for i, r in enumerate(payload_rows):
            data = parse_hex_bytes(r.get('data_hex','')) if r.get('data_hex') else b''
            if data:
                fn = f"{int(r.get('index', i) or 0):03d}_{safe_tag_name(r.get('tag'), i)}.bin"
                (response_dir / fn).write_bytes(data)
                r['data_file'] = fn
                if compact_data_hex:
                    r['data_hex_omitted'] = True
                    r['data_hex'] = ''
            sense = parse_hex_bytes(r.get('sense_data_hex','')) if r.get('sense_data_hex') else b''
            if sense:
                fn = f"{int(r.get('index', i) or 0):03d}_{safe_tag_name(r.get('tag'), i)}.sense.bin"
                (response_dir / fn).write_bytes(sense)
                r['sense_file'] = fn
                if compact_data_hex:
                    r['sense_data_hex_omitted'] = True
                    r['sense_data_hex'] = ''
    if out_path:
        out_path.write_text(''.join(json.dumps(r, separators=(',', ':')) + '\n' for r in rows), encoding='utf-8')


def write_verify(path: Path, rows: list[dict[str, Any]], host_rows: list[dict[str, Any]]) -> None:
    cmd_rows = [r for r in rows if isinstance(r.get('index'), int) and int(r.get('index')) >= 0 and str(r.get('tag')) != 'iscsi_logout']
    ok = sum(1 for r in cmd_rows if r.get('ok'))
    cc = sum(1 for r in cmd_rows if r.get('scsi_status') == 'CHECK CONDITION')
    data = sum(1 for r in cmd_rows if int(r.get('data_len') or 0) > 0)
    data_out = sum(1 for r in cmd_rows if int(r.get('data_out_len') or 0) > 0)
    host_cmd_rows = [r for r in host_rows if isinstance(r.get('index'), int) and int(r.get('index')) >= 0]
    catalog = sum(1 for r in host_cmd_rows if r.get('catalog_match'))
    dynamic_media = sum(1 for r in host_cmd_rows if r.get('dynamic_media_window') or r.get('source') == 'dynamic_media_window')
    protocol_ok = all(r.get('ok') for r in rows if int(r.get('index', -2)) < 0)
    r2t_seen = sum(1 for r in cmd_rows if r.get('used_r2t'))
    exact = (protocol_ok and len(cmd_rows) == len(host_cmd_rows) and ok + cc == len(cmd_rows) and all((hr.get('scsi_status') == cr.get('scsi_status')) for hr,cr in zip(host_cmd_rows, cmd_rows)))
    lines = ['# GDR iSCSI PDU bridge smoke verification','', f'requests={len(cmd_rows)}', f'ok={ok}', f'check_condition={cc}', f'data_in_pdus={data}', f'data_out_pdus={data_out}', f'catalog_matches={catalog}', f'dynamic_media_window_reads={dynamic_media}', f'protocol_negotiation_ok={str(protocol_ok).lower()}', f'r2t_data_out_paths={r2t_seen}', f'exact_iscsi_pdu_smoke={str(exact).lower()}', '']
    lines.append('| index | tag | pdu | status | data_out | data_in | sense | r2t | catalog |')
    lines.append('|---:|---|---|---|---:|---:|---:|---|---|')
    for i, r in enumerate(cmd_rows):
        hr = host_cmd_rows[i] if i < len(host_cmd_rows) else {}
        lines.append(f"| {r.get('index')} | `{r.get('tag','')}` | {r.get('pdu_opcode','')} | {r.get('scsi_status','')} | {r.get('data_out_len',0)} | {r.get('data_len',0)} | {r.get('sense_len',0)} | {str(bool(r.get('used_r2t'))).lower()} | {str(bool(hr.get('catalog_match'))).lower()} |")
    path.write_text('\n'.join(lines) + '\n', encoding='utf-8')
    if not exact:
        raise SystemExit(2)




LOCKED_MAX_LBA = 0x00001B4F
UNLOCKED_MAX_LBA = 0x00345B5F
LOCKED_READ_CAPACITY_10 = bytes.fromhex('00 00 1B 4F 00 00 08 00')
UNLOCKED_READ_CAPACITY_10 = bytes.fromhex('00 34 5B 5F 00 00 08 00')
MODE_3E_LOCKED = bytes.fromhex('00 1A 00 00 00 00 00 00 3E 12 00 01 00 D1 02 DC A2 E7 C8 ED 33 28 22 00 00 00 00 00')
MODE_3E_VERIFY1 = bytes.fromhex('00 1A 00 00 00 00 00 00 3E 12 00 01 01 D1 02 6B A2 E7 C8 ED D2 D1 70 F1 00 00 00 00')
MODE_3E_VERIFY2 = bytes.fromhex('00 1A 00 00 00 00 00 00 3E 12 00 01 01 D1 02 DC A2 E7 C8 ED 33 28 22 00 00 00 00 00')
MODE_3E_UNLOCKED = bytes.fromhex('00 1A 00 00 00 00 00 00 3E 12 01 01 01 D1 02 DC A2 E7 C8 ED 33 28 22 00 00 00 00 00')


def is_mode_sense_3e(cdb: bytes) -> bool:
    return bool(cdb) and ((cdb[0] == 0x5A and len(cdb) >= 3 and (cdb[2] & 0x3F) == 0x3E) or (cdb[0] == 0x1A and len(cdb) >= 3 and (cdb[2] & 0x3F) == 0x3E))


def mode3e_truncate(cdb: bytes, payload10: bytes) -> bytes:
    if cdb and cdb[0] == 0x1A:
        # Convert MODE SENSE(10) page shape to MODE SENSE(6) shape.
        page = payload10[8:]
        payload = bytes([len(page) + 3, 0, 0, 0]) + page
        alloc = cdb[4] if len(cdb) >= 5 else len(payload)
        return _trunc(payload, alloc)
    alloc = _alloc_len_10(cdb, len(payload10)) if cdb and cdb[0] == 0x5A else len(payload10)
    return _trunc(payload10, alloc)


def classify_8050l_mode_select(cdb: bytes, data_out: bytes) -> str:
    if not cdb or not data_out or cdb[0] not in (0x15, 0x55):
        return ''
    # MODE SELECT(6) sticky descramble payload: 00 00 00 00 31 06 01 ...
    if cdb[0] == 0x15 and len(data_out) >= 7 and data_out[4] == 0x31 and data_out[6] == 0x01:
        return 'sticky_descramble_mode31'
    # MODE SELECT(10) 3E page payload begins after the 8-byte mode parameter header.
    if cdb[0] == 0x55 and len(data_out) >= 20 and data_out[8] == 0x3E:
        # Known sequence from the unlock tool / catalog:
        # chal1: page bytes 00 01 00 D1 01 6B ...
        # chal2: page bytes 00 00 01 00 00 DC ...
        # unlock: page bytes 01 01 01 D1 01 DC ...
        p = data_out[8:]
        if len(p) >= 12:
            if p[2] == 0x00 and p[3] == 0x01 and p[4] == 0x00 and p[6] == 0x01 and p[7] == 0x6B:
                return 'challenge1'
            if p[2] == 0x00 and p[3] == 0x00 and p[4] == 0x01 and p[5] == 0x00 and p[7] == 0xDC:
                return 'challenge2'
            if p[2] == 0x01 and p[3] == 0x01 and p[4] == 0x01 and p[6] == 0x01 and p[7] == 0xDC:
                return 'partition1_unlock'
    return ''


def classify_start_stop_unit(cdb: bytes) -> str:
    if not cdb or cdb[0] != 0x1B or len(cdb) < 5:
        return ''
    pc = cdb[4]
    loej = bool(pc & 0x02)
    start = bool(pc & 0x01)
    if loej and not start:
        return 'eject'
    if loej and start:
        return 'load'
    return 'start' if start else 'nop'

class DynamicMediaWindowBridge:
    """HostBridge-compatible backend for v198-v213 live iSCSI use.

    It resolves control/auth/probe CDBs from the firmware-session catalog and
    resolves READ(10)/READ(12) dynamically from the selected ISO using the
    proven media window.  It intentionally does not instantiate GDRScsiParser
    and it refuses stale/fallback firmware provenance through the v197/v198
    audit logic.
    """

    def __init__(self, cfg: dict[str, Any], *, media_path: Path, media_lba_delta: int,
                 sector_size: int = 2048, model_path: str = 'trace.gdr_packet_session_model',
                 expect_firmware: str = CANONICAL_GDR8050L_0L23_BASENAME,
                 sequential_catalog: bool = True, truncate_to_alloc: bool = True,
                 unsupported_check_condition: bool = True,
                 view_mode: str = 'direct',
                 eject_policy: str = 'real'):
        self.cfg = cfg
        self.media_path = Path(media_path)
        self.media_lba_delta = int(media_lba_delta)
        self.sector_size = int(sector_size)
        self.model_path = model_path
        self.sequential_catalog = bool(sequential_catalog)
        self.truncate_to_alloc = bool(truncate_to_alloc)
        self.unsupported_check_condition = bool(unsupported_check_condition)
        self.view_mode = str(view_mode or 'direct')
        self.eject_policy = str(eject_policy or 'real')
        self.auth_stage = 'locked'
        self.auth_phase = 'idle'
        self.sticky_descramble = False
        self.tray_state = 'closed'
        self.media_present = True
        self.state_events: list[dict[str, Any]] = []
        # v213: non-authoritative firmware-shadow observer scaffold for the
        # 8050L MODE SELECT/SENSE 0x3E auth state.  This deliberately does not
        # drive live behavior yet; it records what the future firmware-in-loop
        # replacement must agree with before becoming authoritative.
        self.fw_shadow3e_enabled = True
        self.fw_shadow3e_stage = 'locked'
        self.fw_shadow3e_phase = 'idle'
        self.fw_shadow3e_sticky = False
        self.fw_shadow3e_events: list[dict[str, Any]] = []
        self.provenance = dynamic_enforce_config_provenance(cfg, expect_firmware=expect_firmware)
        if not self.provenance.get('ok'):
            raise RuntimeError('dynamic media firmware provenance failed: ' + '; '.join(self.provenance.get('failures') or []))
        if not self.media_path.exists():
            raise FileNotFoundError(f'dynamic media ISO not found: {self.media_path}')
        model = dynamic_get_session_model(cfg, model_path)
        self.catalog = dynamic_build_catalog(model.get('commands', []))
        self.cursor = 0
        self.media_sector_count = dynamic_read_capacity_of_file(self.media_path, self.sector_size)
        self.media_model = self._load_media_model()

    def _load_media_model(self) -> GDRMediaModel | None:
        # v227: prefer the corrected dumper's adjacent .xdvd.json + .media.json
        # pair when available.  The .xdvd sidecar supplies auth/MMC payloads;
        # the .media profile supplies accurate Redump-style sector layout.
        candidates = []
        try:
            candidates.append(self.media_path.with_suffix('.xdvd.json'))
        except Exception:
            pass
        try:
            candidates.append(self.media_path.with_suffix('.media.json'))
        except Exception:
            pass
        for cand in candidates:
            if cand and cand.exists():
                try:
                    return GDRMediaModel.from_paths(cand, self.media_path)
                except Exception as exc:
                    # Do not fail a known-good old session because media-model
                    # enrichment failed.  The legacy configured delta path below
                    # remains available.
                    self.state_events.append({'event': 'media_model_load_failed', 'path': str(cand), 'error': str(exc)})
        return None

    def _effective_unlocked_delta_hint(self) -> tuple[int, str]:
        """Return the currently preferred unlocked READ delta for trace/reporting.

        The command-line --dynamic-media-lba-delta remains the legacy contiguous
        fallback.  When a validated Redump-style .media.json profile is present,
        actual READ mapping is driven by GDRMediaModel.read_host(); for unlocked
        Xbox view that maps host LBA 0 to game_output_start_lba.  v229 exposes
        both values so logs no longer imply the old +306080 fallback is the live
        Redump mapping.
        """
        try:
            if self.media_model is not None and getattr(self.media_model, 'is_redump_style', False):
                game_start = int(getattr(self.media_model, 'game_output_start_lba', 0) or 0)
                if game_start:
                    return game_start, 'media_profile_redump_game_output_start_lba'
        except Exception:
            pass
        return int(getattr(self, 'media_lba_delta', 0) or 0), 'legacy_configured_media_lba_delta'

    def _media_delta_trace_fields(self) -> dict[str, Any]:
        hint, source = self._effective_unlocked_delta_hint()
        return {
            'dynamic_media_legacy_configured_lba_delta': int(getattr(self, 'media_lba_delta', 0) or 0),
            'dynamic_media_effective_unlocked_lba_delta_hint': hint,
            'dynamic_media_effective_unlocked_lba_delta_source': source,
            'firmware_shadow_lba_map_configured_delta': hint,
            'firmware_shadow_lba_map_legacy_configured_delta': int(getattr(self, 'media_lba_delta', 0) or 0),
            'firmware_shadow_lba_map_delta_source': source,
        }

    def _read_media_stateful(self, host_lba: int, blocks: int) -> tuple[bytes, dict[str, Any]]:
        if self.media_model is not None and getattr(self.media_model, 'is_redump_style', False):
            payload, info = self.media_model.read_host(self.media_path, self.auth_stage, host_lba, blocks, read_map='firmware_window')
            return payload, info
        effective_delta = self.media_lba_delta
        if self.view_mode == 'stateful_xbox' and self.auth_stage != 'unlocked':
            effective_delta = 0
        media_lba = host_lba + effective_delta
        payload = dynamic_read_media_window(self.media_path, media_lba, blocks, self.sector_size)
        return payload, {
            'media_lba': media_lba,
            'media_lba_end': media_lba + max(0, int(blocks) - 1),
            'effective_media_lba_delta': effective_delta,
            'read_map_note': f'media_lba=host_lba+{effective_delta}',
            'media_segment_count': 1,
            'read_segments': [{'host_lba': host_lba, 'iso_lba': media_lba, 'blocks': blocks, 'segment_note': 'legacy_configured_delta'}],
            'media_layout_type': getattr(self.media_model, 'layout_type', '') if self.media_model is not None else '',
            'media_is_redump_style': False,
            'media_profile_loaded': bool(getattr(self.media_model, 'media_profile_loaded', False)) if self.media_model is not None else False,
            'media_profile_path': str(getattr(self.media_model, 'media_profile_path', '') or '') if self.media_model is not None else '',
        }

    def _good(self, index: int, tag: str, cdb: bytes, direction: str, payload: bytes, *, source: str, extra: dict[str, Any] | None = None) -> dict[str, Any]:
        row: dict[str, Any] = {
            'ok': True,
            'index': index,
            'tag': tag,
            'scsi_status': 'GOOD',
            'sense_key': 'NO SENSE',
            'sense_key_hex': '0x00',
            'asc': '0x00',
            'ascq': '0x00',
            'direction': direction,
            'source': source,
            'cdb_hex': hx(cdb),
            'data_len': len(payload),
            'data_sha256': hashlib.sha256(payload).hexdigest() if payload else '',
            'data_hex': hx(payload),
            'data_preview_hex': hx(payload[:64]),
            'dynamic_media_lba_delta': self.media_lba_delta,
            'dynamic_media_configured_lba_delta': self.media_lba_delta,
            'dynamic_media_view_mode': self.view_mode,
            'dynamic_media_eject_policy': self.eject_policy,
            'dynamic_media_auth_stage': self.auth_stage,
            'dynamic_media_sticky_descramble': bool(self.sticky_descramble),
            'dynamic_media_tray_state': self.tray_state,
            'dynamic_media_present': bool(self.media_present),
            'dynamic_media_state_event_count': len(self.state_events),
            'dynamic_media_path': str(self.media_path),
            'dynamic_media_sector_count': self.media_sector_count,
            'media_model_warnings': '|'.join(getattr(self.media_model, 'warnings', []) or []) if self.media_model is not None else '',
            'media_model_validation_warnings': '|'.join(getattr(self.media_model, 'validation_warnings', []) or []) if self.media_model is not None else '',
            'firmware_provenance_ok': bool(self.provenance.get('ok')),
            'firmware_shadow_3e_enabled': bool(getattr(self, 'fw_shadow3e_enabled', False)),
            'firmware_shadow_3e_model': 'v213_non_authoritative_shadow_scaffold',
            'firmware_shadow_3e_stage': getattr(self, 'fw_shadow3e_stage', ''),
            'firmware_shadow_3e_phase': getattr(self, 'fw_shadow3e_phase', ''),
            'firmware_shadow_3e_sticky_descramble': bool(getattr(self, 'fw_shadow3e_sticky', False)),
            'firmware_shadow_3e_event_count': len(getattr(self, 'fw_shadow3e_events', [])),
            'firmware_shadow_lba_map_enabled': True,
            'firmware_shadow_lba_map_model': 'v229_non_authoritative_lba_window_shadow',
            'firmware_shadow_lba_map_source': 'bridge_shadow_not_firmware_derived_yet',
        }
        row.update(self._media_delta_trace_fields())
        if source == 'catalog':
            row['catalog_match'] = True
        if source == 'dynamic_media_window':
            row['dynamic_media_window'] = True
        if extra:
            row.update(extra)
        if 'effective_media_lba_delta' in row:
            try:
                row['dynamic_media_effective_lba_delta'] = int(row.get('effective_media_lba_delta') or 0)
            except Exception:
                row['dynamic_media_effective_lba_delta'] = row.get('effective_media_lba_delta')
        return row

    def _check(self, index: int, tag: str, cdb: bytes, error: str, *, sense_key: int = 0x05, asc: int = 0x20, ascq: int = 0x00, extra: dict[str, Any] | None = None) -> dict[str, Any]:
        sense = fixed_sense(sense_key=sense_key, asc=asc, ascq=ascq)
        row = {
            'ok': False,
            'index': index,
            'tag': tag,
            'scsi_status': 'CHECK CONDITION',
            'sense_key': {0x00: 'NO SENSE', 0x02: 'NOT READY', 0x03: 'MEDIUM ERROR', 0x05: 'ILLEGAL REQUEST', 0x06: 'UNIT ATTENTION'}.get(int(sense_key) & 0x0F, f'0x{int(sense_key) & 0x0F:02X}'),
            'sense_key_hex': f'0x{sense_key:02X}',
            'asc': f'0x{asc:02X}',
            'ascq': f'0x{ascq:02X}',
            'direction': 'none',
            'source': 'dynamic_media_window_error',
            'cdb_hex': hx(cdb),
            'data_len': 0,
            'data_sha256': '',
            'data_hex': '',
            'data_preview_hex': '',
            'sense_data_hex': hx(sense),
            'sense_data_sha256': hashlib.sha256(sense).hexdigest(),
            'error': error,
            'dynamic_media_lba_delta': getattr(self, 'media_lba_delta', 0),
            'dynamic_media_configured_lba_delta': getattr(self, 'media_lba_delta', 0),
            'dynamic_media_view_mode': getattr(self, 'view_mode', ''),
            'dynamic_media_eject_policy': getattr(self, 'eject_policy', ''),
            'dynamic_media_auth_stage': getattr(self, 'auth_stage', ''),
            'dynamic_media_sticky_descramble': bool(getattr(self, 'sticky_descramble', False)),
            'dynamic_media_tray_state': getattr(self, 'tray_state', ''),
            'dynamic_media_present': bool(getattr(self, 'media_present', False)),
            'dynamic_media_state_event_count': len(getattr(self, 'state_events', [])),
            'firmware_provenance_ok': bool(self.provenance.get('ok')),
            'firmware_shadow_3e_enabled': bool(getattr(self, 'fw_shadow3e_enabled', False)),
            'firmware_shadow_3e_model': 'v213_non_authoritative_shadow_scaffold',
            'firmware_shadow_3e_stage': getattr(self, 'fw_shadow3e_stage', ''),
            'firmware_shadow_3e_phase': getattr(self, 'fw_shadow3e_phase', ''),
            'firmware_shadow_3e_sticky_descramble': bool(getattr(self, 'fw_shadow3e_sticky', False)),
            'firmware_shadow_3e_event_count': len(getattr(self, 'fw_shadow3e_events', [])),
            'firmware_shadow_lba_map_enabled': True,
            'firmware_shadow_lba_map_model': 'v229_non_authoritative_lba_window_shadow',
            'firmware_shadow_lba_map_source': 'bridge_shadow_not_firmware_derived_yet',
        }
        row.update(self._media_delta_trace_fields())
        if extra:
            row.update(extra)
        return row

    def _current_visible_max_lba(self) -> int:
        # Host-visible capacity.  This is deliberately based on drive/auth state,
        # not raw ISO file size: locked/video view exposes the small video
        # partition; unlocked/sticky view exposes the full Xbox DVD-9 geometry.
        if self.view_mode == 'stateful_xbox':
            if self.media_model is not None and hasattr(self.media_model, 'current_capacity_sector_count'):
                count = int(self.media_model.current_capacity_sector_count(self.auth_stage) or 0)
                if count:
                    return max(0, count - 1)
            return UNLOCKED_MAX_LBA if self.auth_stage == 'unlocked' else LOCKED_MAX_LBA
        return max(0, int(getattr(self, 'media_sector_count', 1)) - 1)

    def _seek10_lba(self, cdb: bytes) -> int:
        if len(cdb) < 6:
            return 0
        return int.from_bytes(cdb[2:6], 'big')

    def _media_read_capacity10_payload(self) -> bytes:
        if self.media_model is not None and hasattr(self.media_model, 'read_capacity_10_response'):
            data = self.media_model.read_capacity_10_response(self.auth_stage)
            if data:
                return data
        return UNLOCKED_READ_CAPACITY_10 if self.auth_stage == 'unlocked' else LOCKED_READ_CAPACITY_10

    def _media_mode3e_payload10(self) -> bytes:
        if self.media_model is not None and hasattr(self.media_model, 'mode_sense_3e_response'):
            state = 'unlocked' if self.auth_stage == 'unlocked' else 'locked'
            data = self.media_model.mode_sense_3e_response(state)
            # Challenge verify pages are still supplied by the bridge side auth
            # state because media profiles currently provide only stable
            # locked/unlocked observations.
            if data and state in ('locked', 'unlocked') and self.auth_phase not in ('challenge1', 'challenge2'):
                return data
        return self._shadow3e_payload10()


    def _media_dvd_structure_payload(self, cdb: bytes) -> tuple[str, bytes] | None:
        if len(cdb) < 10 or cdb[0] != 0xAD:
            return None
        vendor_selector = cdb[11] & 0xFF if len(cdb) >= 12 else 0
        # Keep Xbox vendor/auth AD C0 on the existing catalog path.
        if vendor_selector == 0xC0:
            return None
        if self.media_model is None or not hasattr(self.media_model, 'get_dvd_structure_payload'):
            return None
        fmt = cdb[7] & 0xFF
        found = self.media_model.get_dvd_structure_payload(fmt)
        if not found:
            return None
        name, payload = found
        alloc = int.from_bytes(cdb[8:10], 'big') if len(cdb) >= 10 else len(payload)
        return (str(name), _trunc(bytes(payload), alloc))


    def _shadow3e_payload10(self) -> bytes:
        if getattr(self, 'fw_shadow3e_stage', 'locked') == 'unlocked':
            return MODE_3E_UNLOCKED
        if getattr(self, 'fw_shadow3e_phase', '') == 'challenge1':
            return MODE_3E_VERIFY1
        if getattr(self, 'fw_shadow3e_phase', '') == 'challenge2':
            return MODE_3E_VERIFY2
        return MODE_3E_LOCKED

    def _shadow3e_update_event(self, event: str, index: int, cdb: bytes, data_out: bytes) -> None:
        if not getattr(self, 'fw_shadow3e_enabled', False) or not event:
            return
        if event == 'challenge1':
            self.fw_shadow3e_phase = 'challenge1'
        elif event == 'challenge2':
            self.fw_shadow3e_phase = 'challenge2'
        elif event == 'partition1_unlock':
            self.fw_shadow3e_stage = 'unlocked'
            self.fw_shadow3e_phase = 'partition1_unlocked'
        elif event == 'sticky_descramble_mode31':
            self.fw_shadow3e_sticky = True
            if self.fw_shadow3e_stage == 'unlocked':
                self.fw_shadow3e_phase = 'unlocked_sticky'
        elif event.startswith('start_stop_eject'):
            self.fw_shadow3e_stage = 'locked'
            self.fw_shadow3e_phase = 'ejected_relocked'
            self.fw_shadow3e_sticky = False
        elif event.startswith('start_stop_load'):
            self.fw_shadow3e_stage = 'locked'
            self.fw_shadow3e_phase = 'media_changed_locked'
            self.fw_shadow3e_sticky = False
        self.fw_shadow3e_events.append({
            'index': index,
            'event': event,
            'cdb_hex': hx(cdb),
            'data_out_len': len(data_out),
            'shadow_stage': self.fw_shadow3e_stage,
            'shadow_phase': self.fw_shadow3e_phase,
            'shadow_sticky': bool(self.fw_shadow3e_sticky),
        })

    def _shadow3e_state_extra(self, *, event: str = '', payload: bytes | None = None, cdb: bytes | None = None) -> dict[str, Any]:
        expected10 = self._shadow3e_payload10()
        expected = mode3e_truncate(cdb or b'', expected10) if cdb else expected10
        bridge_agrees = (
            getattr(self, 'auth_stage', '') == getattr(self, 'fw_shadow3e_stage', '') and
            bool(getattr(self, 'sticky_descramble', False)) == bool(getattr(self, 'fw_shadow3e_sticky', False))
        )
        out: dict[str, Any] = {
            'firmware_shadow_3e_observed': True,
            'firmware_shadow_3e_source': 'v213_non_authoritative_parallel_shadow',
            'firmware_shadow_3e_event': event,
            'firmware_shadow_3e_bridge_agrees': bool(bridge_agrees),
            'firmware_shadow_3e_expected_payload10_sha256': hashlib.sha256(expected10).hexdigest(),
            'firmware_shadow_3e_expected_payload_len': len(expected),
            'firmware_shadow_3e_expected_payload_sha256': hashlib.sha256(expected).hexdigest(),
            'firmware_shadow_3e_expected_preview_hex': hx(expected[:64]),
        }
        if payload is not None:
            out['firmware_shadow_3e_payload_match'] = (bytes(payload) == bytes(expected))
            out['firmware_shadow_3e_payload_sha256'] = hashlib.sha256(payload).hexdigest() if payload else ''
        return out

    def response_for(self, req: dict[str, Any], index: int) -> dict[str, Any]:
        tag = str(req.get('tag') or req.get('title') or f'iscsi_{index:03d}')
        try:
            cdb = dynamic_hex_to_bytes(req.get('cdb_hex') or req.get('cdb'))
            data_out = dynamic_hex_to_bytes(req.get('data_out_hex') or req.get('data_out'))
            if not cdb:
                return self._check(index, tag, b'', 'empty CDB')

            # v204 stateful Xbox auth/view mode.  This is still a SCSI-side
            # state machine, not a full MN103S firmware-in-loop run, but it
            # prevents the live bridge from pretending the drive stayed locked
            # after the 8050L unlock CDB sequence.
            if self.view_mode == 'stateful_xbox':
                if cdb[0] == 0x1B:
                    event = classify_start_stop_unit(cdb)
                    if event == 'eject':
                        self.auth_stage = 'locked'
                        self.auth_phase = 'ejected_relocked'
                        self.sticky_descramble = False
                        if self.eject_policy in ('relock_keep_loaded', 'keep_loaded', 'virtual_keep_loaded'):
                            # Test-friendly virtual-media behavior: Windows Explorer "Eject"
                            # resets auth/window state just like tray eject, but the ISO remains
                            # virtually inserted so the video partition is immediately readable again.
                            self.tray_state = 'closed'
                            self.media_present = True
                            self.auth_phase = 'ejected_relocked_keep_loaded'
                        else:
                            # Realistic behavior: tray is open and the medium is absent until a
                            # later START STOP UNIT load/close command or a fresh target session.
                            self.tray_state = 'open'
                            self.media_present = False
                    elif event == 'load':
                        self.auth_stage = 'locked'
                        self.auth_phase = 'media_changed_locked'
                        self.sticky_descramble = False
                        self.tray_state = 'closed'
                        self.media_present = True
                    state_event = 'start_stop_' + (event or 'unknown')
                    self.state_events.append({'index': index, 'event': state_event, 'cdb_hex': hx(cdb), 'data_out_len': len(data_out), 'tray_state': self.tray_state, 'media_present': bool(self.media_present), 'eject_policy': self.eject_policy})
                    self._shadow3e_update_event(state_event, index, cdb, data_out)
                    extra = {'state_event': state_event, 'state_event_count': len(self.state_events)}
                    extra.update(self._shadow3e_state_extra(event=state_event))
                    return self._good(index, tag, cdb, 'none', b'', source='v206_stateful_start_stop', extra=extra)
                if cdb[0] in (0x00, 0x1C, 0x1E, 0xBB):
                    return self._good(index, tag, cdb, 'none', b'', source='v206_stateful_good_no_data')
                if cdb[0] in (0x15, 0x55):
                    event = classify_8050l_mode_select(cdb, data_out)
                    if event:
                        if event == 'challenge1':
                            self.auth_phase = 'challenge1'
                        elif event == 'challenge2':
                            self.auth_phase = 'challenge2'
                        elif event == 'partition1_unlock':
                            self.auth_stage = 'unlocked'
                            self.auth_phase = 'partition1_unlocked'
                        elif event == 'sticky_descramble_mode31':
                            self.sticky_descramble = True
                            if self.auth_stage == 'unlocked':
                                self.auth_phase = 'unlocked_sticky'
                        self.state_events.append({'index': index, 'event': event, 'cdb_hex': hx(cdb), 'data_out_len': len(data_out)})
                        self._shadow3e_update_event(event, index, cdb, data_out)
                        extra = {'state_event': event, 'state_event_count': len(self.state_events)}
                        extra.update(self._shadow3e_state_extra(event=event))
                        return self._good(index, tag, cdb, 'none', b'', source='v204_stateful_auth_event', extra=extra)
                if cdb[0] == 0x25:
                    if not self.media_present:
                        return self._check(index, tag, cdb, 'medium not present after eject', sense_key=0x02, asc=0x3A, ascq=0x00)
                    payload = self._media_read_capacity10_payload()
                    return self._good(index, tag, cdb, 'in', payload, source='v227_stateful_read_capacity_media_profile')
                if is_mode_sense_3e(cdb):
                    payload = self._media_mode3e_payload10()
                    payload = mode3e_truncate(cdb, payload)
                    extra = self._shadow3e_state_extra(event='mode_sense_3e', payload=payload, cdb=cdb)
                    return self._good(index, tag, cdb, 'in', payload, source='v204_stateful_mode_sense_3e', extra=extra)

            # v203 Windows media-recognition helpers.  These are intentionally
            # transport/catalog-side responses: the auth path and READ data path
            # remain firmware-session/dynamic-media-window backed.
            if cdb[0] == 0x12:
                payload = scsi_inquiry_payload_vpd(cdb)
                if payload:
                    return self._good(index, tag, cdb, 'in', payload, source='v202_inquiry')
            if cdb[0] == 0x43:
                payload = mmc_read_toc_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v202_read_toc')
            if cdb[0] == 0x46:
                payload = mmc_get_configuration_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v202_get_configuration')
            if cdb[0] == 0x4A:
                payload = mmc_get_event_status_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v202_get_event_status')
            if cdb[0] == 0x51:
                payload = mmc_read_disc_information_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v202_read_disc_information')
            if cdb[0] == 0xAC:
                payload = mmc_get_performance_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v205_get_performance_minimal')
            if cdb[0] == 0xA4:
                payload = mmc_report_key_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v208_report_key_rpc_state')
            if cdb[0] == 0xBE:
                read_cd = mmc_read_cd_probe_payload(cdb, read_user_sector=self._read_media_stateful)
                if read_cd is not None:
                    payload, read_cd_info = read_cd
                    return self._good(index, tag, cdb, 'in', payload, source='v234_read_cd_probe', extra=read_cd_info)
            if cdb[0] == 0x52:
                payload = mmc_read_track_information_payload(cdb)
                return self._good(index, tag, cdb, 'in', payload, source='v234_read_track_information_minimal')
            if cdb[0] == 0xB6:
                return self._good(index, tag, cdb, 'none', b'', source='v234_set_streaming_good_no_data', extra={
                    'data_out_len': len(data_out),
                    'set_streaming_probe_accepted': True,
                })
            if cdb[0] == 0xAD:
                media_struct = self._media_dvd_structure_payload(cdb)
                if media_struct is not None:
                    struct_name, payload = media_struct
                    return self._good(index, tag, cdb, 'in', payload, source='v234_read_dvd_structure_media_profile', extra={
                        'read_dvd_structure_format': f'0x{cdb[7] & 0xFF:02X}',
                        'read_dvd_structure_payload': struct_name,
                    })
                payload = mmc_read_dvd_structure_probe_payload(cdb)
                if payload is not None:
                    return self._good(index, tag, cdb, 'in', payload, source='v234_read_dvd_structure_probe')
            if cdb[0] == 0x1D:
                return self._good(index, tag, cdb, 'none', b'', source='v205_send_diagnostic_good')
            if cdb[0] == 0x2B:
                # SEEK(10).  It has no data phase, but real firmware still
                # validates the host-visible address against the current
                # locked/unlocked view.  The boundary probe uses LBA 0x202000:
                # it must fail while locked/video capacity is visible, and pass
                # after the Xbox unlock exposes full DVD-9 geometry.
                if self.view_mode == 'stateful_xbox' and not self.media_present:
                    return self._check(index, tag, cdb, 'medium not present after eject', sense_key=0x02, asc=0x3A, ascq=0x00, extra={
                        'seek_lba': self._seek10_lba(cdb),
                        'visible_max_lba': self._current_visible_max_lba(),
                        'stateful_seek_bounds': True,
                    })
                seek_lba = self._seek10_lba(cdb)
                visible_max = self._current_visible_max_lba()
                if seek_lba > visible_max:
                    return self._check(index, tag, cdb, f'SEEK(10) LBA out of range for current visible capacity: lba={seek_lba} max={visible_max}', sense_key=0x05, asc=0x21, ascq=0x00, extra={
                        'seek_lba': seek_lba,
                        'visible_max_lba': visible_max,
                        'stateful_seek_bounds': True,
                        'expected_boundary_failure': self.view_mode == 'stateful_xbox' and self.auth_stage != 'unlocked',
                        'source': 'v213_seek10_lba_out_of_range',
                    })
                return self._good(index, tag, cdb, 'none', b'', source='v213_seek10_within_visible_bounds', extra={
                    'seek_lba': seek_lba,
                    'visible_max_lba': visible_max,
                    'stateful_seek_bounds': True,
                })
            if cdb[0] == 0xFF:
                return self._good(index, tag, cdb, 'none', b'', source='v205_vendor_probe_good_no_data')
            if cdb[0] == 0x5A:
                payload = mmc_mode_sense10_payload(cdb)
                if payload:
                    return self._good(index, tag, cdb, 'in', payload, source='v202_mode_sense10')

            if dynamic_is_read_cdb(cdb):
                if self.view_mode == 'stateful_xbox' and not self.media_present:
                    return self._check(index, tag, cdb, 'medium not present after eject', sense_key=0x02, asc=0x3A, ascq=0x00)
                host_lba, blocks, read_kind = dynamic_decode_read_cdb(cdb)
                visible_max = self._current_visible_max_lba()
                last_host_lba = host_lba + max(0, int(blocks) - 1)
                if self.view_mode == 'stateful_xbox' and last_host_lba > visible_max:
                    return self._check(index, tag, cdb, f'READ LBA out of range for current visible capacity: first={host_lba} blocks={blocks} last={last_host_lba} max={visible_max}', sense_key=0x05, asc=0x21, ascq=0x00, extra={
                        'host_lba': host_lba,
                        'blocks': blocks,
                        'last_host_lba': last_host_lba,
                        'visible_max_lba': visible_max,
                        'stateful_read_bounds': True,
                        'expected_boundary_failure': self.view_mode == 'stateful_xbox' and self.auth_stage != 'unlocked',
                        'source': 'v213_read_lba_out_of_range',
                        'firmware_shadow_lba_map_observed': True,
                        'firmware_shadow_lba_map_expected_ok': False,
                        'firmware_shadow_lba_map_expected_delta': 0 if self.auth_stage != 'unlocked' else self.media_lba_delta,
                        'firmware_shadow_lba_map_visible_max_lba': visible_max,
                    })
                payload, media_info = self._read_media_stateful(host_lba, blocks)
                media_lba = int(media_info.get('media_lba', host_lba))
                effective_delta = int(media_info.get('effective_media_lba_delta', media_lba - host_lba))
                read_segments = media_info.get('read_segments') or []
                return self._good(index, tag, cdb, 'in', payload, source='dynamic_media_window', extra={
                    'read_kind': read_kind,
                    'host_lba': host_lba,
                    'blocks': blocks,
                    'media_lba': media_lba,
                    'media_lba_end': media_info.get('media_lba_end', media_lba + max(0, int(blocks) - 1)),
                    'sector_size': self.sector_size,
                    'read_map_mode': 'v227_stateful_xbox_media_profile' if bool(media_info.get('media_is_redump_style')) else ('v204_stateful_xbox' if self.view_mode == 'stateful_xbox' else 'v203_dynamic_media_view'),
                    'read_map_note': str(media_info.get('read_map_note') or f'media_lba=host_lba+{effective_delta}'),
                    'effective_media_lba_delta': effective_delta,
                    'dynamic_media_lba_delta': effective_delta,
                    'dynamic_media_configured_lba_delta': self.media_lba_delta,
                    'media_layout_type': media_info.get('media_layout_type', ''),
                    'media_is_redump_style': bool(media_info.get('media_is_redump_style', False)),
                    'media_profile_loaded': bool(media_info.get('media_profile_loaded', False)),
                    'media_profile_path': media_info.get('media_profile_path', ''),
                    'media_segment_count': media_info.get('media_segment_count', len(read_segments) or 1),
                    'media_read_segments_json': json.dumps(read_segments, separators=(',', ':')) if read_segments else '',
                    'firmware_shadow_lba_map_observed': True,
                    'firmware_shadow_lba_map_expected_ok': True,
                    'firmware_shadow_lba_map_expected_delta': effective_delta,
                    'firmware_shadow_lba_map_expected_media_lba': media_lba,
                    'firmware_shadow_lba_map_visible_max_lba': visible_max,
                    'firmware_shadow_lba_map_bridge_agrees': True,
                })

            ent, self.cursor, _cands = dynamic_find_catalog_match(
                self.catalog, cdb, self.cursor,
                sequential=self.sequential_catalog,
                data_out=data_out,
            )
            if ent is None:
                if self.unsupported_check_condition:
                    return self._check(index, tag, cdb, 'unsupported CDB and no dynamic/catalog match')
                raise KeyError(f'unsupported CDB and no dynamic/catalog match: {dynamic_bytes_to_hex(cdb)}')
            direction = str(ent.get('direction') or 'in').lower()
            if direction == 'in':
                payload = dynamic_maybe_truncate_to_allocation(cdb, ent['data_in'], self.truncate_to_alloc)
            else:
                payload = b''
            return self._good(index, tag, cdb, 'in' if direction == 'in' else 'none', payload, source='catalog', extra={
                'catalog_index': ent.get('index'),
                'catalog_tag': ent.get('tag'),
                'catalog_direction': direction,
                'data_out_len': len(data_out),
                'data_out_sha256': hashlib.sha256(data_out).hexdigest() if data_out else '',
            })
        except Exception as e:
            cdb = dynamic_hex_to_bytes(req.get('cdb_hex') or req.get('cdb')) if (req.get('cdb_hex') or req.get('cdb')) else b''
            return self._check(index, tag, cdb, str(e))


def make_bridge(args: argparse.Namespace) -> tuple[Any, Any, Any]:
    if getattr(args, 'dynamic_media_config', None):
        cfg = dynamic_load_json(Path(args.dynamic_media_config))
        media_s = dynamic_resolve_media_path(cfg, getattr(args, 'dynamic_media_media', '') or '')
        if not media_s:
            raise SystemExit('dynamic media config has no media path; pass --dynamic-media-media')
        delta = dynamic_resolve_media_lba_delta(cfg, getattr(args, 'dynamic_media_lba_delta', None))
        bridge = DynamicMediaWindowBridge(
            cfg,
            media_path=Path(media_s),
            media_lba_delta=delta,
            sector_size=int(getattr(args, 'dynamic_media_sector_size', 2048)),
            model_path=str(getattr(args, 'dynamic_media_model_path', 'trace.gdr_packet_session_model')),
            expect_firmware=str(getattr(args, 'expect_firmware', CANONICAL_GDR8050L_0L23_BASENAME)),
            sequential_catalog=not bool(getattr(args, 'dynamic_media_match_any_order', False)),
            truncate_to_alloc=not bool(getattr(args, 'dynamic_media_no_truncate_to_allocation', False)),
            unsupported_check_condition=not bool(getattr(args, 'dynamic_media_fail_unknown', False)),
            view_mode=str(getattr(args, 'dynamic_media_view_mode', 'direct')),
            eject_policy=str(getattr(args, 'dynamic_media_eject_policy', 'real')),
        )
        return None, None, bridge
    xdvd = Path(args.xdvd_json)
    log = Path(args.log) if args.log else xdvd.with_suffix('.log')
    parser = GDRScsiParser(xdvd, args.iso, log if log.exists() else None, read_map=getattr(args, 'xdvd_read_map', 'windows'), read_trace_path=getattr(args, 'read_trace_jsonl', None), read_trace_limit=getattr(args, 'read_trace_limit', 0), device_profile=getattr(args, 'device_profile', None))
    if getattr(args, 'media_model_summary_json', None):
        parser.media_model.write_summary_json(args.media_model_summary_json)
    catalog = Catalog(args.catalog)
    bridge = HostBridge(parser, catalog)
    return parser, catalog, bridge



def _snapshot_rows_for_live_log(rows: list[dict[str, Any]], out_path: Path | None, response_dir: Path | None, *, compact_data_hex: bool = False) -> None:
    if not out_path and not response_dir:
        return
    # Use copies so compact JSONL/payload-file emission never mutates the row
    # that is still needed to build the in-flight SCSI response.
    copied = [dict(r) for r in rows]
    write_outputs(copied, out_path, response_dir, copied, compact_data_hex=compact_data_hex)


class LiveRowSink(list):
    def __init__(self, *, out_path: Path | None, response_dir: Path | None, compact_data_hex: bool):
        super().__init__()
        self.out_path = out_path
        self.response_dir = response_dir
        self.compact_data_hex = bool(compact_data_hex)
    def append(self, row: dict[str, Any]) -> None:  # type: ignore[override]
        super().append(row)
        _snapshot_rows_for_live_log(self, self.out_path, self.response_dir, compact_data_hex=self.compact_data_hex)


def run_server(args: argparse.Namespace) -> int:
    _parser, _catalog, bridge = make_bridge(args)
    rows: list[dict[str, Any]] = LiveRowSink(
        out_path=Path(args.out) if args.out else None,
        response_dir=Path(args.write_response_dir) if args.write_response_dir else None,
        compact_data_hex=bool(getattr(args, 'compact_jsonl_data', False)),
    )
    lock = threading.Lock()
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        srv.bind((args.host, int(args.port)))
        srv.listen(int(args.listen_backlog))
        srv.settimeout(float(args.accept_timeout))
        host, port = srv.getsockname()
        adv_host = str(args.advertise_host or host)
        if adv_host in ('0.0.0.0', '::'):
            adv_host = '127.0.0.1'
        adv_port = int(args.advertise_port or port)
        info = {'host': host, 'port': port, 'advertise_host': adv_host, 'advertise_port': adv_port, 'target_name': args.target_name, 'portal_group_tag': int(args.portal_group_tag)}
        if args.ready_file:
            Path(args.ready_file).write_text(json.dumps(info, indent=2), encoding='utf-8')
        print(f'[ISCSI-TARGET] listening {host}:{port} target={args.target_name}', flush=True)
        served = 0
        max_clients = int(args.max_clients)
        deadline = None
        if float(args.serve_timeout) > 0:
            deadline = time.time() + float(args.serve_timeout)
        try:
            while max_clients <= 0 or served < max_clients:
                if deadline is not None and time.time() >= deadline:
                    print(f'[ISCSI-TARGET] serve_timeout reached after {args.serve_timeout}s', flush=True)
                    break
                try:
                    conn, addr = srv.accept()
                except socket.timeout:
                    continue
                print(f'[ISCSI-TARGET] client {addr[0]}:{addr[1]}', flush=True)
                sess = IscsiSession(str(args.target_name), str(adv_host), int(adv_port), int(args.portal_group_tag))
                try:
                    serve_client(conn, bridge, rows, sess=sess, lock=lock)
                except KeyboardInterrupt:
                    print('[ISCSI-TARGET] keyboard interrupt during client; stopping', flush=True)
                    break
                except Exception as e:
                    rows.append({'index': -1, 'tag': 'iscsi_server_exception', 'ok': False, 'error': str(e)})
                    print(f'[ISCSI-TARGET] client_error={e}', flush=True)
                finally:
                    served += 1
                    print(f'[ISCSI-TARGET] client_done served={served} rows={len(rows)}', flush=True)
                    if args.out:
                        write_outputs(rows, Path(args.out), Path(args.write_response_dir) if args.write_response_dir else None, rows, compact_data_hex=bool(getattr(args, 'compact_jsonl_data', False)))
        except KeyboardInterrupt:
            print('[ISCSI-TARGET] keyboard interrupt; stopping', flush=True)
        print(f'[ISCSI-TARGET] served_clients={served} rows={len(rows)}', flush=True)
    if args.out:
        write_outputs(rows, Path(args.out), Path(args.write_response_dir) if args.write_response_dir else None, rows, compact_data_hex=bool(getattr(args, 'compact_jsonl_data', False)))
    return 0


def run_smoke(args: argparse.Namespace) -> int:
    reqs = load_requests(args.script)
    _parser, _catalog, bridge = make_bridge(args)
    server_rows: list[dict[str, Any]] = []
    ready: queue.Queue[tuple[str,int]] = queue.Queue(maxsize=1)
    errors: queue.Queue[str] = queue.Queue()

    def server_thread() -> None:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
                srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                srv.bind((args.host, 0 if int(args.port) == 0 else int(args.port)))
                srv.listen(1)
                host, port = srv.getsockname()
                ready.put((host, port))
                conn, _addr = srv.accept()
                sess = IscsiSession(str(args.target_name), host, int(port), int(args.portal_group_tag))
                serve_client(conn, bridge, server_rows, sess=sess)
        except Exception as e:
            errors.put(str(e))

    th = threading.Thread(target=server_thread, daemon=True)
    th.start()
    host, port = ready.get(timeout=float(args.timeout))
    rows = client_roundtrip(host, port, reqs, float(args.timeout), exercise_r2t=bool(args.exercise_r2t))
    th.join(timeout=float(args.timeout))
    if not errors.empty():
        raise SystemExit(f'iSCSI PDU bridge server error: {errors.get()}')
    write_outputs(rows, Path(args.out) if args.out else None, Path(args.write_response_dir) if args.write_response_dir else None, server_rows, compact_data_hex=bool(getattr(args, 'compact_jsonl_data', False)))
    if args.verify_md:
        write_verify(Path(args.verify_md), rows, server_rows)
    cmd_rows = [r for r in rows if isinstance(r.get('index'), int) and int(r.get('index')) >= 0 and str(r.get('tag')) != 'iscsi_logout']
    print(f'[ISCSI-PDU] requests={len(reqs)} responses={len(cmd_rows)} ok={sum(1 for r in cmd_rows if r.get("ok"))}', flush=True)
    return 0


def run(args: argparse.Namespace) -> int:
    if args.serve:
        return run_server(args)
    if not args.script:
        raise SystemExit('--script is required unless --serve is used')
    return run_smoke(args)


def main() -> int:
    ap = argparse.ArgumentParser(description='Small iSCSI PDU bridge/target surface for GDR/XDVD host bridge')
    ap.add_argument('xdvd_json', nargs='?', help='media profile / XDVD sidecar path; optional when --session-profile supplies media_profile')
    ap.add_argument('iso', nargs='?')
    ap.add_argument('--session-profile', help='HLDS emulator session JSON/JSONC generated by hlds_session_builder.py')
    ap.add_argument('--device-profile', help='HLDS device profile JSON/JSONC generated by analyzer chain')
    ap.add_argument('--media-profile', help='HLDS media profile / XDVD sidecar generated by dumper/media tooling')
    ap.add_argument('--log')
    ap.add_argument('--script', help='raw CDB script JSON/JSONL to send as SCSI Command PDUs; required unless --serve')
    ap.add_argument('--catalog', help='command catalog for exact-match annotation')
    ap.add_argument('--out', help='write iSCSI JSONL response/log rows')
    ap.add_argument('--write-response-dir', help='write full DATA IN/sense payloads observed behind the bridge')
    ap.add_argument('--verify-md', help='write verification markdown for loopback smoke mode')
    ap.add_argument('--host', default='127.0.0.1')
    ap.add_argument('--port', default='0')
    ap.add_argument('--timeout', default='10')
    ap.add_argument('--serve', action='store_true', help='run as a listening iSCSI target surface instead of loopback smoke')
    ap.add_argument('--ready-file', help='in --serve mode, write JSON with host/port/target_name after bind')
    ap.add_argument('--max-clients', default='1', help='in --serve mode, number of client sessions to accept; 0 means forever')
    ap.add_argument('--accept-timeout', default='0.5', help='seconds between accept() wakeups in --serve mode; enables Ctrl+C on Windows')
    ap.add_argument('--serve-timeout', default='0', help='optional max seconds to keep --serve alive; 0 means forever')
    ap.add_argument('--listen-backlog', default='8')
    ap.add_argument('--target-name', default=None)
    ap.add_argument('--portal-group-tag', default='1')
    ap.add_argument('--advertise-host', default=None, help='host/IP placed in SendTargets TargetAddress; defaults to bound host, or 127.0.0.1 for wildcard binds')
    ap.add_argument('--advertise-port', default=None, help='port placed in SendTargets TargetAddress; defaults to bound port')
    ap.add_argument('--xdvd-read-map', default=None, choices=['windows','direct','v157','firmware_window','firmware_lba_state','firmware','level2','physical_raw','raw_full','raw_iso','option1_raw','full_raw','xiso_alias','xbox_xiso','game_partition','xbox_raw','dual_alias','xbox_alias','game_offset0','game_zero','physical','output'], help='READ(10/12) LBA mapping mode; physical_raw/raw_full is direct full-disc ISO, xiso_alias/dual_alias remaps unlocked game LBAs for XISO extraction')
    ap.add_argument('--read-trace-jsonl', default=None, help='write compact READ(10/12) host_lba->iso_lba trace rows')
    ap.add_argument('--read-trace-limit', type=int, default=0, help='maximum compact read trace rows; 0 means unlimited')
    ap.add_argument('--media-model-summary-json', default=None, help='write v165 media model summary JSON at startup')
    ap.add_argument('--compact-jsonl-data', action='store_true', help='when writing payload files, omit full data_hex/sense_data_hex from JSONL rows')
    ap.add_argument('--exercise-r2t', action='store_true', help='in smoke mode, force one DATA OUT command through R2T/Data-Out instead of immediate data')
    ap.add_argument('--dynamic-media-config', default=None, help='v198-v205 strict-provenance session config; enables dynamic ISO-backed READ(10/12) backend')
    ap.add_argument('--dynamic-media-media', default='', help='override ISO/media path for --dynamic-media-config')
    ap.add_argument('--dynamic-media-lba-delta', type=int, default=None, help='override proven media window delta; defaults to config media.gdr_media_lba_delta')
    ap.add_argument('--dynamic-media-sector-size', type=int, default=2048)
    ap.add_argument('--dynamic-media-model-path', default='trace.gdr_packet_session_model')
    ap.add_argument('--dynamic-media-match-any-order', action='store_true', help='allow catalog responses to match any command order')
    ap.add_argument('--dynamic-media-no-truncate-to-allocation', action='store_true', help='do not truncate catalog responses to allocation length')
    ap.add_argument('--dynamic-media-fail-unknown', action='store_true', help='fail unknown CDBs instead of returning CHECK CONDITION')
    ap.add_argument('--dynamic-media-view-mode', default='direct', choices=['direct','stateful_xbox'], help='dynamic READ view: direct uses the configured fixed delta; stateful_xbox starts direct/locked and switches to unlocked game-window after the 8050L auth sequence')
    ap.add_argument('--dynamic-media-eject-policy', default='real', choices=['real','relock_keep_loaded','keep_loaded','virtual_keep_loaded'], help='stateful_xbox eject behavior: real opens/removes media; relock_keep_loaded resets auth/window but keeps the ISO inserted for Windows/VLC testing')
    ap.add_argument('--expect-firmware', default=CANONICAL_GDR8050L_0L23_BASENAME, help='expected firmware basename for strict provenance audit')
    args = ap.parse_args()
    args = apply_session_to_namespace(args)
    if not args.xdvd_json and not getattr(args, 'dynamic_media_config', None):
        raise SystemExit('xdvd_json/media_profile is required unless supplied by --session-profile or --dynamic-media-config')
    if not getattr(args, 'xdvd_read_map', None):
        args.xdvd_read_map = 'windows'
    if not getattr(args, 'target_name', None):
        args.target_name = 'iqn.2026-06.local.mn103s:gdr8050l'
    return run(args)


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