fix: normalize YAML and snapshot input failures

src/pt_snap_cli/core/snapshot_import_backend.py

@@ -38,7 +38,15 @@ def dump_to_db(
                 )
             except UnsafePickleError as exc:
                 raise ImportExecutionError(f"Snapshot pickle rejected: {exc}") from exc
-            except (OSError, sqlite3.DatabaseError, pickle.UnpicklingError, ValueError) as exc:
+            except (
+                OSError,
+                sqlite3.DatabaseError,
+                pickle.UnpicklingError,
+                ValueError,
+                OverflowError,
+            ) as exc:
+                # SQLite rejects out-of-range integer bindings with OverflowError,
+                # rather than DatabaseError. Keep the original numeric failure.
                 raise ImportExecutionError(f"Snapshot import backend failed: {exc}") from exc
 
             if not ok:

src/pt_snap_cli/query/config.py

@@ -204,8 +204,16 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) ->
         Returns:
             QueryTemplate instance.
         """
+        data = _require_mapping(data, "template")
+        if "name" in data:
+            _require_name(data["name"], "template")
         parameters = {}
-        for name, param_data in data.get("parameters", {}).items():
+        parameter_data = _require_mapping(data.get("parameters", {}), "parameters")
+        for name, param_data in parameter_data.items():
+            _require_name(name, "parameter")
+            param_data = _require_mapping(param_data, f"parameter '{name}'")
+            if not isinstance(param_data.get("type", "str"), str):
+                raise ValueError(f"Parameter '{name}' type must be a string")
             parameters[name] = QueryParameter(
                 name=name,
                 type=param_data.get("type", "str"),
@@ -218,6 +226,8 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) ->
         template_name = data.get("name", "")
         query_variants = _parse_query_variants(template_name, data.get("query_variants"))
         explicit_query = data.get("query")
+        if explicit_query is not None and not isinstance(explicit_query, str):
+            raise ValueError(f"query for '{template_name}' must be a SQL string")
         if query_variants:
             if isinstance(explicit_query, str) and explicit_query.strip():
                 raise ValueError(
@@ -245,6 +255,17 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) ->
         )
 
 
+def _require_mapping(raw: Any, label: str) -> dict[str, Any]:
+    if not isinstance(raw, dict):
+        raise ValueError(f"{label} must be a mapping")
+    return raw
+
+
+def _require_name(raw: Any, label: str) -> None:
+    if not isinstance(raw, str) or not raw.strip():
+        raise ValueError(f"{label} name must be a non-empty string")
+
+
 def _parse_semantics_version(template_name: str, raw: Any) -> int | None:
     if raw is None:
         return None
@@ -408,23 +429,15 @@ def load_yaml(cls, path: str | Path, default_category: str | None = None) -> Que
         Raises:
             FileNotFoundError: If file does not exist.
             yaml.YAMLError: If YAML is invalid.
+            ValueError: If the configuration structure or template metadata is invalid.
         """
         path = Path(path)
         if not path.exists():
             raise FileNotFoundError(f"Config file not found: {path}")
 
         with open(path, encoding="utf-8") as f:
-            data = yaml.safe_load(f) or {}
-
-        queries = {}
-        for name, query_data in data.get("queries", {}).items():
-            query_data["name"] = name
-            queries[name] = QueryTemplate.from_dict(query_data, default_category=default_category)
-
-        return cls(
-            version=data.get("version", "1.0"),
-            queries=queries,
-        )
+            data = yaml.safe_load(f)
+        return cls._from_yaml_data(data, default_category=default_category)
 
     @classmethod
     def load_yaml_from_string(cls, content: str) -> QueryConfig:
@@ -435,13 +448,29 @@ def load_yaml_from_string(cls, content: str) -> QueryConfig:
 
         Returns:
             QueryConfig instance.
