Reject unsupported trace inputs and JSONL evidence truncation

README.md

@@ -106,6 +106,13 @@ finding. Do not send the key or an unredacted trace by email.
 
 ## Usage
 
+Local `analyze()` accepts a nonempty Pisama native trace (`spans`), an ATIF
+trajectory, or native span JSONL. OTLP `resourceSpans` exports are not a local
+input format; use the hosted workflow above for those exports. Unsupported or
+empty inputs raise an error rather than reporting a clean analysis.
+JSONL may contain native span rows, or exactly one native trace envelope; mixing
+envelopes with other rows is rejected rather than dropping later evidence.
+
 ```python
 from pisama import analyze
 

src/pisama/_loader.py

@@ -12,6 +12,14 @@
 
 
 def load_trace(input_data: Union[str, dict[str, Any], Trace]) -> Trace:
+    """Load supported trace input, rejecting empty traces before detection."""
+    trace = _load_trace(input_data)
+    if not trace.spans:
+        raise ValueError("Trace contains no spans; no analysis was performed.")
+    return trace
+
+
+def _load_trace(input_data: Union[str, dict[str, Any], Trace]) -> Trace:
     """Load a Trace from various input formats.
 
     Args:
@@ -77,26 +85,69 @@ def _load_dict(data: dict[str, Any]) -> Trace:
     """Load either an ATIF trajectory or Pisama's native trace shape."""
     if is_atif_trajectory(data):
         return trace_from_atif(data)
+    if "resourceSpans" in data:
+        raise ValueError(
+            "OTLP resourceSpans is not supported by local analyze(); "
+            "use hosted ingestion or provide an ATIF/native trace."
+        )
+    if not isinstance(data.get("spans"), list):
+        raise ValueError("Expected an ATIF trajectory or a native trace with a spans list.")
+    for span in data["spans"]:
+        _validate_native_span(span)
     return Trace.from_dict(data)
 
 
+def _validate_native_span(data: Any) -> None:
+    # IDs and all other individual fields are optional in native spans, but
+    # unrelated dictionaries must not silently become default/blank spans.
+    native_fields = {
+        "span_id",
+        "parent_id",
+        "trace_id",
+        "name",
+        "kind",
+        "platform",
+        "platform_metadata",
+        "start_time",
+        "end_time",
+        "status",
+        "error_message",
+        "attributes",
+        "events",
+        "input_data",
+        "output_data",
+    }
+    if (
+        not isinstance(data, dict)
+        or not native_fields.intersection(data)
+        or {"resourceSpans", "spans", "steps"}.intersection(data)
+    ):
+        raise ValueError("Expected a native span object, not an empty object or trace/OTLP export.")
+
+
 def _load_jsonl(text: str) -> Trace:
     """Parse a JSONL file where each line is a span or event."""
     lines = [line.strip() for line in text.splitlines() if line.strip()]
 
     if not lines:
         raise ValueError("JSONL file is empty")
 
-    # If the first line parses as a full trace (has 'trace_id' + 'spans'),
-    # treat the file as a single-line trace dump.
-    first = json.loads(lines[0])
-    if "trace_id" in first and "spans" in first:
-        return Trace.from_dict(first)
+    rows = [json.loads(line) for line in lines]
+    # A trace envelope is supported only as the sole row. Never discard
+    # subsequent evidence or silently flatten envelopes into blank spans.
+    if any(isinstance(row, dict) and "spans" in row for row in rows):
+        if len(rows) != 1:
+            raise ValueError(
+                "A JSONL trace envelope must be the only row; multiple rows cannot be merged."
+            )
+        return _load_dict(rows[0])
 
     # Otherwise, treat each line as a span dict and wrap them.
     from pisama_core.traces.models import Span
 
-    spans = [Span.from_dict(json.loads(line)) for line in lines]
+    for row in rows:
+        _validate_native_span(row)
+    spans = [Span.from_dict(row) for row in rows]
     trace = Trace()
     for span in spans:
         trace.add_span(span)

tests/test_loader_fail_closed.py

@@ -0,0 +1,103 @@
+"""Real parser and CLI checks: unsupported exports cannot look clean."""
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+from pisama_core.traces.models import Trace
+
+from pisama._loader import load_trace
+
+
+@pytest.mark.parametrize("payload", [{"resourceSpans": []}, {}, {"spans": []}, {"spans": [None]}])
+def test_invalid_trace_rejected_from_dict_json_and_file(payload, tmp_path):
+    path = tmp_path / "trace.json"
+    path.write_text(json.dumps(payload))
+    for value in (payload, json.dumps(payload), str(path)):
+        with pytest.raises(ValueError):
+            load_trace(value)
+
+
+def test_empty_trace_object_rejected():
+    with pytest.raises(ValueError, match="no spans"):
+        load_trace(Trace())
+
+
+@pytest.mark.parametrize("suffix", [".json", ".jsonl"])
+def test_otlp_cli_fails_without_clean_message(tmp_path, suffix):
+    path = tmp_path / ("trace" + suffix)
+    path.write_text(
+        json.dumps({"resourceSpans": [{"scopeSpans": [{"spans": [{"name": "synthetic"}]}]}]})
+    )
+    env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")}
+    result = subprocess.run(
+        [sys.executable, "-c", "from pisama.cli.main import main; main()", "analyze", str(path)],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=15,
+    )
+    assert result.returncode == 1
+    assert "OTLP" in result.stderr
+    assert "No issues detected" not in result.stdout + result.stderr
+
+
+def test_native_span_without_id_remains_supported(tmp_path):
+    path = tmp_path / "trace.jsonl"
+    path.write_text(json.dumps({"name": "synthetic-span"}))
+    assert len(load_trace(str(path)).spans) == 1
+
+
+@pytest.mark.parametrize("span", [{}, {"unrelated": "value"}, {"resourceSpans": []}, None, []])
+def test_unsupported_span_shapes_rejected_in_native_and_jsonl(span, tmp_path):
+    with pytest.raises(ValueError):
+        load_trace({"spans": [span]})
+    path = tmp_path / "trace.jsonl"
+    path.write_text(json.dumps(span))
+    with pytest.raises(ValueError):
+        load_trace(str(path))
+
+
+@pytest.mark.parametrize("envelope_first", [True, False])
+@pytest.mark.parametrize("second_envelope", [True, False])
+def test_jsonl_envelope_cannot_discard_other_rows(tmp_path, envelope_first, second_envelope):
+    envelope = {"trace_id": "synthetic", "spans": [{"name": "first"}]}
+    evidence = {"name": "error-span", "status": "error", "error_message": "synthetic failure"}
+    other = {"trace_id": "second", "spans": [evidence]} if second_envelope else evidence
+    rows = [envelope, other] if envelope_first else [other, envelope]
+    path = tmp_path / "trace.jsonl"
+    path.write_text("\n".join(json.dumps(row) for row in rows))
+    with pytest.raises(ValueError, match="only row"):
+        load_trace(str(path))
+
+
+def test_single_native_envelope_and_multiple_native_spans_supported(tmp_path):
+    spans = [{"name": "first"}, {"name": "second", "status": "error"}]
+    path = tmp_path / "trace.jsonl"
+    for payload in (json.dumps({"spans": spans}), "\n".join(map(json.dumps, spans))):
+        path.write_text(payload)
+        loaded = load_trace(str(path))
+        assert len(loaded.spans) == 2
+        assert loaded.spans[1].name == "second"
+
+
+def test_cli_rejects_truncated_jsonl_analysis(tmp_path):
+    path = tmp_path / "trace.jsonl"
+    path.write_text(
+        json.dumps({"trace_id": "synthetic", "spans": [{"name": "first"}]})
+        + "\n"
+        + json.dumps({"name": "later-error", "status": "error"})
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", "from pisama.cli.main import main; main()", "analyze", str(path)],
+        capture_output=True,
+        text=True,
+        env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")},
+        timeout=15,
+    )
+    assert result.returncode == 1
+    assert "only row" in result.stderr
+    assert "No issues detected" not in result.stdout + result.stderr