Skip zero shot sessions during cloud upload

src/openflight/cloud/commands.py

@@ -128,6 +128,7 @@ def cmd_push(
         "offline": False,
         "needs_relink": False,
         "dry_run": dry_run,
+        "skipped_empty": 0,
     }
 
     if not dry_run and not config.is_active():
@@ -158,11 +159,23 @@ def cmd_push(
         # keep the (tiny) shot lines. Loading it whole would risk OOM on a Pi.
         result = filtering.filter_session_file(path, config.device_id)
         session_id = result.session_id
+        shot_count = result.kept_type_counts.get("shot_detected", 0)
+        session_ended = result.kept_type_counts.get("session_end", 0) > 0
 
         if dry_run:
             _describe_dry_run(path.name, session_id, result, out)
             continue
 
+        if shot_count == 0:
+            if session_ended:
+                spool.mark_skipped(path, reason="no_shots", shot_count=0)
+                summary["skipped_empty"] += 1
+                out(f"{path.name}: skipped (0 shots).")
+            else:
+                summary["deferred"] += 1
+                out(f"{path.name}: waiting for shots before upload.")
+            continue
+
         try:
             body = filtering.build_upload_body(result)
         except filtering.BodyTooLargeError as exc:

src/openflight/cloud/spool.py

@@ -19,6 +19,7 @@
 
 PUSHED_SUFFIX = ".pushed"
 PARKED_SUFFIX = ".parked"
+SKIPPED_SUFFIX = ".skipped"
 STATE_SUFFIX = ".state"
 
 SESSION_GLOB = "session_*.jsonl"
@@ -59,9 +60,18 @@ def is_parked(path: Path) -> bool:
     return _sidecar(path, PARKED_SUFFIX).exists()
 
 
+def is_skipped(path: Path) -> bool:
+    """True if a ``.skipped`` marker exists for this session."""
+    return _sidecar(path, SKIPPED_SUFFIX).exists()
+
+
 def pending_sessions(log_dir: Path) -> List[Path]:
-    """Session files that are neither pushed nor parked."""
-    return [p for p in session_files(log_dir) if not is_pushed(p) and not is_parked(p)]
+    """Session files that are not in a terminal upload state."""
+    return [
+        p
+        for p in session_files(log_dir)
+        if not is_pushed(p) and not is_parked(p) and not is_skipped(p)
+    ]
 
 
 def read_attempts(path: Path) -> int:
@@ -150,6 +160,17 @@ def mark_parked(path: Path, reason: str, attempts: int, last_error: Optional[str
     )
 
 
+def mark_skipped(path: Path, reason: str, shot_count: Optional[int] = None) -> None:
+    """Mark a session as intentionally skipped by upload policy."""
+    _write_json(
+        _sidecar(path, SKIPPED_SUFFIX),
+        {"reason": reason, "shot_count": shot_count, "skipped_at": _now()},
+    )
+    state_path = _sidecar(path, STATE_SUFFIX)
+    if state_path.exists():
+        state_path.unlink()
+
+
 def clear_markers(path: Path, include_pushed: bool = False) -> List[str]:
     """Remove terminal/retry markers so a session becomes pending again.
 
@@ -158,7 +179,7 @@ def clear_markers(path: Path, include_pushed: bool = False) -> List[str]:
     the server already stored (idempotent server-side). Returns the suffixes
     actually removed.
     """
-    suffixes = [PARKED_SUFFIX, STATE_SUFFIX]
+    suffixes = [PARKED_SUFFIX, SKIPPED_SUFFIX, STATE_SUFFIX]
     if include_pushed:
         suffixes.append(PUSHED_SUFFIX)
     cleared: List[str] = []
@@ -175,10 +196,14 @@ def summarize(log_dir: Path) -> Dict[str, int]:
     files = session_files(log_dir)
     pushed = sum(1 for p in files if is_pushed(p))
     parked = sum(1 for p in files if is_parked(p))
-    pending = sum(1 for p in files if not is_pushed(p) and not is_parked(p))
+    skipped = sum(1 for p in files if is_skipped(p))
+    pending = sum(
+        1 for p in files if not is_pushed(p) and not is_parked(p) and not is_skipped(p)
+    )
     return {
         "total": len(files),
         "pushed": pushed,
         "parked": parked,
+        "skipped": skipped,
         "pending": pending,
     }

tests/test_cloud_commands.py

@@ -98,8 +98,18 @@ def test_dry_run_does_not_upload_or_mark(self, tmp_path):
         assert "rolling_buffer_capture" not in printed
 
     def test_relink_stops_and_flags(self, tmp_path):
-        _write_session(tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"})
-        _write_session(tmp_path, "session_b.jsonl", {"type": "session_start", "session_uuid": "b"})
+        _write_session(
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
+        )
+        _write_session(
+            tmp_path,
+            "session_b.jsonl",
+            {"type": "session_start", "session_uuid": "b"},
+            {"type": "shot_detected", "ball_speed_mph": 91},
+        )
         client = FakeClient(
             uploads=[UploadResult(401, action="relink", reason="invalid_or_revoked_token")]
         )
@@ -110,15 +120,21 @@ def test_relink_stops_and_flags(self, tmp_path):
 
     def test_park_on_422(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         client = FakeClient(uploads=[UploadResult(422, action="park", reason="invalid_gzip")])
         commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None)
         assert spool.is_parked(path)
 
     def test_quota_sets_cooldown_not_park(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         client = FakeClient(uploads=[UploadResult(402, action="quota", reason="quota_exceeded")])
         commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None)
@@ -127,7 +143,10 @@ def test_quota_sets_cooldown_not_park(self, tmp_path):
 
     def test_5xx_records_failure_and_leaves_pending(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         client = FakeClient(uploads=[UploadResult(503, action="retry")])
         commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None)
@@ -149,19 +168,57 @@ def test_oversize_body_parks(self, tmp_path, monkeypatch):
 
     def test_skips_sessions_in_cooldown(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         spool.record_cooldown(path, "quota_exceeded", seconds=spool.QUOTA_COOLDOWN_S)
         client = FakeClient(uploads=[UploadResult(201, action="success")])
         result = commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None)
         assert client.uploaded == []
         assert result["deferred"] == 1
 
+    def test_ended_zero_shot_session_is_skipped_not_uploaded(self, tmp_path):
+        path = _write_session(
+            tmp_path,
+            "session_empty.jsonl",
+            {"type": "session_start", "session_uuid": "empty"},
+            {"type": "session_end"},
+        )
+        client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=0)])
+        out = []
+        result = commands.cmd_push(_linked_config(), tmp_path, client, out=out.append)
+
+        assert client.uploaded == []
+        assert spool.is_skipped(path)
+        assert result["skipped_empty"] == 1
+        assert "skipped (0 shots)" in "\n".join(out)
+
+    def test_active_zero_shot_session_waits_for_later_shots(self, tmp_path):
+        path = _write_session(
+            tmp_path,
+            "session_active.jsonl",
+            {"type": "session_start", "session_uuid": "active"},
+        )
+        client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=0)])
+        out = []
+        result = commands.cmd_push(_linked_config(), tmp_path, client, out=out.append)
+
+        assert client.uploaded == []
+        assert not spool.is_skipped(path)
+        assert path in spool.pending_sessions(tmp_path)
+        assert result["deferred"] == 1
+        assert "waiting for shots" in "\n".join(out)
+
 
 class TestPushRetry:
     def test_retry_all_reuploads_parked_session(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         spool.mark_parked(path, reason="max_attempts", attempts=20, last_error="503")
         client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=1)])