+
+        Raises:
+            yaml.YAMLError: If YAML is invalid.
+            ValueError: If the configuration structure or template metadata is invalid.
         """
-        data = yaml.safe_load(content) or {}
+        return cls._from_yaml_data(yaml.safe_load(content))
+
+    @classmethod
+    def _from_yaml_data(cls, raw: Any, default_category: str | None = None) -> QueryConfig:
+        # An empty YAML document remains an empty configuration. Other falsy
+        # values ([], false, 0, "") are not mappings and must not silently pass.
+        data = _require_mapping({} if raw is None else raw, "configuration")
+        query_definitions = _require_mapping(data.get("queries", {}), "queries")
 
         queries = {}
-        for name, query_data in data.get("queries", {}).items():
-            query_data["name"] = name
-            queries[name] = QueryTemplate.from_dict(query_data)
+        for name, query_data in query_definitions.items():
+            _require_name(name, "query")
+            query_data = _require_mapping(query_data, f"query '{name}'")
+            # Do not overwrite caller data or shared YAML aliases with the
+            # enclosing name. Each query gets its own top-level mapping.
+            queries[name] = QueryTemplate.from_dict(
+                {**query_data, "name": name}, default_category=default_category
+            )
 
         return cls(
             version=data.get("version", "1.0"),

src/pt_snap_cli/snapshot/PROVENANCE.md

@@ -321,3 +321,21 @@ JSON tuple-to-array normalization; unsupported deep evidence is omitted while
 the original text stack and snapshot import remain accepted. No allocator,
 event identity, source mapping, license or executable fixture changed. No
 external implementation was incorporated.
+
+## 2026-11-01 — Snapshot input error normalization (issue #280)
+
+First-party changes in `base/entities.py` validate the mappings, sequences and
+required keys consumed by entity construction, finite allocator numeric fields,
+string-valued block states, and nonempty trace action strings. Producer block
+state strings remain unchanged rather than being restricted to replay constants;
+the existing SQL value mapping and missing-column default are unchanged. Failures
+of existing trace integer conversion become `ValueError` with the cause preserved.
+`tools/adaptors/snapshot2db.py`
+checks the root/device-trace container before device selection. Numeric strings
+in traces, optional defaults, sentinels, tuple containers, event IDs and successful
+replay behavior remain unchanged. The core import adapter also normalizes SQLite
+integer-binding overflow at its existing domain-error boundary, without imposing
+a new representation-wide numeric range. Generated regressions use trusted
+in-memory representations behind the loader seam, not new executable pickle
+fixtures. No parser/layout, unknown-field policy or upstream-code change is made;
+source mappings, attribution and license records are unchanged.

src/pt_snap_cli/snapshot/base/entities.py

@@ -1,7 +1,32 @@
+import math
 from dataclasses import dataclass, field
 from typing import Any, Literal, cast
 
 
+def _validate_mapping(raw: Any, label: str, required: tuple[str, ...] = ()) -> None:
+    if not isinstance(raw, dict):
+        raise ValueError(f"{label} must be a mapping")
+    for name in required:
+        if name not in raw:
+            raise ValueError(f"{label} is missing required field '{name}'")
+
+
+def _validate_sequence(raw: Any, label: str) -> None:
+    # Tuples already work for trusted in-memory representations; retain that
+    # compatibility rather than requiring a particular serialized container.
+    if not isinstance(raw, (list, tuple)):
+        raise ValueError(f"{label} must be a list or tuple")
+
+
+def _validate_numbers(raw: dict[str, Any], label: str, names: tuple[str, ...]) -> None:
+    for name in names:
+        value = raw[name]
+        if not isinstance(value, (int, float)) or (
+            isinstance(value, float) and not math.isfinite(value)
+        ):
+            raise ValueError(f"{label} field '{name}' must be a finite number")
+
+
 class Frame:
     filename: str = ""
     line: int = -1
@@ -11,6 +36,7 @@ class Frame:
 
     @classmethod
     def from_dict(cls, frame_dict: dict[str, Any]):
+        _validate_mapping(frame_dict, "Frame", ("filename", "line", "name"))
         frame = cls()
         frame.filename = frame_dict["filename"]
         frame.line = frame_dict["line"]
@@ -39,10 +65,17 @@ def _format_callstack(frames: list[Frame] | list[dict[str, Any]]) -> str:
 
 
 def _validate_raw_frames(frames: list[dict[str, Any]]) -> None:
+    _validate_sequence(frames, "Frames")
     for frame in frames:
-        _ = frame["filename"]
-        _ = frame["line"]
-        _ = frame["name"]
+        _validate_mapping(frame, "Frame", ("filename", "line", "name"))
+
+
+def _trace_integer(trace_dict: dict[str, Any], name: str, default: int) -> int:
+    """Retain integer conversion while rejecting malformed numeric metadata."""
+    try:
+        return int(trace_dict.get(name, default))
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise ValueError(f"Trace field '{name}' must be integer-compatible") from exc
 
 
 @dataclass
@@ -82,19 +115,24 @@ class TraceEntry:
 
     @classmethod
     def from_dict(cls, trace_dict: dict[str, Any], _raw_frames: bool = False):
+        _validate_mapping(trace_dict, "Trace entry")
+        action = trace_dict.get("action")
+        if not isinstance(action, str) or not action.strip():
+            raise ValueError("Trace field 'action' must be a non-empty string")
         frame_dicts = trace_dict.get("frames", [])
+        _validate_sequence(frame_dicts, "Frames")
         if _raw_frames:
             _validate_raw_frames(frame_dicts)
         trace_entry = cls(
-            action=trace_dict["action"],
-            addr=int(trace_dict.get("addr", -1)),
-            size=int(trace_dict.get("size", 0)),
-            stream=int(trace_dict.get("stream", 0)),
-            device_free=int(trace_dict.get("device_free", -1)),
+            action=action,
+            addr=_trace_integer(trace_dict, "addr", -1),
+            size=_trace_integer(trace_dict, "size", 0),
+            stream=_trace_integer(trace_dict, "stream", 0),
+            device_free=_trace_integer(trace_dict, "device_free", -1),
             _origin=trace_dict,
             frames=[] if _raw_frames else [Frame.from_dict(frame) for frame in frame_dicts],
             _raw_frames=frame_dicts if _raw_frames else None,
-            idx=int(trace_dict["id"]) if "id" in trace_dict else -1,
+            idx=_trace_integer(trace_dict, "id", -1),
         )
         return trace_entry
 
@@ -146,11 +184,9 @@ class Block:
     requested_size: int = 0  # size requested during malloc, may be smaller than
     # size due to rounding
     address: int = -1
-    state: Literal[
-        "active_allocated",  # used by a tensor
-        "active_pending_free",  # waiting for another stream to finish using this, then it will become free
-        "inactive",
-    ] = BlockState.INACTIVE  # free for reuse
+    # Preserve producer state strings; BlockState names the replay states, not
+    # a closed input enum (PyTorch also emits "active_awaiting_free").
+    state: str = BlockState.INACTIVE  # free for reuse
     frames: list[Frame] = field(
         default_factory=list
     )  # stack trace from where the allocation occurred
@@ -163,14 +199,20 @@ class Block:
 
     @classmethod
     def from_dict(cls, block_dict: dict[str, Any], _raw_frames: bool = False):
+        _validate_mapping(block_dict, "Block", ("size", "requested_size", "address", "state"))
+        _validate_numbers(block_dict, "Block", ("size", "requested_size", "address"))
+        state = block_dict["state"]
+        if not isinstance(state, str):
+            raise ValueError("Block field 'state' must be a string")
         frame_dicts = block_dict.get("frames", [])
+        _validate_sequence(frame_dicts, "Frames")
         if _raw_frames:
             _validate_raw_frames(frame_dicts)
         block = cls(
             size=block_dict["size"],
             requested_size=block_dict["requested_size"],
             address=block_dict["address"],
-            state=block_dict["state"],
+            state=state,
             frames=[] if _raw_frames else [Frame.from_dict(frame) for frame in frame_dicts],
             _raw_frames=frame_dicts if _raw_frames else None,
         )
@@ -231,7 +273,12 @@ def from_dict(
         ignore_inactive_blocks: bool = False,
         _raw_frames: bool = False,
     ):
+        numeric_fields = ("address", "total_size", "stream", "allocated_size", "active_size")
+        _validate_mapping(segment_dict, "Segment", (*numeric_fields, "segment_type", "blocks"))
+        _validate_numbers(segment_dict, "Segment", numeric_fields)
+        _validate_sequence(segment_dict["blocks"], "Segment blocks")
         frame_dicts = segment_dict.get("frames", [])
+        _validate_sequence(frame_dicts, "Frames")
         if _raw_frames:
             _validate_raw_frames(frame_dicts)
         segment = cls(
@@ -248,6 +295,7 @@ def from_dict(
             is_expandable=segment_dict.get("is_expandable", False),
         )
         for block in segment_dict["blocks"]:
+            _validate_mapping(block, "Block", ("state",))
             if ignore_inactive_blocks and block["state"] == BlockState.INACTIVE:
                 continue
             _block = Block.from_dict(block, _raw_frames=_raw_frames)
@@ -328,9 +376,13 @@ def from_dict(
         ignore_inactive_blocks: bool = False,
         _raw_frames: bool = False,
     ):
+        _validate_mapping(snapshot_dict, "Snapshot")
         segments_dict = snapshot_dict.get("segments", [])
         device_traces = snapshot_dict.get("device_traces", [])
+        _validate_sequence(segments_dict, "Snapshot segments")
+        _validate_sequence(device_traces, "Snapshot device_traces")
         device_trace_list = device_traces[device] if 0 <= device < len(device_traces) else []
+        _validate_sequence(device_trace_list, "Device traces")
         snapshot = cls()
         snapshot.segments = []
         snapshot.trace_entries = []
@@ -339,6 +391,7 @@ def from_dict(
         snapshot.total_activated = 0
         # 读取dump_snapshot时内存状态
         for segment_dict in segments_dict:
+            _validate_mapping(segment_dict, "Segment")
             # 当segment原始数据中缺少device字段明确指向其所属设备时，默认其归属为device0
             # 此时如果from_dict指定device为0或未指定而缺省为0，则未知归属的device也会纳入分析
             if segment_dict.get("device", 0) != device:

src/pt_snap_cli/snapshot/tools/adaptors/snapshot2db.py

@@ -191,8 +191,15 @@ def dump(pickle_file: str | Path, dump_file: str | Path, device: int | None = No
     except Exception as e:
         dump_logger.error(f"Failed to load pickle file: {e}")
         return False
+    if not isinstance(data, dict):
+        raise ValueError("Snapshot must be a mapping")
     # Snapshot format: one list of trace dictionaries per device.
+    if not isinstance(data.get("device_traces", []), (list, tuple)):
+        raise ValueError("Snapshot device_traces must be a list or tuple")
     device_traces = cast(list[list[object]], data.get("device_traces", []))
+    for index, traces in enumerate(device_traces):
+        if not isinstance(traces, (list, tuple)):
+            raise ValueError(f"Snapshot device_traces[{index}] must be a list or tuple")
     # 当指定device为空时dump所有记录了跟踪事件的device，否则仅dump指定device
     need_dump_devices = [device for device in range(len(device_traces)) if device_traces[device]]
     dump_logger.info(f"Recognized have trace events devices {need_dump_devices}.")