fix: persist advance_in_progress to prevent cross-process session skip

tests/test_service.py

@@ -526,6 +526,101 @@ def test_pending_stop_confirmation_persists_across_process_restart(settings, ser
     assert second._session._should_advance_session(restarted_session, second.playback_snapshot()) is True
 
 
+def test_advance_in_progress_persisted_blocks_cross_process_worker(settings, service, tmp_path) -> None:
+    # When a separate process (e.g. `vesper ask`) starts a session track, it
+    # persists advance_in_progress=True so the long-lived server's background
+    # worker does not see the queue-clear stop as a finished track and advance
+    # prematurely. This reproduces the skip bug: without the persisted flag,
+    # the worker sees is_playing=false during the CLI's startup and advances.
+    # See #114.
+    database_path = tmp_path / "cross-process-advance.db"
+    rpc = service._rpc.__class__()
+    first = CiderAgentService(
+        Settings(
+            http_host=settings.http_host,
+            http_port=settings.http_port,
+            public_base_url=settings.public_base_url,
+            cider_base_url=settings.cider_base_url,
+            cider_api_token=settings.cider_api_token,
+            default_search_source=settings.default_search_source,
+            resolver_backend=settings.resolver_backend,
+            resolver_base_url=settings.resolver_base_url,
+            resolver_model=settings.resolver_model,
+            resolver_api_key=settings.resolver_api_key,
+            resolver_include_reasoning=settings.resolver_include_reasoning,
+            resolver_include_raw_output=settings.resolver_include_raw_output,
+            response_detail=settings.response_detail,
+            session_recent_tracks_limit=settings.session_recent_tracks_limit,
+            global_recent_tracks_limit=settings.global_recent_tracks_limit,
+            request_timeout_seconds=settings.request_timeout_seconds,
+            verify_tls=settings.verify_tls,
+            log_level=settings.log_level,
+            database_path=database_path,
+            config_path=settings.config_path,
+        ),
+        rpc_client=rpc,
+        preference_store=PreferenceStore(database_path),
+        resolver=service._resolver.__class__(),
+    )
+    first.play_session("play upbeat music")
+    session = first._preferences.get_active_session()
+    assert session is not None
+
+    # Simulate a second process starting a session track: it sets and persists
+    # advance_in_progress=True (as _play_session_track now does at its start).
+    first._session._set_session_runtime(session["id"], advance_in_progress=True)
+    first._preferences.upsert_session_runtime(session["id"], advance_in_progress=True)
+
+    # The server's worker observes a stopped snapshot (queue was cleared).
+    rpc.is_playing = False
+    rpc.current_track = None
+    first._session._set_session_runtime(session["id"], last_advance_at=0.0)
+    first._preferences.upsert_session_runtime(session["id"], last_advance_at="1970-01-01T00:00:00+00:00")
+
+    # The worker must NOT advance: advance_in_progress is persisted True by
+    # the other process and _effective_session_runtime reads it.
+    assert first._session._should_advance_session(session, first.playback_snapshot()) is False
+
+    # Once the other process finishes and clears the flag, the worker can
+    # proceed (after the two-snapshot confirmation, as before). Clear the
+    # pending-stop confirmation first so the two-phase test is independent.
+    first._session._clear_pending_stop_confirmation(session["id"])
+    first._session._set_session_runtime(session["id"], advance_in_progress=False)
+    first._preferences.upsert_session_runtime(session["id"], advance_in_progress=False)
+    # First stopped snapshot arms the confirmation.
+    assert first._session._should_advance_session(session, first.playback_snapshot()) is False
+    # Second stopped snapshot confirms and advances.
+    assert first._session._should_advance_session(session, first.playback_snapshot()) is True
+
+
+def test_session_worker_does_not_advance_when_track_played_below_min_duration(service) -> None:
+    # Minimum-play-duration backstop: if the current track has a
+    # current_playback_time below SESSION_MIN_PLAY_SECONDS, don't advance even
+    # if Cider reports stopped. This catches buffering/startup noise where
+    # Cider briefly reports is_playing=false. See #114.
+    service.play_session("play upbeat music")
+    session = service._preferences.get_active_session()
+    assert session is not None
+
+    service._rpc.is_playing = False
+    # Track has barely started playing (3s, below the 10s threshold). Use an
+    # ambiguous remainingTime so the track would otherwise pass through to
+    # the two-snapshot confirmation path.
+    service._rpc.current_track["attributes"]["currentPlaybackTime"] = 3
+    service._rpc.current_track["attributes"]["remainingTime"] = 0
+    service._session._set_session_runtime(session["id"], last_advance_at=0.0)
+    service._preferences.upsert_session_runtime(session["id"], last_advance_at="1970-01-01T00:00:00+00:00")
+
+    assert service._session._should_advance_session(session, service.playback_snapshot()) is False
+
+    # Once playback time exceeds the threshold, the min-duration guard no
+    # longer blocks and the normal stop-confirmation logic resumes. With a
+    # genuinely-finished track (currentPlaybackTime past duration), the
+    # finished state skips the two-snapshot confirmation and advances.
+    service._rpc.current_track["attributes"]["currentPlaybackTime"] = 180
+    assert service._session._should_advance_session(session, service.playback_snapshot()) is True
+
+
 def test_preference_store_backfills_pending_stop_columns(tmp_path) -> None:
     import sqlite3
 
@@ -560,6 +655,41 @@ def test_preference_store_backfills_pending_stop_columns(tmp_path) -> None:
     assert store_runtime["pending_stop_track_id"] == "<missing>"
 
 
+def test_preference_store_backfills_advance_in_progress_column(tmp_path) -> None:
+    import sqlite3
+
+    database_path = tmp_path / "legacy-runtime-no-advance.db"
+    # Simulate a database written before advance_in_progress existed: the
+    # runtime table has pending_stop columns but lacks advance_in_progress.
+    with sqlite3.connect(database_path) as conn:
+        conn.execute(
+            """
+            CREATE TABLE session_runtime (
+                session_id INTEGER PRIMARY KEY,
+                active_intent TEXT NOT NULL DEFAULT 'active',
+                last_advance_at TEXT,
+                last_selected_track_id TEXT,
+                last_known_playback_state TEXT,
+                pending_stop_track_id TEXT,
+                pending_stop_observed_at TEXT,
+                updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+            )
+            """
+        )
+        conn.execute("INSERT INTO session_runtime (session_id) VALUES (1)")
+
+    # Instantiating the store runs the schema migration, adding the column.
+    store = PreferenceStore(database_path)
+    runtime = store.get_session_runtime(1)
+    assert runtime is not None
+    # Backfilled column defaults to False (0), never None.
+    assert runtime["advance_in_progress"] is False
+
+    # The upsert must work on the migrated schema.
+    store.upsert_session_runtime(1, advance_in_progress=True)
+    assert store.get_session_runtime(1)["advance_in_progress"] is True
+
+
 def test_active_session_reconcile_tolerates_empty_now_playing_info_list(settings, service) -> None:
     service._preferences.start_session(request_text="play upbeat music")
 

tests/test_storage.py

@@ -221,6 +221,7 @@ def test_upsert_session_runtime_preserves_omitted_fields(settings) -> None:
         last_advance_at="1970-01-01T00:00:00+00:00",
         last_selected_track_id="track-1",
         last_known_playback_state="playing",
+        advance_in_progress=True,
     )
     # Update only one field; the rest must be preserved.
     store.upsert_session_runtime(session["id"], last_selected_track_id="track-2")
@@ -231,6 +232,24 @@ def test_upsert_session_runtime_preserves_omitted_fields(settings) -> None:
     assert runtime["last_advance_at"] == "1970-01-01T00:00:00+00:00"
     assert runtime["last_known_playback_state"] == "playing"
     assert runtime["last_selected_track_id"] == "track-2"
+    # advance_in_progress was omitted (None) in the second upsert, so it must
+    # preserve the True value set in the seed. See #114.
+    assert runtime["advance_in_progress"] is True
+
+
+def test_upsert_session_runtime_advances_in_progress_can_be_cleared(settings) -> None:
+    # advance_in_progress is a bool stored as INTEGER NOT NULL DEFAULT 0. The
+    # COALESCE pattern must allow setting it False (0) after it was True,
+    # because 0 is non-NULL and is written as-is (unlike None which preserves).
+    # This is what _play_session_track does at the end of an advance. See #114.
+    store = PreferenceStore(settings.database_path)
+    session = store.start_session(request_text="play some music")
+
+    store.upsert_session_runtime(session["id"], advance_in_progress=True)
+    assert store.get_session_runtime(session["id"])["advance_in_progress"] is True
+
+    store.upsert_session_runtime(session["id"], advance_in_progress=False)
+    assert store.get_session_runtime(session["id"])["advance_in_progress"] is False
 
 
 def test_upsert_session_runtime_defaults_active_intent_for_new_row(settings) -> None:

vesper/service.py

