fix: reconcile admitted schedule candidates

src/wf_api/schedules.py

@@ -154,6 +154,25 @@ class WorkflowScheduleApi:
             raise KeyError("workflow artifact store is not configured")
         return self.context.artifact_store
 
+    def _candidate_is_admitted(
+        self, *, schedule_id: str, intended_at: datetime
+    ) -> bool:
+        """Return whether durable run admission owns this occurrence.
+
+        Candidate and admission files are written at adjacent lifecycle
+        boundaries. Suppressing the candidate while the admission is already
+        visible keeps an interleaved read truthful; startup recovery also
+        removes the stale candidate durably.
+        """
+        run_store = self.context.run_store
+        if run_store is None:
+            return False
+        return any(
+            admission.schedule_id == schedule_id
+            and admission.scheduled_at == intended_at
+            for admission in run_store.list_admissions()
+        )
+
     @staticmethod
     def _check_trigger(schedule: Schedule) -> None:
         """Reject triggers the poll loop could not build a source for."""
@@ -476,6 +495,11 @@ class WorkflowScheduleApi:
             raise ValueError("limit must be between 1 and 100")
         first_page = cursor in (None, "0")
         candidate = store.get_candidate(schedule_id)
+        if candidate is not None and self._candidate_is_admitted(
+            schedule_id=schedule_id,
+            intended_at=candidate.intended_at,
+        ):
+            candidate = None
         pending_candidate = candidate if first_page else None
         stored_cursor = None if cursor == _PENDING_PAGE_CURSOR else cursor
         stored_limit = limit - 1 if pending_candidate is not None else limit

src/wf_scheduling/recovery.py

@@ -88,6 +88,25 @@ CORRUPT_VIEW_REASON = (
 )
 
 
+def _clear_candidate_for_admission(schedule_store: Any, admission: Any) -> bool:
+    """Clear a candidate whose occurrence is already durably admitted.
+
+    Admission is the authoritative occurrence identity. Candidate clearing
+    follows it as a separate file write, so recovery must remove a candidate
+    left behind by that torn boundary. A newer candidate for another instant
+    is left untouched.
+    """
+    schedule_id = getattr(admission, "schedule_id", None)
+    scheduled_at = getattr(admission, "scheduled_at", None)
+    if schedule_id is None or scheduled_at is None:
+        return False
+    candidate = schedule_store.get_candidate(schedule_id)
+    if candidate is None or candidate.intended_at != scheduled_at:
+        return False
+    schedule_store.save_candidate(None, schedule_id=schedule_id)
+    return True
+
+
 def recover(
     *,
     schedule_store: Any,
@@ -127,12 +146,18 @@ def recover(
         # Scoped live repair: never touch admissions, views, markers, or
         # histories belonging to any other run. The live poll sweep owns
         # pending dispatch; this path only reconciles the one broken run.
+        try:
+            admission = run_store.get_admission(only_run_id)
+        except KeyError:
+            admission = None
+        if admission is not None and _clear_candidate_for_admission(
+            schedule_store, admission
+        ):
+            diags.append(f"{only_run_id}:stale-candidate-cleared")
         try:
             runs = [run_store.get_run(only_run_id)]
         except KeyError:
-            try:
+            if admission is None:
-                admission = run_store.get_admission(only_run_id)
-            except KeyError:
                 return [f"{only_run_id}:run-unknown"]
             from wf_api.run_lifecycle import materialize_admitted_view
 
@@ -144,6 +169,8 @@ def recover(
         # materialized views are completed here and flagged pending for the
         # capacity-checked poll sweep (exactly once, occurrence already consumed).
         for admission in run_store.list_admissions():
+            if _clear_candidate_for_admission(schedule_store, admission):
+                diags.append(f"{admission.id}:stale-candidate-cleared")
             try:
                 run_store.get_run(admission.id)
             except KeyError:

tests/wf_api/test_schedules.py

@@ -38,6 +38,8 @@ from wf_core import InterruptRequest, RunState, RunStatus
 from wf_platform import CapabilitySource
 from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry
 from wf_scheduling.models import PendingCandidate
+from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.recovery import recover
 from wf_scheduling.store import (
     FileScheduleStore,
     ScheduleExistsError,
@@ -787,6 +789,66 @@ 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_admitted_occurrence_replaces_retained_candidate_after_recovery(
+    tmp_path: Path,
+) -> None:
+    """A torn candidate clear cannot expose an admitted occurrence as pending."""
+    root = tmp_path / "admitted_candidate"
+    api, sched_store, run_store, artifact_store, _ = _harness(root)
+    await api.create_schedule(
+        schedule_id="s", deployment_id="dep.personal", trigger=_cron()
+    )
+    intended = ts(2026, 9, 8, 13, 0)
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_store.allocate_run_id(),
+        environment=_env(artifact_store),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id="s",
+        schedule_revision=1,
+    )
+    FileScheduleHistoryRecorder(sched_store).record(
+        HistoryEntry(
+            schedule_id="s",
+            kind="admitted",
+            resolved_at=intended,
+            run_id=admission.id,
+            revision=1,
+            reason="rev=1",
+            created_at=intended,
+        )
+    )
+    # Fault injection: admission/history are durable, but candidate clearing
+    # was lost at the following persistence boundary.
+    sched_store.save_candidate(
+        PendingCandidate(schedule_id="s", intended_at=intended, revision=1),
+        schedule_id="s",
+    )
+
+    before_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1)
+    assert before_recovery["total"] == 1
+    assert before_recovery["occurrences"][0]["kind"] == "admitted"
+
+    ownership = SchedulerOwnership(root, owner="test").acquire()
+    try:
+        recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=intended,
+            ownership=ownership,
+        )
+    finally:
+        ownership.release()
+
+    assert sched_store.get_candidate("s") is None
+    after_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1)
+    assert after_recovery["total"] == 1
+    assert after_recovery["occurrences"][0]["kind"] == "admitted"
+    assert after_recovery["occurrences"][0]["run_id"] == admission.id
+
+
 async def test_occurrences_pending_limit_one_traverses_every_row(
     tmp_path: Path,
 ) -> None: