sched: entry-level history cursor so tied occurrence entries paginate exactly once (B4)

src/wf_scheduling/models.py

@@ -164,6 +164,16 @@ class OccurrenceRecord(BaseModel):
             "checkpoint id and therefore a new entry."
         ),
     )
+    seq: int | None = Field(
+        default=None,
+        description=(
+            "History-entry ordinal assigned at append time, for stable "
+            "pagination across multiple entries of one occurrence "
+            "(admission and stopped entries share resolved_at and "
+            "occurrence_id). Rows persisted before this field existed "
+            "carry None and order by file position."
+        ),
+    )
     interval_start: datetime | None = None
     interval_end: datetime | None = None
     interval_count: int = 0

src/wf_scheduling/store.py

@@ -7,8 +7,9 @@ Layout per schedule ``<root>/schedules/<id>/``:
 - ``candidate.json`` — at most one pending latest candidate.
 - ``consumed.json`` — ``{"consumed_through": <utc iso>}`` watermark.
 - ``history.json`` — occurrence-history list for cursor pagination over
-  ``(resolved_at, occurrence_id)`` (keyset; integer offsets accepted as a
+  ``(resolved_at, occurrence_id, seq)`` (keyset; integer offsets accepted as a
-  legacy fallback).
+  legacy fallback, as are two-part ``"<iso>|<id>"`` cursors issued before
+  the entry ordinal existed).
 - ``_poll_cursor.json`` — round-robin fairness cursor (persisted).
 
 Per-file writes are atomic (tmp + rename); the lock is process-local only.
@@ -161,10 +162,26 @@ class FileScheduleStore:
 
     # -- history with cursor pagination -----------------------------------
     def append_history(self, record: OccurrenceRecord) -> None:
-        """Append one occurrence-history entry for a schedule."""
+        """Append one occurrence-history entry for a schedule.
+
+        Stamps the entry ordinal (one past the highest effective ordinal so
+        far) so pagination distinguishes multiple entries sharing one
+        occurrence's ``(resolved_at, occurrence_id)`` — admission and each
+        stopped result each keep their own position.
+        """
         with self._lock:
             path = self._history_path(record.schedule_id)
             entries = self._read_history_locked(record.schedule_id)
+            highest = -1
+            for index, item in enumerate(entries):
+                ordinal = item.get("seq")
+                effective = (
+                    ordinal if isinstance(ordinal, int) else index
+                )
+                if effective > highest:
+                    highest = effective
+            if record.seq is None:
+                record = record.model_copy(update={"seq": highest + 1})
             entries.append(record.model_dump(mode="json"))
             self._write_json(path, entries)
 
@@ -193,12 +210,23 @@ class FileScheduleStore:
 
     @staticmethod
     def _sort_key(
-        item: OccurrenceRecord,
+        item: tuple[OccurrenceRecord, int],
-    ) -> tuple[bool, datetime, str]:
+    ) -> tuple[bool, datetime, str, float]:
+        """Order one history entry behind its stable entry ordinal.
+
+        Admission and stopped entries for one occurrence share
+        ``(resolved_at, occurrence_id)`` by construction; the ordinal (the
+        stamped ``seq``, or the file position for rows persisted before it
+        existed) keeps every entry at its own position so a keyset cursor
+        can resume between tied entries instead of skipping them.
+        """
+        record, index = item
+        ordinal = record.seq if record.seq is not None else index
         return (
-            item.resolved_at is None,
+            record.resolved_at is None,
-            item.resolved_at or datetime.max.replace(tzinfo=UTC),
+            record.resolved_at or datetime.max.replace(tzinfo=UTC),
-            item.occurrence_id,
+            record.occurrence_id,
+            float(ordinal),
         )
 
     def list_occurrences(
@@ -208,37 +236,61 @@ class FileScheduleStore:
         cursor: str | None = None,
         limit: int = 50,
     ) -> dict[str, object]:
-        """Return one history page ordered by ``(resolved_at, occurrence_id)``.
+        """Return one history page ordered by ``(resolved_at, occurrence_id, seq)``.
 
