debug(vp): env-flagged command-flow trace for slicer↔printer #1622 round-2 triage

  The shape-of-payload dump shipped earlier rules out cache wipes —
  shaddowlink's round-1 captures show AMS data reaches the slicer
  byte-identical to what the printer sent. The remaining symptom
  (picking a generic filament in archive mode "unloads" the slot) lives
  on the command path, which the snapshot dump doesn't see: it writes
  only the cached _latest_print_state and the periodic 1Hz push.

  Add append_event() in _debug.py — same env flag, separate file at
  <log_dir>/vp_wire/<vp>_cmd.jsonl. One JSONL line per event with UTC
  iso timestamp, direction (slicer_to_bridge / printer_to_slicer), MQTT
  topic, <channel>.<command> grep handle, and parsed payload. Wired at
  two points: mqtt_server._handle_publish for slicer publishes (after
  JSON decode so the trace matches what the bridge actually parsed) and
  mqtt_bridge._on_printer_raw "everything else" branch for printer
  responses (after serial rewrite so the trace matches what the slicer
  sees on the wire). Pushall / get_version stay out — both are handled
  locally and never round-trip through the bridge.

  Bytes payloads get the same \x00-tolerance fix from #927 so
  OrcaSlicer's C-string-null publishes parse cleanly; un-parseable
  bytes fall back to {"raw": "..."} so every line stays valid JSON.

backend/app/services/virtual_printer/_debug.py

@@ -1,19 +1,31 @@
 """Env-flagged wire-payload dump for VP MQTT debug (gated; off by default).
 
-Set ``BAMBUDDY_VP_DUMP_WIRE=1`` to write the most recent inbound (bridge
-cache input) and outbound (slicer-facing 1Hz push) MQTT payloads to disk,
-one file per VP per direction, overwritten each tick.
-
-Used to triage shape-of-payload bugs (e.g. #1622) where the question is
-"is the bridge missing fields in the cache, or is something else stripping
-them on the way out to the slicer?" Compare ``*_in.json`` and ``*_out.json``
-for the failing VP against a known-good VP (e.g. H2D vs P1S).
-
-Layout: ``<log_dir>/vp_wire/<sanitized_vp_name>_<direction>.json``
+Set ``BAMBUDDY_VP_DUMP_WIRE=1`` to enable two complementary capture modes:
+
+1. ``dump_wire``: most recent inbound (bridge cache input) and outbound
+   (slicer-facing 1Hz push) MQTT payloads, one file per VP per direction,
+   overwritten each tick. Triages shape-of-payload bugs (e.g. #1622 round 1)
+   where the question is "is the bridge missing fields in the cache, or is
+   something else stripping them on the way out to the slicer?" Compare
+   ``*_in.json`` and ``*_out.json`` for the failing VP against a known-good
+   one (e.g. H2D vs P1S).
+
+2. ``append_event``: time-ordered JSONL log of every slicer↔printer command
+   payload that flows through the VP (excludes the cached-as-base 1Hz push,
+   which dump_wire already covers). Triages command-flow bugs (e.g. #1622
+   round 2) where the cached state looks right but a slicer-initiated
+   write (ams_filament_setting / extrusion_cali_set / xcam / system) ends
+   up corrupting state. One line per event with wall-clock timestamp.
+
+Layout:
+- snapshot: ``<log_dir>/vp_wire/<sanitized_vp_name>_<direction>.json``
+- events:   ``<log_dir>/vp_wire/<sanitized_vp_name>_cmd.jsonl``
 
 Failure modes are swallowed at debug level — debug instrumentation must
 never break the bridge or slicer-facing 1Hz loop. Disable by unsetting the
 env var; the in-progress files stay on disk and can be deleted manually.
+``_cmd.jsonl`` appends forever while enabled; for long debug sessions,
+delete between captures rather than relying on rotation.
 """
 
 from __future__ import annotations
@@ -22,6 +34,7 @@ import json
 import logging
 import os
 import re
+from datetime import datetime, timezone
 
 from backend.app.core.config import settings as app_settings
 
@@ -64,3 +77,63 @@ def dump_wire(vp_name: str, direction: str, payload: dict | bytes | str) -> None
         tmp.replace(path)
     except OSError as e:
         logger.debug("[%s] vp_wire dump (%s) failed: %s", vp_name, direction, e)
