sched: checkpoint-first reconcile across torn checkpoint/summary writes (F2)

src/wf_scheduling/recovery.py

@@ -20,9 +20,11 @@ clear → executor → stopped persist → executing clear → history):
 - any admitted run with the executing mark: the executor may already have
   produced external effects; failed (ACTIVE attempt: ambiguous, else
   abandoned) with external-effects disclosure, never redispatched.
-- admitted + view with neither mark and no ACTIVE attempt: legacy or
+- admitted + view with neither mark and no ACTIVE attempt: a durable
-  manually cleared state whose outcome is unprovable; failed closed
+  stopped checkpoint reconciles the summary first (checkpoint-first rule);
-  without replay (current code never produces this shape on crash).
+  only when no checkpoint exists is the outcome unprovable, and the run is
+  failed closed without replay (current code never produces this shape on
+  crash).
 - stopped summary + executing/pending marks: completion-window leftovers
   (stopped result persisted, marker clearing lost); markers are cleared,
   the stopped status stands, nothing is re-executed. Checkpoint-vs-summary
@@ -94,6 +96,21 @@ def recover(
         status = getattr(run.status, "value", run.status)
         attempt = run_store.get_resume_attempt(run.id)
         active = attempt is not None and attempt.state == "ACTIVE"
+        # Checkpoint-first reconcile across the torn
+        # save_checkpoint/save_run boundary: a durable stopped checkpoint
+        # wins over a stale summary (status, checkpoint pointer, and
+        # readiness are rewritten together from the checkpoint's own
+        # persisted fields; nothing is fabricated). Attempt matching below
+        # still distinguishes fresh results from stale ones.
+        try:
+            changed, status = _reconcile_summary_from_checkpoint(run_store, run, now)
+        except (ValueError, OSError) as exc:
+            _fail_run(run_store, run, f"corrupt checkpoint: {exc}", now, history)
+            diags.append(f"{run.id}:failed-closed")
+            continue
+        if changed:
+            run = run_store.get_run(run.id)
+            diags.append(f"{run.id}:summary-reconciled")
         if status in (
             StoredRunStatus.INTERRUPTED.value,
             StoredRunStatus.COMPLETED.value,
@@ -217,6 +234,49 @@ def recover(
     return diags
 
 
+def _reconcile_summary_from_checkpoint(
+    run_store: Any, run: Any, now: datetime
+) -> tuple[bool, str]:
+    """Rewrite a stale summary from its durable stopped checkpoint.
+
+    Returns ``(changed, status)``. Only the checkpoint's own persisted
+    fields (status, checkpoint id, readiness derived exactly as
+    ``persist_stopped_run`` derives it) are copied; creation time,
+    environment, and diagnostics are preserved. A run without checkpoints
+    is untouched (an admitted summary with no checkpoint is genuinely
+    undispatched-or-unknown, handled by the marker rules).
+    """
+    from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus
+
+    try:
+        latest = run_store.get_latest_checkpoint(run.id)
+    except KeyError:
+        return False, getattr(run.status, "value", run.status)
+    expected = StoredRunStatus(latest.reason.value)
+    readiness = (
+        ResumeReadiness.READY
+        if expected is StoredRunStatus.INTERRUPTED
+        else ResumeReadiness.NOT_APPLICABLE
+    )
+    if (
+        run.status == expected
+        and run.latest_checkpoint_id == latest.id
+        and run.resume_readiness == readiness
+    ):
+        return False, getattr(run.status, "value", run.status)
+    run_store.save_run(
+        run.model_copy(
+            update={
+                "status": expected,
+                "latest_checkpoint_id": latest.id,
+                "resume_readiness": readiness,
+                "updated_at": now,
+            }
+        )
+    )
+    return True, expected.value
+
+
 def _attribution(
     run_store: Any, run_id: str
 ) -> tuple[str | None, datetime | None, int | None]:

tests/scheduling/controlled.py

@@ -88,3 +88,24 @@ class ScriptedDispatcher:
     def finish(self, admission: RunAdmission, outcome: str) -> RunState:
         """Build the stopped state that settles a hanging run."""
         return stopped_state(admission, outcome)
+
+
+class WorkflowDispatcher:
+    """Execution double driving a real workflow to its first stopped state.
+
+    Dispatch runs the workflow through the genuine ``wf_core`` runtime and
+    returns the resulting stopped state, so scheduler tests exercise real
+    execution (including durable interruptions) without enabling any server.
+    """
+
+    def __init__(self, workflow: Any, registry: dict[str, Any] | None = None) -> None:
+        self.workflow = workflow
+        self.registry = registry or {}
+
+    def dispatch(self, *, admission: RunAdmission, now: datetime) -> DispatchResult:
+        from wf_core import execute_workflow
+
+        state = execute_workflow(
+            self.workflow, dict(admission.resolved_input), self.registry
+        )
+        return Stopped(result=state)

tests/scheduling/test_torn_reconcile.py

@@ -0,0 +1,454 @@
+"""Torn checkpoint/summary reconciliation: the checkpoint wins (R4/F2).
+
+``persist_stopped_run`` writes the checkpoint before the summary; a crash
+in between leaves a stale summary. Recovery inspects and validates the
+durable stopped checkpoint before deciding from the summary, reconciling
+status, checkpoint pointer, readiness, and attempt state together — for
+first executions and resumes, matching and mismatching attempts, without
+fabricating execution output.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, cast
+
+from tests.artifacts.test_run_store import artifact as _artifact
+from tests.artifacts.test_run_store import deployment as _deployment
+from tests.scheduling.controlled import (
+    DictDeployments,
+    WorkflowDispatcher,
+    fixture_environment,
+)
+from wf_api.run_lifecycle import (
+    load_stored_run,
+    materialize_admitted_view,
+    persist_admission,
+    persist_stopped_run,
+    restore_interrupted_run,
+)
+from wf_artifacts import PinnedRunEnvironment
+from wf_artifacts.runs.models import ResumeAttempt
+from wf_artifacts.runs.store import FileRunStore
+from wf_core import (
+    END,
+    Edge,
+    InterruptNode,
+    RunState,
+    RunStatus,
+    SchemaRef,
+    StateSchema,
+    Workflow,
+    load_run_state,
+    resume_workflow,
+)
+from wf_scheduling import recovery as sched_recovery
+from wf_scheduling.models import Schedule
+from wf_scheduling.ownership import SchedulerOwnership
+from wf_scheduling.poll import Scheduler
+from wf_scheduling.prepare import SchedulePreparer
+from wf_scheduling.store import FileScheduleStore
+
+
+def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime:
+    return datetime(y, mo, d, h, mi, tzinfo=UTC)
+
+
+def _sched_model(sid: str, **kw: Any) -> Schedule:
+    now = ts(2026, 9, 8, 12, 0)
+    base: dict[str, Any] = {
+        "id": sid,
+        "deployment_id": "dep-1",
+        "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"},
+        "input_bindings": [],
+        "created_at": now.isoformat(),
+        "updated_at": now.isoformat(),
+    }
+    base.update(kw)
+    return Schedule.model_validate(base)
+
+
+def _env() -> PinnedRunEnvironment:
+    return PinnedRunEnvironment(
+        deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[]
+    )
+
+
+class FailSaveRunOnce(FileRunStore):
+    """Real store that drops exactly one summary write (torn boundary)."""
+
+    def __init__(self, root: Path) -> None:
+        super().__init__(root)
+        self.armed = False
+
+    def save_run(self, run: Any) -> None:
+        if self.armed:
+            self.armed = False
+            raise OSError("injected save_run failure")
+        super().save_run(run)
+
+
+def _recover(
+    sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime
+) -> list[str]:
+    ownership = SchedulerOwnership(sched_store.root, owner="test").acquire()
+    try:
+        return sched_recovery.recover(
+            schedule_store=sched_store,
+            run_store=run_store,
+            now=now,
+            ownership=ownership,
+        )
+    finally:
+        ownership.release()
+
+
+def _admit_view(
+    run_store: FileRunStore,
+    intended: datetime,
+    *,
+    schedule_id: str = "a",
+) -> str:
+    run_id = run_store.allocate_run_id()
+    admission = persist_admission(
+        store=run_store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=intended,
+        schedule_id=schedule_id,
+        schedule_revision=1,
+    )
+    materialize_admitted_view(store=run_store, admission=admission)
+    return run_id
+
+
+def _torn_persist(
+    run_store: FailSaveRunOnce,
+    run_id: str,
+    status: RunStatus,
+    *,
+    attempt_id: int | None = None,
+) -> None:
+    record = run_store.get_run(run_id)
+    run_store.armed = True
+    try:
+        persist_stopped_run(
+            store=run_store,
+            environment=record.environment,
+            run=RunState(
+                workflow_name="sched", status=status, workflow_input={}, state={}
+            ),
+            run_id=run_id,
+            attempt_id=attempt_id,
+        )
+        raise AssertionError("save_run failure must propagate")
+    except OSError as exc:
+        assert "injected save_run failure" in str(exc)
+
+
+def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, Any]]:
+    page = store.list_occurrences(sid, limit=100)
+    return [
+        r for r in cast(list[dict[str, Any]], page["occurrences"]) if r["kind"] == kind
+    ]
+
+
+def _double_interrupt_workflow() -> Workflow:
+    schema = SchemaRef(type="object", properties={})
+    return Workflow(
+        name="ask_twice",
+        input_schema=schema,
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=schema,
+        outcomes=["ok"],
+        start="ask1",
+        node_defs=[],
+        nodes=[
+            InterruptNode(id="ask1", type="interrupt", kind="approval"),
+            InterruptNode(id="ask2", type="interrupt", kind="approval"),
+        ],
+        edges=[
+            Edge.model_validate({"from": "ask1", "outcome": "submitted", "to": "ask2"}),
+            Edge.model_validate({"from": "ask2", "outcome": "submitted", "to": END}),
+        ],
+    )
+
+
+def test_torn_first_interrupt_reconciles_not_abandons(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    _torn_persist(run_store, run_id, RunStatus.INTERRUPTED)
+    assert run_store.get_run(run_id).status.value == "admitted"
+    diags = _recover(sched_store, run_store, intended)
+    assert any("summary-reconciled" in d for d in diags)
+    assert any("waiting-resumable" in d for d in diags)
+    assert not any("failed-closed" in d for d in diags)
+    record = run_store.get_run(run_id)
+    assert record.status.value == "interrupted"
+    assert record.latest_checkpoint_id == f"{run_id}.000001"
+    assert record.resume_readiness.value == "ready"
+    assert len(_entries(sched_store, "a", "interrupted")) == 1
+    # Inspection and resume eligibility agree with the reconciled result.
+    _, state = load_stored_run(run_store, run_id)
+    assert state.status is RunStatus.INTERRUPTED
+    settled, resumed = restore_interrupted_run(run_store, run_id)
+    assert settled.resume_readiness.value == "ready"
+    assert resumed.status is RunStatus.INTERRUPTED
+
+
+def test_torn_first_completed_reconciles(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    _torn_persist(run_store, run_id, RunStatus.COMPLETED)
+    diags = _recover(sched_store, run_store, intended)
+    record = run_store.get_run(run_id)
+    assert record.status.value == "completed"
+    assert record.latest_checkpoint_id == f"{run_id}.000001"
+    assert record.resume_readiness.value == "not_applicable"
+    assert any("summary-reconciled" in d for d in diags)
+    assert len(_entries(sched_store, "a", "completed")) == 1
+
+
+def test_torn_first_failed_reconciles(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    _torn_persist(run_store, run_id, RunStatus.FAILED)
+    diags = _recover(sched_store, run_store, intended)
+    record = run_store.get_run(run_id)
+    assert record.status.value == "failed"
+    assert record.latest_checkpoint_id == f"{run_id}.000001"
+    assert any("summary-reconciled" in d for d in diags)
+    assert len(_entries(sched_store, "a", "failed")) == 1
+
+
+def test_torn_resume_completed_matches_active_attempt(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    record = run_store.get_run(run_id)
+    persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(
+            workflow_name="sched",
+            status=RunStatus.INTERRUPTED,
+            workflow_input={},
+            state={},
+        ),
+        run_id=run_id,
+    )
+    attempt_id = run_store.allocate_resume_attempt_id()
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=attempt_id,
+            state="ACTIVE",
+            created_at=intended,
+            updated_at=intended,
+        )
+    )
+    _torn_persist(run_store, run_id, RunStatus.COMPLETED, attempt_id=attempt_id)
+    diags = _recover(sched_store, run_store, intended)
+    # The fresh COMPLETED checkpoint wins over the stale INTERRUPTED summary:
+    # reconciled as completed, never as fresh/resumable.
+    record = run_store.get_run(run_id)
+    assert record.status.value == "completed"
+    assert record.latest_checkpoint_id == f"{run_id}.000002"
+    assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+    assert any("summary-reconciled" in d for d in diags)
+    assert not any("fresh-result-resumable" in d for d in diags)
+    assert len(_entries(sched_store, "a", "completed")) == 1
+
+
+def test_torn_resume_interrupted_matches_active_attempt(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    record = run_store.get_run(run_id)
+    persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(
+            workflow_name="sched",
+            status=RunStatus.INTERRUPTED,
+            workflow_input={},
+            state={},
+        ),
+        run_id=run_id,
+    )
+    attempt_id = run_store.allocate_resume_attempt_id()
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=attempt_id,
+            state="ACTIVE",
+            created_at=intended,
+            updated_at=intended,
+        )
+    )
+    _torn_persist(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id)
+    diags = _recover(sched_store, run_store, intended)
+    record = run_store.get_run(run_id)
+    assert record.status.value == "interrupted"
+    assert record.latest_checkpoint_id == f"{run_id}.000002"
+    assert record.resume_readiness.value == "ready"
+    assert run_store.get_resume_attempt(run_id).state == "DONE"  # type: ignore[union-attr]
+    assert any("fresh-result-resumable" in d for d in diags)
+    assert len(_entries(sched_store, "a", "interrupted")) == 1
+
+
+def test_stale_checkpoint_never_clears_active_attempt(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    record = run_store.get_run(run_id)
+    persist_stopped_run(
+        store=run_store,
+        environment=record.environment,
+        run=RunState(
+            workflow_name="sched",
+            status=RunStatus.INTERRUPTED,
+            workflow_input={},
+            state={},
+        ),
+        run_id=run_id,
+        attempt_id=3,
+    )
+    run_store.save_resume_attempt(
+        ResumeAttempt(
+            run_id=run_id,
+            attempt_id=5,
+            state="ACTIVE",
+            created_at=intended,
+            updated_at=intended,
+        )
+    )
+    diags = _recover(sched_store, run_store, intended)
+    assert any("failed-closed" in d for d in diags)
+    assert run_store.get_run(run_id).status.value == "failed"
+    assert run_store.get_resume_attempt(run_id).state == "ACTIVE"  # type: ignore[union-attr]
+
+
+def test_double_recovery_after_reconcile_is_stable(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    _torn_persist(run_store, run_id, RunStatus.INTERRUPTED)
+    first = _recover(sched_store, run_store, intended)
+    assert any("summary-reconciled" in d for d in first)
+    before = run_store.get_run(run_id)
+    second = _recover(sched_store, run_store, intended + timedelta(minutes=1))
+    after = run_store.get_run(run_id)
+    assert not any("summary-reconciled" in d for d in second)
+    assert not any("terminal-reconciled" in d for d in second)
+    assert after == before
+    assert len(_entries(sched_store, "a", "interrupted")) == 1
+
+
+def test_real_workflow_interrupt_resume_interrupt_with_torn_resume(
+    tmp_path: Path,
+) -> None:
+    workflow = _double_interrupt_workflow()
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FailSaveRunOnce(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire()
+    try:
+        sched = Scheduler(
+            schedule_store=sched_store,
+            run_store=run_store,
+            sources={},
+            capacity=4,
+            preparer=SchedulePreparer(
+                DictDeployments({"dep-1": {"rev": 1, "required": []}}),
+                fixture_environment,
+            ),
+            dispatcher=WorkflowDispatcher(workflow, {}),
+            ownership=ownership,
+        )
+        sched_store.save_consumed("a", intended - timedelta(hours=1))
+        from wf_scheduling.calendar import OneShotSource
+
+        sched.sources["a"] = OneShotSource(intended)
+        result = sched.poll(intended)
+        run_id = next(iter(run_store.list_runs())).id
+        assert result == {"a": f"admit:{run_id}"}
+        assert run_store.get_run(run_id).status.value == "interrupted"
+        ownership.release()
+        # Resume through the real runtime to a second durable interruption,
+        # then tear the summary write.
+        fresh = FileRunStore(tmp_path / "runs")
+        checkpoint = fresh.get_latest_checkpoint(run_id)
+        st1 = load_run_state(checkpoint.state.model_dump(mode="json"))
+        st2 = resume_workflow(workflow, st1, {}, resume_payload={})
+        assert st2.status is RunStatus.INTERRUPTED
+        attempt_id = fresh.allocate_resume_attempt_id()
+        fresh.save_resume_attempt(
+            ResumeAttempt(
+                run_id=run_id,
+                attempt_id=attempt_id,
+                state="ACTIVE",
+                created_at=intended,
+                updated_at=intended,
+            )
+        )
+        armed = FailSaveRunOnce(tmp_path / "runs")
+        armed.armed = True
+        record = armed.get_run(run_id)
+        try:
+            persist_stopped_run(
+                store=armed,
+                environment=record.environment,
+                run=st2,
+                run_id=run_id,
+                attempt_id=attempt_id,
+            )
+            raise AssertionError("torn resume persist must raise")
+        except OSError as exc:
+            assert "injected save_run failure" in str(exc)
+        diags = _recover(sched_store, FileRunStore(tmp_path / "runs"), intended)
+        assert any("summary-reconciled" in d for d in diags)
+        assert any("fresh-result-resumable" in d for d in diags)
+        settled = FileRunStore(tmp_path / "runs").get_run(run_id)
+        assert settled.status.value == "interrupted"
+        assert settled.latest_checkpoint_id == f"{run_id}.000002"
+        assert settled.resume_readiness.value == "ready"
+        _, state = load_stored_run(FileRunStore(tmp_path / "runs"), run_id)
+        assert state.status is RunStatus.INTERRUPTED
+    finally:
+        ownership.release()
+
+
+def test_corrupt_checkpoint_fails_closed(tmp_path: Path) -> None:
+    sched_store = FileScheduleStore(tmp_path / "sched")
+    run_store = FileRunStore(tmp_path / "runs")
+    sched_store.create_schedule(_sched_model("a"))
+    intended = ts(2026, 9, 8, 12, 0)
+    run_id = _admit_view(run_store, intended)
+    checkpoint_dir = tmp_path / "runs" / run_id / "checkpoints"
+    checkpoint_dir.mkdir(parents=True, exist_ok=True)
+    (checkpoint_dir / "000001.json").write_text("not json", encoding="utf-8")
+    diags = _recover(sched_store, run_store, intended)
+    assert any("failed-closed" in d for d in diags)
+    assert run_store.get_run(run_id).status.value == "failed"