From 17c3fcfbac769a455d27c88e069f7d6871c143c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=AE=D1=80=D0=B8=D0=B9=20=D0=A7=D0=B5=D1=80=D0=BD=D0=B5?= =?UTF-8?q?=D0=BD=D0=BA=D0=BE?= Date: Mon, 24 Aug 2026 10:49:00 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BE=D1=82=D0=BF=D1=80=D0=B0=D0=B2=D0=BA?= =?UTF-8?q?=D0=B0=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D1=85=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20mapping=20=D1=81=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82?= =?UTF-8?q?=D0=BE=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 3 +- hockey_data/agent_bridge.py | 325 +++++++++++++----- hockey_data/static/admin-directories.js | 20 +- .../test_build97_mapping_chunked_transport.py | 154 +++++++++ 4 files changed, 412 insertions(+), 90 deletions(-) create mode 100644 tests/test_build97_mapping_chunked_transport.py diff --git a/app.py b/app.py index 2d181df..8ac1bad 100644 --- a/app.py +++ b/app.py @@ -29,7 +29,8 @@ from ui_builder import install_ui_builder from khl_site.khl_data_center import APP as khl_site_app BASE_DIR = Path(__file__).resolve().parent -BUILD_VERSION = "2026.08.20.17" +BUILD_VERSION = "2026.08.21.1" +# compatibility: BUILD_VERSION = "2026.08.20.17" # compatibility: BUILD_VERSION = "2026.08.20.16" # compatibility: BUILD_VERSION = "2026.08.20.15" # compatibility: BUILD_VERSION = "2026.08.20.14" diff --git a/hockey_data/agent_bridge.py b/hockey_data/agent_bridge.py index 14c310b..850dd72 100644 --- a/hockey_data/agent_bridge.py +++ b/hockey_data/agent_bridge.py @@ -24,6 +24,11 @@ AGENT_PROTOCOL_VERSION = 1 AGENT_ONLINE_WINDOW_SECONDS = 35 _DEVICE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{6,128}$") +# BUILD97: Mapping can easily exceed 300 links. Keep WebSocket/vMix packets small +# and predictable instead of sending the whole profile in one giant batch. +MAPPING_BATCH_MAX_COMMANDS = 40 +MAPPING_BATCH_MAX_BYTES = 48 * 1024 + def _utcnow() -> datetime: return datetime.utcnow() @@ -241,6 +246,9 @@ class VmixAgentHub: # BUILD90: one FIFO lock per Agent. Shortcut batches, Mapping refreshes and # timer control commands can no longer interleave on the same vMix instance. self._vmix_send_locks: dict[str, asyncio.Lock] = {} + # BUILD97: serialize Mapping applications per Agent while still letting + # runtime/timer commands use the normal vMix FIFO between Mapping chunks. + self._mapping_apply_locks: dict[str, asyncio.Lock] = {} self.mapping_data = MappingDataService(database, settings=settings) self._auto_refresh_task: asyncio.Task[None] | None = None self._auto_refresh_next: dict[tuple[str, str], float] = {} @@ -673,6 +681,173 @@ class VmixAgentHub: self._vmix_send_locks[device_id] = lock return lock + def _device_mapping_apply_lock(self, device_id: str) -> asyncio.Lock: + lock = self._mapping_apply_locks.get(device_id) + if lock is None: + lock = asyncio.Lock() + self._mapping_apply_locks[device_id] = lock + return lock + + @staticmethod + def _mapping_batch_chunks(entries: list[dict[str, Any]]) -> list[list[tuple[int, dict[str, Any]]]]: + """Group Mapping commands by vMix Input, then cap packet count and bytes. + + The returned tuples keep the original index so ACKs can be mapped back to + the original Mapping links even though commands are grouped by Input. + """ + groups: dict[str, list[tuple[int, dict[str, Any]]]] = {} + for index, entry in enumerate(entries): + command = entry.get("command") if isinstance(entry, dict) else None + input_ref = str((command or {}).get("Input") or "").strip() + groups.setdefault(input_ref, []).append((index, entry)) + + chunks: list[list[tuple[int, dict[str, Any]]]] = [] + for group in groups.values(): + current: list[tuple[int, dict[str, Any]]] = [] + current_bytes = 0 + for indexed_entry in group: + command = indexed_entry[1].get("command") or {} + try: + command_bytes = len(json.dumps(command, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) + 96 + except Exception: + command_bytes = 512 + if current and ( + len(current) >= MAPPING_BATCH_MAX_COMMANDS + or current_bytes + command_bytes > MAPPING_BATCH_MAX_BYTES + ): + chunks.append(current) + current = [] + current_bytes = 0 + current.append(indexed_entry) + current_bytes += command_bytes + if current: + chunks.append(current) + return chunks + + async def _send_mapping_entries_resilient( + self, + device_id: str, + *, + assignment_id: str, + match_id: str, + entries: list[dict[str, Any]], + use_batch: bool, + ) -> dict[str, Any]: + """Deliver Mapping commands reliably, with chunking and adaptive fallback. + + Large mappings are grouped by Input and sent in bounded batches. A timed + out/failed batch is recursively split; a single failed command falls back + to the legacy one-command transport. Mapping commands are idempotent + (SetText/SetImage/SetColor/visibility), so this retry strategy is safe. + """ + outcomes: list[dict[str, Any] | None] = [None] * len(entries) + stats = { + "chunks_total": 0, + "chunks_ok": 0, + "chunks_failed": 0, + "packets_sent": 0, + "retries": 0, + "fallback_commands": 0, + "input_groups": len({ + str(((entry.get("command") or {}).get("Input") or "")).strip() + for entry in entries + }), + } + + async def send_single( + index: int, entry: dict[str, Any], previous_reason: str = "", *, count_fallback: bool = True + ) -> None: + if count_fallback: + stats["fallback_commands"] += 1 + try: + ack = await self.send_vmix_command( + device_id, assignment_id=assignment_id, match_id=match_id, + command=entry.get("command") or {}, timeout=5.0, + ) + if bool(ack.get("ok")): + outcomes[index] = dict(ack) + else: + outcomes[index] = { + "ok": False, + "reason": str(ack.get("reason") or ack.get("error") or previous_reason or "vmix_error"), + } + except Exception as error: + outcomes[index] = { + "ok": False, + "reason": str(getattr(error, "detail", error) or previous_reason or "vmix_error")[:300], + } + + async def send_chunk(chunk: list[tuple[int, dict[str, Any]]]) -> None: + if not chunk: + return + commands = [entry.get("command") or {} for _, entry in chunk] + stats["packets_sent"] += 1 + try: + ack = await self.send_vmix_batch( + device_id, assignment_id=assignment_id, match_id=match_id, commands=commands, + timeout=max(6.0, min(10.0, 4.0 + len(commands) * 0.06)), + ) + ack_results = ack.get("results") if isinstance(ack.get("results"), list) else None + if ack_results is None: + if bool(ack.get("ok")): + for index, _entry in chunk: + outcomes[index] = {"ok": True} + return + raise RuntimeError(str(ack.get("reason") or ack.get("error") or "vmix_batch_error")) + + failed: list[tuple[int, dict[str, Any], str]] = [] + for position, (index, entry) in enumerate(chunk): + item_ack = ack_results[position] if position < len(ack_results) and isinstance(ack_results[position], dict) else {} + if bool(item_ack.get("ok")): + outcomes[index] = dict(item_ack) + else: + failed.append(( + index, entry, + str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"), + )) + # A batch ACK can contain isolated command failures. Retry only + # those commands once through the single-command transport. + for index, entry, reason in failed: + stats["retries"] += 1 + await send_single(index, entry, reason) + return + except Exception as error: + reason = str(getattr(error, "detail", error))[:300] + if len(chunk) > 1: + stats["retries"] += 1 + middle = max(1, len(chunk) // 2) + await send_chunk(chunk[:middle]) + await send_chunk(chunk[middle:]) + return + index, entry = chunk[0] + stats["retries"] += 1 + await send_single(index, entry, reason) + + async with self._device_mapping_apply_lock(device_id): + if use_batch: + chunks = self._mapping_batch_chunks(entries) + stats["chunks_total"] = len(chunks) + for chunk in chunks: + await send_chunk(chunk) + failed_now = sum(1 for index, _entry in chunk if outcomes[index] is None or not bool(outcomes[index].get("ok"))) + if failed_now == 0: + stats["chunks_ok"] += 1 + else: + stats["chunks_failed"] += 1 + else: + stats["chunks_total"] = len(entries) + for index, entry in enumerate(entries): + await send_single(index, entry, count_fallback=False) + if outcomes[index] is not None and bool(outcomes[index].get("ok")): + stats["chunks_ok"] += 1 + else: + stats["chunks_failed"] += 1 + + return { + "results": [item if isinstance(item, dict) else {"ok": False, "reason": "mapping_transport_no_ack"} for item in outcomes], + **stats, + } + async def receive_command_ack(self, device_id: str, message: dict[str, Any]) -> None: request_id = str(message.get("request_id") or "").strip() if not request_id: @@ -1037,55 +1212,40 @@ class VmixAgentHub: result["ok"] = not result["errors"] return result - if self._agent_supports_batch(agent_version): - try: - ack = await self.send_vmix_batch( - device_id, - assignment_id=assignment_id, - match_id=match_id, - commands=[entry["command"] for entry in pending_entries], - timeout=max(6.0, min(15.0, 4.0 + len(pending_entries) * 0.03)), - ) - ack_results = ack.get("results") if isinstance(ack.get("results"), list) else [] - for index, entry in enumerate(pending_entries): - item_ack = ack_results[index] if index < len(ack_results) and isinstance(ack_results[index], dict) else {} - if bool(item_ack.get("ok")): - if entry.get("entry_kind") == "rule": - result["rules_applied"] += 1 - else: - result["applied"] += 1 - self._mapping_value_cache[entry["cache_key"]] = entry["value"] - else: - result["errors"].append({ - "key": entry["data_key"], - "reason": str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"), - }) - except HTTPException as error: - result["errors"].append({"key": "*batch*", "reason": str(error.detail)}) - except Exception as error: - result["errors"].append({"key": "*batch*", "reason": str(error)[:300]}) - else: - for entry in pending_entries: - try: - ack = await self.send_vmix_command( - device_id, - assignment_id=assignment_id, - match_id=match_id, - command=entry["command"], - timeout=3.0, - ) - if bool(ack.get("ok")): - if entry.get("entry_kind") == "rule": - result["rules_applied"] += 1 - else: - result["applied"] += 1 - self._mapping_value_cache[entry["cache_key"]] = entry["value"] - else: - result["errors"].append({"key": entry["data_key"], "reason": str(ack.get("reason") or ack.get("error") or "vmix_error")}) - except HTTPException as error: - result["errors"].append({"key": entry["data_key"], "reason": str(error.detail)}) - except Exception as error: - result["errors"].append({"key": entry["data_key"], "reason": str(error)[:300]}) + use_batch = self._agent_supports_batch(agent_version) + result["transport"] = "chunked_batch" if use_batch else "legacy" + transport = await self._send_mapping_entries_resilient( + device_id, + assignment_id=assignment_id, + match_id=match_id, + entries=pending_entries, + use_batch=use_batch, + ) + result["batch_chunks_total"] = int(transport.get("chunks_total") or 0) + result["batch_chunks_applied"] = int(transport.get("chunks_ok") or 0) + result["batch_chunks_failed"] = int(transport.get("chunks_failed") or 0) + result["batch_packets_sent"] = int(transport.get("packets_sent") or 0) + result["batch_retries"] = int(transport.get("retries") or 0) + result["fallback_commands"] = int(transport.get("fallback_commands") or 0) + result["input_groups"] = int(transport.get("input_groups") or 0) + result["batch_max_commands"] = MAPPING_BATCH_MAX_COMMANDS + + ack_results = transport.get("results") if isinstance(transport.get("results"), list) else [] + for index, entry in enumerate(pending_entries): + item_ack = ack_results[index] if index < len(ack_results) and isinstance(ack_results[index], dict) else {} + if bool(item_ack.get("ok")): + if entry.get("entry_kind") == "rule": + result["rules_applied"] += 1 + else: + result["applied"] += 1 + self._mapping_value_cache[entry["cache_key"]] = entry["value"] + else: + result["errors"].append({ + "key": entry["data_key"], + "input": str((entry.get("command") or {}).get("Input") or ""), + "field": str((entry.get("command") or {}).get("SelectedName") or ""), + "reason": str(item_ack.get("reason") or item_ack.get("error") or "mapping_transport_error"), + }) result["ok"] = not result["errors"] return result @@ -1519,41 +1679,30 @@ class VmixAgentHub: "command": {"Function": function, "Input": input_ref, "SelectedName": selected_name, "Value": value}, }) - if self._agent_supports_batch(agent_version) and prepared: - try: - ack = await self.send_vmix_batch( - device_id, assignment_id=assignment_id, match_id=match_id, - commands=[item["command"] for item in prepared], - timeout=max(6.0, min(15.0, 4.0 + len(prepared) * 0.03)), - ) - ack_results = ack.get("results") if isinstance(ack.get("results"), list) else [] - for pos, item in enumerate(prepared): - item_ack = ack_results[pos] if pos < len(ack_results) and isinstance(ack_results[pos], dict) else {} - if bool(item_ack.get("ok")): - result["applied"] += 1 - else: - result["errors"].append({ - "index": item["index"], "key": item["key"], "field": item["field"], - "reason": str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"), - }) - except Exception as error: - result["errors"].append({"index": -1, "key": "*batch*", "reason": str(getattr(error, "detail", error))[:300]}) - else: - for item in prepared: - try: - ack = await self.send_vmix_command( - device_id, assignment_id=assignment_id, match_id=match_id, - command=item["command"], - ) - if bool(ack.get("ok")): - result["applied"] += 1 - else: - result["errors"].append({ - "index": item["index"], "key": item["key"], "field": item["field"], - "reason": str(ack.get("reason") or ack.get("error") or "vmix_error"), - }) - except Exception as error: - result["errors"].append({"index": item["index"], "key": item["key"], "field": item["field"], "reason": str(error)[:300]}) + if prepared: + use_batch = self._agent_supports_batch(agent_version) + result["transport"] = "chunked_batch" if use_batch else "legacy" + transport = await self._send_mapping_entries_resilient( + device_id, assignment_id=assignment_id, match_id=match_id, + entries=prepared, use_batch=use_batch, + ) + result["batch_chunks_total"] = int(transport.get("chunks_total") or 0) + result["batch_chunks_applied"] = int(transport.get("chunks_ok") or 0) + result["batch_chunks_failed"] = int(transport.get("chunks_failed") or 0) + result["batch_packets_sent"] = int(transport.get("packets_sent") or 0) + result["batch_retries"] = int(transport.get("retries") or 0) + result["fallback_commands"] = int(transport.get("fallback_commands") or 0) + result["input_groups"] = int(transport.get("input_groups") or 0) + ack_results = transport.get("results") if isinstance(transport.get("results"), list) else [] + for pos, item in enumerate(prepared): + item_ack = ack_results[pos] if pos < len(ack_results) and isinstance(ack_results[pos], dict) else {} + if bool(item_ack.get("ok")): + result["applied"] += 1 + else: + result["errors"].append({ + "index": item["index"], "key": item["key"], "field": item["field"], + "reason": str(item_ack.get("reason") or item_ack.get("error") or "mapping_transport_error"), + }) result["ok"] = not result["errors"] return result @@ -3815,7 +3964,10 @@ def create_hockey_agent_router( only_changed=only_changed, extra_context={"active_tab": str(active_tab or "").strip()} if str(active_tab or "").strip() else None, ) - if not result.get("ok") and result.get("reason"): + # BUILD97: field-level vMix errors are returned as diagnostics instead of + # being collapsed into a misleading HTTP 409 (e.g. "manual_apply"). + fatal_reasons = {"device_not_found", "device_not_paired", "no_match_assignment", "no_project_inventory", "vmix_not_connected", "mapping_missing"} + if not result.get("ok") and str(result.get("reason") or "") in fatal_reasons: raise HTTPException(status_code=409, detail=str(result.get("reason"))) return result @@ -3854,7 +4006,8 @@ def create_hockey_agent_router( @router.post("/api/hockey/admin/vmix-mapping/apply/{device_id}", dependencies=admin) async def admin_apply_device_mapping(device_id: str) -> dict[str, Any]: result = await hub.apply_mapping_to_device(device_id, reason="admin_manual_apply") - if not result.get("ok") and result.get("reason"): + fatal_reasons = {"device_not_found", "device_not_paired", "no_match_assignment", "no_project_inventory", "vmix_not_connected", "mapping_missing"} + if not result.get("ok") and str(result.get("reason") or "") in fatal_reasons: raise HTTPException(status_code=409, detail=str(result.get("reason"))) return result diff --git a/hockey_data/static/admin-directories.js b/hockey_data/static/admin-directories.js index a77e427..b83d8cf 100644 --- a/hockey_data/static/admin-directories.js +++ b/hockey_data/static/admin-directories.js @@ -114,6 +114,20 @@ setTimeout(() => URL.revokeObjectURL(url), 1000); } + function mappingApplyTransportSummary(result) { + const chunksTotal = Number(result?.batch_chunks_total || 0); + const chunksApplied = Number(result?.batch_chunks_applied || 0); + const retries = Number(result?.batch_retries || 0); + const fallback = Number(result?.fallback_commands || 0); + const inputs = Number(result?.input_groups || 0); + const parts = []; + if (inputs) parts.push(`Input: ${inputs}`); + if (chunksTotal) parts.push(`пакетов ${chunksApplied}/${chunksTotal}`); + if (retries) parts.push(`повторов ${retries}`); + if (fallback) parts.push(`одиночных ${fallback}`); + return parts.length ? ` · ${parts.join(" · ")}` : ""; + } + function mappingTransferReport(result, action = "Mapping перенесён") { const report = result?.report || {}; const mapped = Number(report.mapped || 0); @@ -2846,7 +2860,7 @@ if (!commands.length) return setStatus("У текущего Input нет настроенных связей с доступными данными.", true), renderMapping(); try { const result = await request("/api/hockey/admin/vmix-mapping/test-batch", { method: "POST", body: JSON.stringify({ device_id: deviceId, commands }) }); - const tail = `${missing.length ? ` · нет данных ${missing.length}` : ""}${result.errors?.length ? ` · ошибок ${result.errors.length}` : ""}`; + const tail = `${missing.length ? ` · нет данных ${missing.length}` : ""}${result.errors?.length ? ` · ошибок ${result.errors.length}` : ""}${mappingApplyTransportSummary(result)}`; setStatus(`Input ${input?.number ? `#${input.number} ` : ""}${input?.title || ""}: отправлено ${Number(result.applied || 0)} из ${Number(result.total || commands.length)} полей${tail}.`, Boolean(result.errors?.length)); } catch (error) { setStatus(error.message, true); } renderMapping(); @@ -2886,8 +2900,8 @@ if (!deviceId) return setStatus("Выберите online Agent для применения Mapping.", true), renderMapping(); try { const result = await request(`/api/hockey/admin/vmix-mapping/apply/${encodeURIComponent(deviceId)}`, { method: "POST" }); - const suffix = result.errors?.length ? ` · ошибок ${result.errors.length}` : ""; - setStatus(`Mapping применён: ${Number(result.applied || 0)} из ${Number(result.total || 0)} связей${suffix}.`); + const suffix = `${result.errors?.length ? ` · ошибок ${result.errors.length}` : ""}${mappingApplyTransportSummary(result)}`; + setStatus(`Mapping применён: ${Number(result.applied || 0)} из ${Number(result.total || 0)} связей${suffix}.`, Boolean(result.errors?.length)); } catch (error) { setStatus(error.message, true); } renderMapping(); }); diff --git a/tests/test_build97_mapping_chunked_transport.py b/tests/test_build97_mapping_chunked_transport.py new file mode 100644 index 0000000..6d1e764 --- /dev/null +++ b/tests/test_build97_mapping_chunked_transport.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import asyncio +from pathlib import Path + +from fastapi import HTTPException + +from hockey_data.agent_bridge import MAPPING_BATCH_MAX_COMMANDS, VmixAgentHub +from tests.support import LocalTestDatabase + + +def _entry(index: int, input_ref: str, *, value_size: int = 4) -> dict: + return { + "data_key": f"source.row.{index}", + "command": { + "Function": "SetText", + "Input": input_ref, + "SelectedName": f"Field{index}.Text", + "Value": "X" * value_size, + }, + } + + +def test_mapping_chunk_planner_groups_by_input_and_caps_packets(tmp_path: Path) -> None: + database = LocalTestDatabase(tmp_path / "mapping-chunks.sqlite3") + database.create_all() + hub = VmixAgentHub(database) # type: ignore[arg-type] + + entries = [*[_entry(i, "INPUT-A") for i in range(95)], *[_entry(100 + i, "INPUT-B") for i in range(17)]] + chunks = hub._mapping_batch_chunks(entries) + + assert len(chunks) == 4 # A: 40 + 40 + 15, B: 17 + assert all(len(chunk) <= MAPPING_BATCH_MAX_COMMANDS for chunk in chunks) + assert all(len({item[1]["command"]["Input"] for item in chunk}) == 1 for chunk in chunks) + # ACK positions can still be restored to original Mapping order. + assert [index for chunk in chunks for index, _entry_item in chunk] == list(range(len(entries))) + + +def test_mapping_chunk_transport_sends_300_plus_links_in_small_batches(tmp_path: Path) -> None: + database = LocalTestDatabase(tmp_path / "mapping-300.sqlite3") + database.create_all() + hub = VmixAgentHub(database) # type: ignore[arg-type] + packets: list[list[dict]] = [] + + async def fake_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict: + assert device_id == "GFX-MAP-300" + packets.append(commands) + return {"ok": True, "results": [{"ok": True} for _ in commands]} + + async def scenario() -> None: + hub.send_vmix_batch = fake_batch # type: ignore[method-assign] + entries: list[dict] = [] + # 12 Inputs x 31 commands = 372 Mapping commands. + for input_index in range(12): + entries.extend(_entry(input_index * 1000 + row, f"INPUT-{input_index:02d}") for row in range(31)) + result = await hub._send_mapping_entries_resilient( + "GFX-MAP-300", + assignment_id="assignment-300", + match_id="game-300", + entries=entries, + use_batch=True, + ) + assert all(item["ok"] for item in result["results"]) + assert result["input_groups"] == 12 + assert result["chunks_total"] == 12 + assert result["chunks_ok"] == 12 + assert result["chunks_failed"] == 0 + assert result["fallback_commands"] == 0 + assert sum(len(packet) for packet in packets) == 372 + assert max(len(packet) for packet in packets) <= MAPPING_BATCH_MAX_COMMANDS + assert all(len({command["Input"] for command in packet}) == 1 for packet in packets) + + asyncio.run(scenario()) + + +def test_mapping_transport_splits_timeout_batch_and_keeps_going(tmp_path: Path) -> None: + database = LocalTestDatabase(tmp_path / "mapping-retry.sqlite3") + database.create_all() + hub = VmixAgentHub(database) # type: ignore[arg-type] + packet_sizes: list[int] = [] + + async def flaky_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict: + packet_sizes.append(len(commands)) + if len(commands) > 10: + raise HTTPException(status_code=504, detail="simulated batch timeout") + return {"ok": True, "results": [{"ok": True} for _ in commands]} + + async def scenario() -> None: + hub.send_vmix_batch = flaky_batch # type: ignore[method-assign] + entries = [_entry(i, "ONE-BIG-INPUT") for i in range(35)] + result = await hub._send_mapping_entries_resilient( + "GFX-MAP-RETRY", + assignment_id="assignment-retry", + match_id="game-retry", + entries=entries, + use_batch=True, + ) + assert all(item["ok"] for item in result["results"]) + assert result["chunks_total"] == 1 + assert result["chunks_ok"] == 1 + assert result["chunks_failed"] == 0 + assert result["retries"] > 0 + assert packet_sizes[0] == 35 + assert any(size <= 10 for size in packet_sizes[1:]) + + asyncio.run(scenario()) + + +def test_mapping_transport_retries_isolated_failed_command(tmp_path: Path) -> None: + database = LocalTestDatabase(tmp_path / "mapping-single-fallback.sqlite3") + database.create_all() + hub = VmixAgentHub(database) # type: ignore[arg-type] + singles: list[dict] = [] + + async def partial_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict: + results = [{"ok": True} for _ in commands] + results[3] = {"ok": False, "reason": "temporary failure"} + return {"ok": False, "results": results} + + async def single_ok(device_id: str, *, assignment_id: str, match_id: str, command: dict, timeout: float = 5.0) -> dict: + singles.append(command) + return {"ok": True} + + async def scenario() -> None: + hub.send_vmix_batch = partial_batch # type: ignore[method-assign] + hub.send_vmix_command = single_ok # type: ignore[method-assign] + entries = [_entry(i, "INPUT-FALLBACK") for i in range(8)] + result = await hub._send_mapping_entries_resilient( + "GFX-MAP-FALLBACK", + assignment_id="assignment-fallback", + match_id="game-fallback", + entries=entries, + use_batch=True, + ) + assert all(item["ok"] for item in result["results"]) + assert result["fallback_commands"] == 1 + assert result["retries"] == 1 + assert len(singles) == 1 + assert singles[0]["SelectedName"] == "Field3.Text" + + asyncio.run(scenario()) + + +def test_build97_runtime_version_and_mapping_diagnostics() -> None: + root = Path(__file__).resolve().parents[1] + app = (root / "app.py").read_text(encoding="utf-8") + bridge = (root / "hockey_data" / "agent_bridge.py").read_text(encoding="utf-8") + admin = (root / "hockey_data" / "static" / "admin-directories.js").read_text(encoding="utf-8") + + assert 'BUILD_VERSION = "2026.08.21.1"' in app + assert 'result["transport"] = "chunked_batch" if use_batch else "legacy"' in bridge + assert 'MAPPING_BATCH_MAX_COMMANDS = 40' in bridge + assert 'mappingApplyTransportSummary' in admin + assert 'пакетов ${chunksApplied}/${chunksTotal}' in admin