@@ -174,7 +231,10 @@ def test_retry_all_reuploads_parked_session(self, tmp_path):
 
     def test_retry_all_reuploads_cooled_down_session(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_a.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         spool.record_cooldown(path, "quota_exceeded", seconds=spool.QUOTA_COOLDOWN_S)
         client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=1)])
@@ -193,7 +253,10 @@ def test_retry_all_leaves_pushed_sessions_alone(self, tmp_path):
 
     def test_retry_named_session_force_reuploads_pushed(self, tmp_path):
         path = _write_session(
-            tmp_path, "session_20260527_x.jsonl", {"type": "session_start", "session_uuid": "a"}
+            tmp_path,
+            "session_20260527_x.jsonl",
+            {"type": "session_start", "session_uuid": "a"},
+            {"type": "shot_detected", "ball_speed_mph": 90},
         )
         spool.mark_pushed(path, "a", 0)  # previously "uploaded" with 0 shots
         client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=5)])

tests/test_cloud_spool.py

@@ -29,8 +29,10 @@ def test_pending_excludes_pushed_and_parked(self, tmp_path):
         a = _session(tmp_path, "session_a.jsonl")
         b = _session(tmp_path, "session_b.jsonl")
         c = _session(tmp_path, "session_c.jsonl")
+        d = _session(tmp_path, "session_d.jsonl")
         spool.mark_pushed(b, session_id="id-b", shot_count=3)
         spool.mark_parked(c, reason="quota_exceeded", attempts=20, last_error="402")
