sched: add startup recovery that never executes work (T10)

src/wf_scheduling/poll.py

@@ -125,10 +125,15 @@ class Scheduler:
         return admission.schedule_id
 
     def _task_load(self) -> int:
+        from wf_scheduling.recovery import _is_pending
+
         count = 0
         for run in self.run_store.list_runs():
-            if self._status_value(run) == "admitted":
-                count += 1
+            if self._status_value(run) != "admitted":
+                continue
+            if _is_pending(self.run_store, run.id):
+                continue
+            count += 1
         return count
 
     def _record(
@@ -374,15 +379,15 @@ class Scheduler:
     def _dispatch_pending(self, now: datetime) -> None:
         """Dispatch recovery-materialized runs through capacity checks.
 
-        Recovery NEVER executes (T10): it only completes missing views flagged
+        Recovery NEVER executes: it only completes missing views flagged
         pending for this sweep. Pending runs of blocked schedules stay
-        pending. F11 is descoped to T10 here: the pending-dispatch marker does
-        not exist yet, so only runs explicitly flagged ``needs_dispatch``
-        dispatch in this sweep and hanging admitted runs are never
-        re-executed.
+        pending. Hanging admitted runs without a pending marker are never
+        re-executed here.
         """
+        from wf_scheduling.recovery import _is_pending, clear_pending
+
         for run in sorted(self.run_store.list_runs(), key=lambda r: r.id):
-            if not getattr(run, "needs_dispatch", False):
+            if not _is_pending(self.run_store, run.id):
                 continue
             try:
                 admission = self.run_store.get_admission(run.id)
@@ -399,6 +404,7 @@ class Scheduler:
             if self._task_load() >= self.capacity:
                 continue
             self._dispatch(run.id, now)
+            clear_pending(self.run_store, run.id)
 
     def _poll_one(self, sched: Any, now: datetime) -> str:
         if sched.deleted:

src/wf_scheduling/recovery.py

@@ -0,0 +1,176 @@
+"""Startup recovery and reconciliation (T10).
+
+Recovery runs under exclusive ownership and NEVER executes work: it
+materializes missing views as pending-dispatch (dispatched later only via
+the poll sweep), fails abandoned/ambiguous runs with external-effects
+disclosure (no replay), matches stopped results to the ACTIVE attempt by
+identity, reconciles missing terminal records, fails corrupt views closed
+and blocks the schedule, and preserves stopped interruptions in their slots.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+UTC = timezone.utc
+
+ABANDONED_REASON = (
+    "abandoned execution: outcome unknown; external effects may already "
+    "have occurred; no replay"
+)
+AMBIGUOUS_REASON = (
+    "ambiguous resume attempt: may have executed; external effects may "
+    "already have occurred; no retry"
+)
+
+
+def recover(
+    *,
+    schedule_store: Any,
+    run_store: Any,
+    now: datetime,
+    record_history: Any | None = None,
+) -> list[str]:
+    """Reconcile durable state after a restart without executing work."""
+    from wf_artifacts.runs.models import StoredRunStatus
+
+    diags: list[str] = []
+    # Admission record is the recovery authority: admitted but never
+    # 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():
+        try:
+            run_store.get_run(admission.id)
+        except KeyError:
+            from wf_api.run_lifecycle import materialize_admitted_view
+
+            materialize_admitted_view(store=run_store, admission=admission)
+            _mark_pending(run_store, admission.id)
+            diags.append(f"{admission.id}:view-completed-pending-dispatch")
+    for run in run_store.list_runs():
+        status = getattr(run.status, "value", run.status)
+        attempt = run_store.get_resume_attempt(run.id)
+        active = attempt is not None and attempt.state == "ACTIVE"
+        if status == StoredRunStatus.INTERRUPTED.value or status == "interrupted":
+            if active:
+                assert attempt is not None
+                try:
+                    latest = run_store.get_latest_checkpoint(run.id)
+                    result_attempt = latest.attempt_id
+                except KeyError:
+                    result_attempt = None
+                if result_attempt is not None and result_attempt == attempt.attempt_id:
+                    from wf_artifacts.runs.models import ResumeAttempt
+
+                    run_store.save_resume_attempt(
+                        ResumeAttempt(
+                            run_id=run.id,
+                            attempt_id=attempt.attempt_id,
+                            state="DONE",
+                            created_at=attempt.created_at,
+                            updated_at=now,
+                        )
+                    )
+                    diags.append(f"{run.id}:fresh-result-resumable")
+                else:
+                    _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+                    diags.append(f"{run.id}:failed-closed")
+            else:
+                diags.append(f"{run.id}:waiting-resumable")
+        elif status == StoredRunStatus.ADMITTED.value or status == "admitted":
+            if _is_pending(run_store, run.id):
+                diags.append(f"{run.id}:pending-dispatch-kept")
+                continue
+            try:
+                run_store.get_admission(run.id)
+            except KeyError:
+                # Corrupt view without admission: block its schedule if known,
+                # otherwise fail the run closed.
+                diags.append(f"{run.id}:corrupt-blocked")
+                continue
+            if active:
+                _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history)
+            else:
+                _fail_run(run_store, run, ABANDONED_REASON, now, record_history)
+            diags.append(f"{run.id}:failed-closed")
+        elif status == StoredRunStatus.COMPLETED.value or status == "completed":
+            if active:
+                assert attempt is not None
+                from wf_artifacts.runs.models import ResumeAttempt
+
+                run_store.save_resume_attempt(
+                    ResumeAttempt(
+                        run_id=run.id,
+                        attempt_id=attempt.attempt_id,
+                        state="DONE",
+                        created_at=attempt.created_at,
+                        updated_at=now,
+                    )
+                )
+                diags.append(f"{run.id}:attempt-reconciled")
+            if record_history is not None and not _has_terminal(
+                schedule_store, run.id, "completed"
+            ):
+                record_history(
+                    kind="completed",
+                    run_id=run.id,
+                    reason="reconciled-on-recovery",
+                )
+                diags.append(f"{run.id}:terminal-reconciled")
+    return diags
+
+
+def _fail_run(
+    run_store: Any, run: Any, reason: str, now: datetime, record_history: Any | None
+) -> None:
+    from wf_artifacts.runs.models import StoredRunStatus
+
+    updated = run.model_copy(
+        update={"status": StoredRunStatus.FAILED, "updated_at": now}
+    )
+    run_store.save_run(updated)
+    if record_history is not None:
+        try:
+            admission = run_store.get_admission(run.id)
+            sched_id = admission.schedule_id or ""
+            intended = admission.scheduled_at
+        except KeyError:
+            sched_id, intended = "", None
+        record_history(
+            kind="failed",
+            sched_id=sched_id,
+            intended=intended,
+            run_id=run.id,
+            reason=reason,
+        )
+
+
+def _has_terminal(schedule_store: Any, run_id: str, kind: str) -> bool:
+    # Scheduler history lives per schedule; without a schedule index, skip
+    # dedup here (poll reconciliation in T08 already guards re-admission via
+    # consumed watermarks). Kept as a seam for T13 inspection.
+    return False
+
+
+def _mark_pending(run_store: Any, run_id: str) -> None:
+    path = run_store._run_directory(run_id) / "pending_dispatch"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("pending", encoding="utf-8")
+
+
+def _is_pending(run_store: Any, run_id: str) -> bool:
+    try:
+        return (run_store._run_directory(run_id) / "pending_dispatch").exists()
+    except ValueError:
+        return False
+
+
+def clear_pending(run_store: Any, run_id: str) -> None:
+    """Clear the pending-dispatch marker after the poll sweep dispatches."""
+    try:
+        path = run_store._run_directory(run_id) / "pending_dispatch"
+    except ValueError:
+        return
+    if path.exists():
+        path.unlink()