@@ -47,6 +47,13 @@ class CiderAgentService:
 
     SESSION_REFILL_INTERVAL_SECONDS = 5.0
     SESSION_ADVANCE_COOLDOWN_SECONDS = 5.0
+    # Minimum seconds a track must have played before an auto-advance is
+    # considered. This is a backstop against Cider reporting is_playing=false
+    # during buffering/startup (when current_playback_time is near zero), which
+    # could otherwise satisfy the stop-confirmation and trigger a premature
+    # skip. The primary guard is the cross-process advance_in_progress flag,
+    # but this catches cases where Cider's playback reporting is noisy. See #114.
+    SESSION_MIN_PLAY_SECONDS = 10.0
     TRACK_SELECTION_POOL_SIZE = 3
     SESSION_SEARCH_RESULT_LIMIT = 100
     SESSION_SEARCH_PAGE_LIMIT = 50

vesper/session.py

@@ -172,6 +172,7 @@ def _fetch_session_source_results(
     SESSION_SEARCH_PAGE_LIMIT: int
     SESSION_STOREFRONT: str
     SESSION_ADVANCE_COOLDOWN_SECONDS: float
+    SESSION_MIN_PLAY_SECONDS: float
     PREFERENCE_SEED_SEARCH_LIMIT: int
     PREFERENCE_SEED_ARTIST_CAP: int
     PREFERENCE_SEED_POOL_QUERY: str
@@ -723,7 +724,20 @@ def _play_session_track(self, session: dict[str, Any], *, selection_strategy: st
                 pending_stop_track_id=None,
                 pending_stop_observed_at=None,
             )
-            self._persist_session_runtime(session["id"], suspended=False, last_advance_at=self._host.current_timestamp())
+            # Persist last_advance_at AND advance_in_progress at the start, not
+            # the end, so a separate process's worker sees the in-flight
+            # advance (cross-process coordination) and the cooldown is measured
+            # from when the advance began rather than when it finished. The
+            # previous behavior persisted last_advance_at only at the end, so a
+            # `vesper ask` CLI process clearing Cider's queue and starting a
+            # track left a stale timestamp that the server worker's cooldown
+            # check could not catch. See #114.
+            self._persist_session_runtime(
+                session["id"],
+                suspended=False,
+                last_advance_at=self._host.current_timestamp(),
+                advance_in_progress=True,
+            )
             try:
                 self._check_worker_cancelled()
                 playback_started_at = time.perf_counter()
@@ -809,6 +823,7 @@ def _play_session_track(self, session: dict[str, Any], *, selection_strategy: st
                     last_advance_at=self._host.current_timestamp(),
                     last_selected_track_id=_clean_id(lead_track.get("play_params", {}).get("id")),
                     last_known_playback_state="playing",
+                    advance_in_progress=False,
                 )
                 result = {
                     "status": "ok",
@@ -835,8 +850,11 @@ def _play_session_track(self, session: dict[str, Any], *, selection_strategy: st
                 # unexpected) leaving ``_play_session_track`` must clear
                 # ``advance_in_progress`` so the session runtime is never
                 # wedged, then re-raise so the original error propagates
-                # unchanged. See issue #47.
+                # unchanged. See issue #47. Persist the clear too, so a crash
+                # in a separate process (e.g. `vesper ask`) does not leave the
+                # flag wedged True for another process's worker. See #114.
                 self._set_session_runtime(session["id"], advance_in_progress=False, planning_playback_snapshot=None)
+                self._persist_session_runtime(session["id"], advance_in_progress=False)
                 raise
 
     def _plan_session_query(self, session: dict[str, Any], *, count: int, force_replan: bool = False) -> SessionQueryPlan:

vesper/session_runtime.py

@@ -140,13 +140,31 @@ def _should_advance_session(self, session: dict[str, Any], playback: dict[str, A
             "track_state": track_state,
             "seconds_since_last_advance": elapsed,
             "cooldown_seconds": self._host.SESSION_ADVANCE_COOLDOWN_SECONDS,
+            "min_play_seconds": self._host.SESSION_MIN_PLAY_SECONDS,
         }
         if runtime.get("suspended"):
             self._log_session_auto_advance_decision(debug_payload, advance=False, blocked_by="session_suspended")
             return False
         if runtime.get("advance_in_progress"):
             self._log_session_auto_advance_decision(debug_payload, advance=False, blocked_by="advance_in_progress")
             return False
+        # Minimum-play-duration backstop. If the current track has a
+        # current_playback_time below the threshold, it has not played long
+        # enough to be genuinely finished. Cider can report is_playing=false
+        # during buffering/startup with a near-zero playback time; without
+        # this guard that could satisfy the stop-confirmation and trigger a
+        # premature skip. The primary guard is the cross-process
+        # advance_in_progress flag; this catches noisy playback reporting. See #114.
+        current_playback_time = self._numeric_value(track.get("current_playback_time"))
+        if current_playback_time is not None and current_playback_time < self._host.SESSION_MIN_PLAY_SECONDS:
+            self._clear_pending_stop_confirmation(session["id"])
+            debug_payload["runtime_after_clear"] = self._session_runtime_confirmation_state(session["id"])
+            self._log_session_auto_advance_decision(
+                debug_payload,
+                advance=False,
+                blocked_by="min_play_duration_not_met",
+            )
+            return False
         is_playing = playback.get("is_playing")
         if is_playing is True:
             self._clear_pending_stop_confirmation(session["id"])
@@ -290,6 +308,14 @@ def _effective_session_runtime(self, session_id: int) -> dict[str, Any]:
         if stored:
             runtime["pending_stop_track_id"] = stored.get("pending_stop_track_id")
             runtime["pending_stop_observed_at"] = stored.get("pending_stop_observed_at")
+        # advance_in_progress is cross-process state: a separate process (e.g.
+        # the `vesper ask` CLI) starting a session track sets it so this
+        # process's worker does not see the queue-clear stop as a finished
+        # track and advance prematurely. Persisted value is authoritative so a
+        # stale in-memory "not in progress" can't override another process's
+        # in-flight advance. See #114.
+        if stored:
+            runtime["advance_in_progress"] = bool(stored.get("advance_in_progress"))
         return runtime
 
     def _get_session_runtime(self, session_id: int) -> dict[str, Any]:
@@ -356,6 +382,7 @@ def _persist_session_runtime(
         last_selected_track_id: str | None = None,
         last_known_playback_state: str | None = None,
         preserve_last_advance: bool = False,
+        advance_in_progress: bool | None = None,
     ) -> None:
         runtime = self._preferences.get_session_runtime(session_id)
         resolved_last_advance_at = runtime.get("last_advance_at") if runtime and preserve_last_advance else last_advance_at
@@ -365,6 +392,7 @@ def _persist_session_runtime(
             last_advance_at=resolved_last_advance_at,
             last_selected_track_id=last_selected_track_id,
             last_known_playback_state=last_known_playback_state,
+            advance_in_progress=advance_in_progress,
         )
 
     def _seconds_since_runtime_timestamp(self, value: Any) -> float | None:

vesper/storage/__init__.py

@@ -204,6 +204,7 @@ def upsert_session_runtime(
         last_advance_at: str | None = None,
         last_selected_track_id: str | None = None,
         last_known_playback_state: str | None = None,
+        advance_in_progress: bool | None = None,
     ) -> dict[str, Any]:
         return session_data.upsert_session_runtime(
             self._database_path,
@@ -212,6 +213,7 @@ def upsert_session_runtime(
             last_advance_at=last_advance_at,
             last_selected_track_id=last_selected_track_id,
             last_known_playback_state=last_known_playback_state,
+            advance_in_progress=advance_in_progress,
         )
 
     def update_session_pending_stop(

vesper/storage/schema.py

@@ -188,6 +188,13 @@ def initialize(database_path: Path) -> None:
                 pending_stop_track_id TEXT,
                 -- Same UTC ISO-8601 wall-clock format as last_advance_at.
                 pending_stop_observed_at TEXT,
+                -- Cross-process advance coordination. A separate process
+                -- (e.g. the `vesper ask` CLI) starting a session track sets
+                -- this so the long-lived server's background worker does not
+                -- see the queue-clear stop as a finished track and advance
+                -- prematurely. Mirrors the in-memory advance_in_progress
+                -- flag so a second process can observe it. See #114.
+                advance_in_progress INTEGER NOT NULL DEFAULT 0,
                 updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                 FOREIGN KEY(session_id) REFERENCES sessions(id)
             )
@@ -264,3 +271,5 @@ def ensure_session_runtime_columns(connection: sqlite3.Connection) -> None:
         connection.execute("ALTER TABLE session_runtime ADD COLUMN pending_stop_track_id TEXT")
     if "pending_stop_observed_at" not in existing:
         connection.execute("ALTER TABLE session_runtime ADD COLUMN pending_stop_observed_at TEXT")
+    if "advance_in_progress" not in existing:
+        connection.execute("ALTER TABLE session_runtime ADD COLUMN advance_in_progress INTEGER NOT NULL DEFAULT 0")