+        spool.mark_skipped(d, reason="no_shots", shot_count=0)
         pending = {p.name for p in spool.pending_sessions(tmp_path)}
         assert pending == {"session_a.jsonl"}
         assert a  # referenced
@@ -68,6 +70,25 @@ def test_mark_parked_creates_sidecar(self, tmp_path):
         assert spool.is_parked(path)
 
 
+class TestSkippedMarker:
+    def test_mark_skipped_creates_sidecar(self, tmp_path):
+        path = _session(tmp_path)
+        spool.mark_skipped(path, reason="no_shots", shot_count=0)
+        marker = tmp_path / (path.name + ".skipped")
+        assert marker.exists()
+        data = json.loads(marker.read_text())
+        assert data["reason"] == "no_shots"
+        assert data["shot_count"] == 0
+        assert spool.is_skipped(path)
+
+    def test_mark_skipped_clears_attempt_state(self, tmp_path):
+        path = _session(tmp_path)
+        spool.record_failure(path, "5xx")
+        spool.mark_skipped(path, reason="no_shots", shot_count=0)
+        assert spool.read_attempts(path) == 0
+        assert not (tmp_path / (path.name + ".state")).exists()
+
+
 class TestAttemptCounter:
     def test_attempts_start_at_zero(self, tmp_path):
         path = _session(tmp_path)
@@ -117,13 +138,16 @@ def test_clears_parked_and_state_keeps_pushed_by_default(self, tmp_path):
         path = _session(tmp_path)
         spool.mark_pushed(path, "id", 1)
         spool.mark_parked(path, reason="r", attempts=2, last_error="e")
+        spool.mark_skipped(path, reason="no_shots", shot_count=0)
         spool.record_failure(path, "e")  # writes .state
         cleared = spool.clear_markers(path)
         assert spool.PARKED_SUFFIX in cleared
+        assert spool.SKIPPED_SUFFIX in cleared
         assert spool.STATE_SUFFIX in cleared
         assert spool.PUSHED_SUFFIX not in cleared
         assert spool.is_pushed(path)
         assert not spool.is_parked(path)
+        assert not spool.is_skipped(path)
 
     def test_include_pushed_clears_everything(self, tmp_path):
         path = _session(tmp_path)
@@ -151,9 +175,11 @@ def test_summarizes_counts(self, tmp_path):
         c = _session(tmp_path, "session_c.jsonl")
         spool.mark_pushed(a, session_id="a", shot_count=1)
         spool.mark_parked(b, reason="r", attempts=20, last_error="402")
+        spool.mark_skipped(c, reason="no_shots", shot_count=0)
         assert c
         summary = spool.summarize(tmp_path)
         assert summary["pushed"] == 1
         assert summary["parked"] == 1
-        assert summary["pending"] == 1
+        assert summary["skipped"] == 1
+        assert summary["pending"] == 0
         assert summary["total"] == 3