fix(session): bound _wait_for_previous_archive_done to prevent permanent SessionCommit queue stall

openviking/session/session.py

@@ -9,6 +9,7 @@
 import inspect
 import json
 import re
+import time
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Literal, Optional
@@ -68,6 +69,7 @@
 logger = get_logger(__name__)
 
 _ARCHIVE_WAIT_POLL_SECONDS = 0.1
+_ARCHIVE_WAIT_TIMEOUT_SECONDS = 1800.0
 _PHASE2_QUEUE_WAIT_TIMEOUT_SECONDS = 1800.0
 _MEMORY_EXTRACTION_MAX_RETRIES = 3
 _MEMORY_EXTRACTION_RETRY_BASE_DELAY_SECONDS = 1.0
@@ -3743,11 +3745,32 @@ def _archive_index_from_uri(archive_uri: str) -> int:
             raise ValueError(f"Invalid archive URI: {archive_uri}")
         return int(match.group(1))
 
-    async def _wait_for_previous_archive_done(self, archive_index: int) -> bool:
-        """Wait until every earlier archive reaches a terminal state."""
+    async def _wait_for_previous_archive_done(
+        self,
+        archive_index: int,
+        timeout: float = _ARCHIVE_WAIT_TIMEOUT_SECONDS,
+    ) -> bool:
+        """Wait until every earlier archive reaches a terminal state.
+
+        The wait is bounded: an earlier archive can stay pending forever when
+        its Phase 1 completed (``phase1.status=ready``) but the matching queue
+        item was lost (process crash, upgrade, or a historical enqueue bug).
+        Such an orphan is invisible from here, so after ``timeout`` seconds of
+        polling the still-pending predecessors are marked terminally failed and
+        a ``TimeoutError`` is raised. The caller's failure path then writes the
+        current archive's ``.failed.json`` and fails the task, so the worker
+        slot is released instead of being wedged forever; the next commit
+        replays the raw messages of every failed archive through the existing
+        ``_prepare_phase2_archive_messages`` roll-forward, so no data is lost.
+
+        Raises:
+            TimeoutError: earlier archives were still pending after ``timeout``
+                seconds; they have been marked failed for later raw replay.
+        """
         if archive_index <= 1 or not self._viking_fs:
             return True
 
+        deadline = time.monotonic() + timeout
         while True:
             earlier_states = [
                 state for state in await self._scan_archive_states() if state.index < archive_index
@@ -3775,6 +3798,32 @@ async def _wait_for_previous_archive_done(self, archive_index: int) -> bool:
                 reconciled = True
             if reconciled:
                 continue
+            if time.monotonic() >= deadline:
+                stuck_ids = [state.archive_id for state in pending_states]
+                for state in pending_states:
+                    logger.error(
+                        "Archive %s stayed pending for over %.0fs while blocking "
+                        "archive_%03d Phase 2; its queue item is presumed lost. "
+                        "Marking it failed so its raw messages replay into a "
+                        "later commit.",
+                        state.archive_id,
+                        timeout,
+                        archive_index,
+                    )
+                    await self._write_failed_marker(
+                        state.archive_uri,
+                        stage="phase2_wait_timeout",
+                        error=(
+                            f"Archive stayed pending for over {timeout:.0f}s; "
+                            "its Phase 2 queue item is presumed lost"
+                        ),
+                    )
+                raise TimeoutError(
+                    f"Timed out after {timeout:.0f}s waiting for earlier archives "
+                    f"{stuck_ids} to finish before archive_{archive_index:03d} "
+                    "Phase 2; they were marked failed for raw replay by a later "
+                    "commit"
+                )
             await asyncio.sleep(_ARCHIVE_WAIT_POLL_SECONDS)
 
     async def _prepare_phase2_archive_messages(

tests/session/test_session_retention_integration.py

@@ -357,6 +357,45 @@ async def test_phase2_waits_for_all_earlier_pending_archives(
     assert await asyncio.wait_for(waiter, timeout=0.5)
 
 
+async def test_phase2_wait_times_out_and_fails_orphaned_pending_archive(
+    client: AsyncOpenViking,
+    monkeypatch,
+):
+    """A pending predecessor whose queue item was lost must not stall forever.
+
+    After the bounded wait expires, the orphan is marked terminally failed for
+    raw replay and the current commit fails cleanly instead of wedging its
+    worker slot (see #3396).
+    """
+    session = client.session(session_id="phase2_wait_timeout_orphan_test")
+    await session.ensure_exists()
+    first_uri = await _write_archive(
+        session,
+        1,
+        [_text_message("u1", "user", "orphaned pending one")],
+    )
+    monkeypatch.setattr("openviking.session.session._ARCHIVE_WAIT_POLL_SECONDS", 0.01)
+
+    with pytest.raises(TimeoutError, match="archive_001"):
+        await asyncio.wait_for(
+            session._wait_for_previous_archive_done(2, timeout=0.05),
+            timeout=2.0,
+        )
+
+    failed = json.loads(
+        await session._viking_fs.read_file(f"{first_uri}/.failed.json", ctx=session.ctx)
+    )
+    assert failed["stage"] == "phase2_wait_timeout"
+    states = {state.archive_id: state.state for state in await session._scan_archive_states()}
+    assert states["archive_001"] == "failed"
+
+    # The next commit's wait now sees only terminal predecessors and proceeds.
+    assert await asyncio.wait_for(
+        session._wait_for_previous_archive_done(2, timeout=0.05),
+        timeout=2.0,
+    )
+
+
 async def test_missing_previous_archive_directory_does_not_block_phase2(
     client: AsyncOpenViking,
 ):