-        Keyset cursors are ``"<resolved_at_iso>|<occurrence_id>"`` (empty iso
+        Keyset cursors are ``"<resolved_at_iso>|<occurrence_id>|<seq>"``
-        for rows without a resolved instant); plain integer offsets remain
+        (empty iso for rows without a resolved instant). Two-part
-        accepted as a legacy fallback.
+        ``"<iso>|<id>"`` cursors issued before the entry ordinal existed
+        remain accepted and resume exactly as they did then (after the
+        whole tied group); plain integer offsets remain accepted as a
+        legacy fallback.
         """
         if limit < 1 or limit > 100:
             raise ValueError("limit must be between 1 and 100")
-        records = [
+        indexed = [
-            OccurrenceRecord.model_validate(item)
+            (OccurrenceRecord.model_validate(item), index)
-            for item in self._read_history_locked(schedule_id)
+            for index, item in enumerate(self._read_history_locked(schedule_id))
         ]
-        records.sort(key=self._sort_key)
+        indexed.sort(key=self._sort_key)
+        records = [record for record, _ in indexed]
+        keys = [self._sort_key(item) for item in indexed]
         total = len(records)
         start_index = 0
         if cursor is not None:
             if "|" in cursor:
-                iso_part, _, oid_part = cursor.partition("|")
+                # The occurrence id itself contains "|" separators, so split
-                key = (
+                # the trailing ordinal off the right: a new cursor ends in
+                # an integer seq, anything else is a pre-ordinal cursor
+                # (whose id tail is always an ISO instant, never an int).
+                head, _, seq_part = cursor.rpartition("|")
+                try:
+                    ordinal = float(int(seq_part))
+                except ValueError:
+                    # Pre-ordinal cursor: resume exactly as before, after
+                    # the whole tied group (no duplicates, no crash).
+                    iso_part, _, oid_part = cursor.partition("|")
+                    ordinal = float("inf")
+                else:
+                    iso_part, _, oid_part = head.partition("|")
+                key: tuple[bool, datetime, str, float] = (
                     True,
                     datetime.max.replace(tzinfo=UTC),
                     oid_part,
+                    ordinal,
                 )
                 if iso_part:
                     try:
-                        key = (False, datetime.fromisoformat(iso_part), oid_part)
+                        key = (
+                            False,
+                            datetime.fromisoformat(iso_part),
+                            oid_part,
+                            ordinal,
+                        )
                     except ValueError as exc:
                         raise ValueError("invalid keyset cursor") from exc
                 start_index = 0
-                for index, item in enumerate(records):
+                for index, item_key in enumerate(keys):
-                    if self._sort_key(item) > key:
+                    if item_key > key:
                         start_index = index
                         break
                 else:
@@ -248,15 +300,19 @@ class FileScheduleStore:
                     start_index = int(cursor)
                 except ValueError as exc:
                     raise ValueError(
-                        "cursor must be a keyset '<iso>|<id>' or integer offset"
+                        "cursor must be a keyset '<iso>|<id>|<seq>' or integer offset"
                     ) from exc
                 if start_index < 0:
                     raise ValueError("cursor must be a non-negative integer offset")
         page = records[start_index : start_index + limit]
+        page_keys = keys[start_index : start_index + limit]
         if start_index + limit < total and page:
             last = page[-1]
+            _, _, _, last_ordinal = page_keys[-1]
             iso = "" if last.resolved_at is None else last.resolved_at.isoformat()
-            next_cursor: str | None = f"{iso}|{last.occurrence_id}"
+            next_cursor: str | None = (
+                f"{iso}|{last.occurrence_id}|{int(last_ordinal)}"
+            )
         else:
             next_cursor = None
         return {

tests/scheduling/test_schedule_store.py

@@ -222,3 +222,92 @@ def test_deployment_revision_increments_on_save(tmp_path: Path) -> None:
         )
     )
     assert artifacts.get_deployment("dep-1").revision == 2
+
+
+def _tied_entry(
+    kind: str,
+    checkpoint_id: str | None,
+    created_minute: int,
+) -> OccurrenceRecord:
+    """One history entry for the shared 12:00 occurrence of schedule a."""
+    return OccurrenceRecord(
+        schedule_id="a",
+        occurrence_id="a|2026-09-08T12:00:00+00:00",
+        kind=kind,  # type: ignore[arg-type]
+        resolved_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC),
+        run_id="run-1",
+        revision=1,
+        checkpoint_id=checkpoint_id,
+        created_at=datetime(2026, 9, 8, 12, created_minute, tzinfo=UTC),
+    )
+
+
+def _traverse(store: FileScheduleStore, limit: int) -> list[dict]:
+    """Walk every page to the end, returning all rows in visit order."""
+    rows: list[dict] = []
+    cursor: str | None = None
+    while True:
+        page = store.list_occurrences("a", cursor=cursor, limit=limit)
+        rows.extend(page["occurrences"])  # type: ignore[arg-type]
+        cursor = page["next_cursor"]  # type: ignore[assignment]
+        if cursor is None:
+            assert page["total"] == len(rows)
+            return rows
+
+
+def test_history_pagination_visits_every_tied_entry_once(tmp_path: Path) -> None:
+    """Admission + repeated interruptions + completion share one cursor tie.
+
+    Every stored entry must appear exactly once across small-page
+    traversal (B4): the cursor carries the entry ordinal, not just the
+    shared ``(resolved_at, occurrence_id)`` tie.
+    """
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    store.append_history(_tied_entry("admitted", None, 0))
+    store.append_history(_tied_entry("interrupted", "run-1.000001", 5))
+    store.append_history(_tied_entry("interrupted", "run-1.000002", 9))
+    store.append_history(_tied_entry("completed", "run-1.000003", 14))
+
+    rows = _traverse(store, limit=1)
+    assert [(row["kind"], row["checkpoint_id"]) for row in rows] == [
+        ("admitted", None),
+        ("interrupted", "run-1.000001"),
+        ("interrupted", "run-1.000002"),
+        ("completed", "run-1.000003"),
+    ]
+    first = store.list_occurrences("a", limit=1)
+    assert first["next_cursor"] is not None
+    assert len(str(first["next_cursor"]).split("|")) >= 3
+
+
+def test_history_pagination_legacy_rows_keep_file_order(tmp_path: Path) -> None:
+    """Rows persisted before the entry ordinal order by file position.
+
+    Real persisted data without ``seq`` must still traverse exactly once;
+    pre-ordinal two-part cursors stay accepted and resume after the tied
+    group exactly as they did before (no duplicates, no crash).
+    """
+    import json
+
+    store = FileScheduleStore(tmp_path)
+    store.create_schedule(_schedule("a"))
+    history_path = tmp_path / "schedules" / "a" / "history.json"
+    history_path.parent.mkdir(parents=True, exist_ok=True)
+    legacy = [
+        _tied_entry("admitted", None, 0).model_dump(mode="json"),
+        _tied_entry("interrupted", "run-1.000001", 5).model_dump(mode="json"),
+    ]
+    for item in legacy:
+        del item["seq"]
+    history_path.write_text(json.dumps(legacy), encoding="utf-8")
+
+    rows = _traverse(store, limit=1)
+    assert [row["kind"] for row in rows] == ["admitted", "interrupted"]
+
+    legacy_cursor = "2026-09-08T12:00:00+00:00|a|2026-09-08T12:00:00+00:00"
+    resumed = store.list_occurrences("a", cursor=legacy_cursor, limit=1)
+    assert resumed["occurrences"] == []
+    assert resumed["next_cursor"] is None
+    with pytest.raises(ValueError):
+        store.list_occurrences("a", cursor="a|b|c|d", limit=1)

tests/wf_api/test_schedules.py

@@ -767,6 +767,74 @@ async def test_occurrences_pending_synthesis_first_page_only(tmp_path: Path) ->
     assert all(row["kind"] != "pending" for row in plain["occurrences"])
 
 
+async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) -> None:
+    """One occurrence's admission + stops each appear exactly once (B4).
+
+    Small-page traversal through the public API must not lose the tied
+    stopped entries sharing the occurrence's ``(resolved_at,
+    occurrence_id)``.
+    """
+    api, sched_store, _, _, _ = _harness(tmp_path / "tied")
+    await api.create_schedule(
+        schedule_id="s", deployment_id="dep.personal", trigger=_cron()
+    )
+    base = ts(2026, 9, 8, 12, 0)
+    recorder = FileScheduleHistoryRecorder(sched_store)
+    recorder.record(
+        HistoryEntry(
+            schedule_id="s",
+            kind="admitted",
+            resolved_at=base,
+            run_id="run-1",
+            revision=1,
+            reason="rev=1",
+            created_at=base,
+        )
+    )
+    recorder.record(
+        HistoryEntry(
+            schedule_id="s",
+            kind="interrupted",
+            resolved_at=base,
+            run_id="run-1",
+            revision=1,
+            reason="fresh-result",
+            checkpoint_id="run-1.000001",
+            created_at=base + timedelta(minutes=5),
+        )
+    )
+    recorder.record(
+        HistoryEntry(
+            schedule_id="s",
+            kind="completed",
+            resolved_at=base,
+            run_id="run-1",
+            revision=1,
+            reason="reconciled-on-recovery",
+            checkpoint_id="run-1.000002",
+            created_at=base + timedelta(minutes=9),
+        )
+    )
+
+    seen: list[tuple[str, str | None]] = []
+    cursor: str | None = None
+    while True:
+        page = await api.list_schedule_occurrences(
+            schedule_id="s", cursor=cursor, limit=1
+        )
+        assert page["total"] == 3
+        for row in page["occurrences"]:
+            seen.append((row["kind"], row["checkpoint_id"]))
+        cursor = page["next_cursor"]
+        if cursor is None:
+            break
+    assert seen == [
+        ("admitted", None),
+        ("interrupted", "run-1.000001"),
+        ("completed", "run-1.000002"),
+    ]
+
+
 async def test_inspect_admitted_run_without_fabrication(tmp_path: Path) -> None:
     api, sched_store, run_store, artifact_store, context = _harness(
         tmp_path / "admitted"

tests/wf_transport_rpc_http/test_schedules_rpc.py

@@ -265,6 +265,58 @@ async def test_rpc_schedule_occurrences_pagination_and_pending(tmp_path) -> None
         assert all(row["kind"] != "pending" for row in plain["occurrences"])
 
 
+async def test_rpc_schedule_occurrences_tied_entries_traverse_once(tmp_path) -> None:
+    """Tied admission/stopped entries each arrive exactly once over RPC (B4)."""
+    server = await _seed_server(tmp_path)
+    client, http_client = _client_for(server)
+    store = FileScheduleStore(tmp_path / "store")
+    base = datetime(2026, 9, 8, 12, 0, tzinfo=UTC)
+    async with http_client:
+        await client.create_schedule(
+            schedule_id="s",
+            deployment_id="dep.personal",
+            trigger=_cron(),
+        )
+        recorder = FileScheduleHistoryRecorder(store)
+        recorder.record(
+            HistoryEntry(
+                schedule_id="s",
+                kind="admitted",
+                resolved_at=base,
+                run_id="run-1",
+                revision=1,
+                reason="rev=1",
+                created_at=base,
+            )
+        )
+        recorder.record(
+            HistoryEntry(
+                schedule_id="s",
+                kind="interrupted",
+                resolved_at=base,
+                run_id="run-1",
+                revision=1,
+                reason="fresh-result",
+                checkpoint_id="run-1.000001",
+                created_at=base + timedelta(minutes=5),
+            )
+        )
+
+        seen: list[tuple[str, object]] = []
+        cursor: object = None
+        while True:
+            page = await client.list_schedule_occurrences(
+                schedule_id="s", cursor=cursor, limit=1  # type: ignore[arg-type]
+            )
+            assert page["total"] == 2
+            for row in page["occurrences"]:
+                seen.append((row["kind"], row["checkpoint_id"]))
+            cursor = page["next_cursor"]
+            if cursor is None:
+                break
+        assert seen == [("admitted", None), ("interrupted", "run-1.000001")]
+
+
 async def test_rpc_schedule_error_mapping(tmp_path) -> None:
     server = await _seed_server(tmp_path)
     client, http_client = _client_for(server)