+
+
+def _command_label(payload: dict) -> str:
+    """Best-effort one-word label for the command, used as a grep handle in the JSONL.
+
+    Bambu's MQTT request/response shape is ``{"<channel>": {"command": "<name>", ...}}``
+    where channel is ``print``/``pushing``/``info``/``system``/``xcam``/etc.
+    Returns ``"<channel>.<command>"`` when we can find it, ``"?"`` otherwise.
+    """
+    if not isinstance(payload, dict):
+        return "?"
+    for channel, body in payload.items():
+        if isinstance(body, dict):
+            cmd = body.get("command")
+            if isinstance(cmd, str) and cmd:
+                return f"{channel}.{cmd}"
+    return "?"
+
+
+def append_event(vp_name: str, direction: str, topic: str, payload: dict | bytes | str) -> None:
+    """Append one event line to ``<log_dir>/vp_wire/<vp_name>_cmd.jsonl``.
+
+    No-op when the env flag is unset. ``direction`` should be one of
+    ``"slicer_to_bridge"`` or ``"printer_to_slicer"`` so a diff between
+    a working VP and a broken VP can be read top-to-bottom in causal order.
+    Bytes payloads are utf-8 decoded then json-parsed best-effort; un-parseable
+    payloads are logged as ``{"raw": "<text>"}`` so the line is still valid JSON.
+    """
+    if not _enabled():
+        return
+    try:
+        target_dir = app_settings.log_dir / "vp_wire"
+        target_dir.mkdir(parents=True, exist_ok=True)
+        path = target_dir / f"{_sanitize(vp_name)}_cmd.jsonl"
+
+        if isinstance(payload, bytes):
+            try:
+                parsed: dict | str = json.loads(payload.decode("utf-8", errors="replace").rstrip("\x00 \r\n\t"))
+            except (json.JSONDecodeError, UnicodeDecodeError):
+                parsed = {"raw": payload.decode("utf-8", errors="replace")}
+        elif isinstance(payload, str):
+            try:
+                parsed = json.loads(payload.rstrip("\x00 \r\n\t"))
+            except json.JSONDecodeError:
+                parsed = {"raw": payload}
+        else:
+            parsed = payload
+
+        record = {
+            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
+            "dir": direction,
+            "topic": topic,
+            "cmd": _command_label(parsed) if isinstance(parsed, dict) else "?",
+            "payload": parsed,
+        }
+        line = json.dumps(record, default=str) + "\n"
+        with path.open("a", encoding="utf-8") as fp:
+            fp.write(line)
+    except OSError as e:
+        logger.debug("[%s] vp_wire append (%s) failed: %s", vp_name, direction, e)

backend/app/services/virtual_printer/mqtt_bridge.py

@@ -41,7 +41,7 @@ import logging
 import socket
 from typing import TYPE_CHECKING
 
-from backend.app.services.virtual_printer._debug import dump_wire
+from backend.app.services.virtual_printer._debug import append_event, dump_wire
 
 if TYPE_CHECKING:
     from backend.app.services.bambu_mqtt import BambuMQTTClient
@@ -670,6 +670,13 @@ class MQTTBridge:
         if target_bytes in payload:
             payload = payload.replace(target_bytes, self.vp_serial.encode("ascii"))
         vp_topic = f"device/{self.vp_serial}/{suffix}"
+        # Env-flagged command trace (#1622): every printer-originated response
+        # that gets fanned to the slicer (extrusion_cali_get / ams write acks /
+        # xcam / system / etc.) gets a line in vp_wire/<vp>_cmd.jsonl. Pair
+        # with the slicer-side publishes captured in mqtt_server. Off by
+        # default. Capture AFTER serial rewrite so the dump matches what the
+        # slicer actually sees on the wire.
+        append_event(self.vp_name, "printer_to_slicer", vp_topic, payload)
         try:
             asyncio.run_coroutine_threadsafe(
                 self._mqtt_server.push_raw_to_clients(vp_topic, payload),

backend/app/services/virtual_printer/mqtt_server.py

@@ -15,7 +15,7 @@ from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from backend.app.services.virtual_printer._debug import dump_wire
+from backend.app.services.virtual_printer._debug import append_event, dump_wire
 
 if TYPE_CHECKING:
     from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge
@@ -1212,6 +1212,11 @@ class SimpleMQTTServer:
                 )
                 return
 
+            # Env-flagged command trace (#1622): every slicer-originated publish
+            # gets a line in vp_wire/<vp>_cmd.jsonl alongside the printer-side
+            # responses captured in mqtt_bridge. Off by default.
+            append_event(self.vp_name, "slicer_to_bridge", topic, data)
+
             # The synthetic flow below is the original (pre-bridge) behaviour and is
             # what the proven-working FTP "Send" depends on. Do NOT replace any
             # synthetic response with a forward — only ADD forwarding alongside,

backend/tests/unit/test_vp_wire_dump.py

@@ -121,3 +121,86 @@ def test_env_check_is_per_call_not_module_load(_isolated_log_dir, monkeypatch):
         assert (_isolated_log_dir / "vp_wire" / "VP1_out.json").is_file()
     finally:
         os.environ.pop("BAMBUDDY_VP_DUMP_WIRE", None)
+
+
+# --- append_event (command-flow trace) --------------------------------------
+
+
+def _read_jsonl(path):
+    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+
+
+def test_append_event_disabled_by_default_writes_nothing(_isolated_log_dir, monkeypatch):
+    monkeypatch.delenv("BAMBUDDY_VP_DUMP_WIRE", raising=False)
+    _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", {"hello": "world"})
+    assert not (_isolated_log_dir / "vp_wire").exists()
+
+
+def test_append_event_appends_one_jsonl_line_per_call(_isolated_log_dir, monkeypatch):
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", {"print": {"command": "ams_filament_setting"}})
+    _debug.append_event(
+        "VP1", "printer_to_slicer", "device/X/report", {"print": {"command": "ams_filament_setting", "result": "ok"}}
+    )
+    path = _isolated_log_dir / "vp_wire" / "VP1_cmd.jsonl"
+    rows = _read_jsonl(path)
+    assert len(rows) == 2
+    assert rows[0]["dir"] == "slicer_to_bridge"
+    assert rows[0]["topic"] == "device/X/request"
+    assert rows[0]["cmd"] == "print.ams_filament_setting"
+    assert rows[0]["payload"] == {"print": {"command": "ams_filament_setting"}}
+    assert rows[1]["dir"] == "printer_to_slicer"
+    assert rows[1]["cmd"] == "print.ams_filament_setting"
+
+
+def test_append_event_parses_bytes_payload(_isolated_log_dir, monkeypatch):
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    raw = b'{"info": {"command": "get_version", "sequence_id": "0"}}'
+    _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", raw)
+    rows = _read_jsonl(_isolated_log_dir / "vp_wire" / "VP1_cmd.jsonl")
+    assert rows[0]["payload"] == {"info": {"command": "get_version", "sequence_id": "0"}}
+    assert rows[0]["cmd"] == "info.get_version"
+
+
+def test_append_event_unparseable_payload_kept_as_raw(_isolated_log_dir, monkeypatch):
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    _debug.append_event("VP1", "printer_to_slicer", "device/X/report", b"not-json-just-bytes")
+    rows = _read_jsonl(_isolated_log_dir / "vp_wire" / "VP1_cmd.jsonl")
+    assert rows[0]["payload"] == {"raw": "not-json-just-bytes"}
+    assert rows[0]["cmd"] == "?"
+
+
+def test_append_event_handles_trailing_null_from_orca(_isolated_log_dir, monkeypatch):
+    """Same #927 quirk as ``_handle_publish``: OrcaSlicer can ship publishes with a
+    trailing C-string null. The trace must still parse so the dump matches what
+    the bridge actually saw, not raw text."""
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", b'{"info":{"command":"get_version"}}\x00')
+    rows = _read_jsonl(_isolated_log_dir / "vp_wire" / "VP1_cmd.jsonl")
+    assert rows[0]["payload"] == {"info": {"command": "get_version"}}
+
+
+def test_append_event_sanitizes_vp_name(_isolated_log_dir, monkeypatch):
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    _debug.append_event("../../etc/passwd", "slicer_to_bridge", "device/X/request", {"x": 1})
+    files = list((_isolated_log_dir / "vp_wire").glob("*_cmd.jsonl"))
+    assert len(files) == 1
+    assert "/" not in files[0].name
+
+
+def test_append_event_includes_iso_timestamp(_isolated_log_dir, monkeypatch):
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", {"x": 1})
+    rows = _read_jsonl(_isolated_log_dir / "vp_wire" / "VP1_cmd.jsonl")
+    ts = rows[0]["ts"]
+    # ISO-8601 with timezone (Z or +00:00 suffix from UTC).
+    assert "T" in ts and (ts.endswith("+00:00") or ts.endswith("Z"))
+
+
+def test_append_event_failure_swallowed(_isolated_log_dir, monkeypatch):
+    """Debug instrumentation must never crash the bridge or slicer loop."""
+    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
+    blocker = _isolated_log_dir / "blocker"
+    blocker.write_text("not a dir")
+    with patch.object(app_settings, "log_dir", blocker):
+        _debug.append_event("VP1", "slicer_to_bridge", "device/X/request", {"x": 1})  # must not raise