sched: address R2 fail-closed and fault-proof findings

src/wf_api/run_lifecycle.py

@@ -28,6 +28,7 @@ from wf_core import (
     load_run_state,
     load_run_state_with_upgrade,
 )
+from wf_core.models.json_values import validate_strict_json_value
 
 
 def create_pinned_environment(
@@ -62,8 +63,6 @@ def persist_admission(
     single authority for the run identity and frozen invocation. A failed
     durable admission must never dispatch.
     """
-    from wf_core.models.json_values import validate_strict_json_value
-
     frozen = validate_strict_json_value(dict(resolved_input))
     if not isinstance(frozen, dict):
         raise ValueError("resolved workflow input must be a JSON object")
@@ -91,6 +90,8 @@ def materialize_admitted_view(
 
     The view carries no checkpoint, trace, output, or step counts: the outcome
     is unknown until dispatch completes and persists a stopped checkpoint.
+    A view without a matching admission is corrupt and fails closed instead
+    of being silently returned (recovery authority is the admission record).
     """
     now = datetime.now(UTC)
     try:
@@ -98,6 +99,11 @@ def materialize_admitted_view(
     except KeyError:
         existing = None
     if existing is not None:
+        stored = store.get_admission(admission.id)
+        if stored.id != existing.id or stored.environment != existing.environment:
+            raise ValueError(
+                f"run view {admission.id!r} contradicts its admission record"
+            )
         return existing
     record = WorkflowRunRecord(
         id=admission.id,
@@ -116,12 +122,18 @@ def recover_admission_view(*, store: RunStore, run_id: str) -> WorkflowRunRecord
     """Reconcile a missing run view from its admission record.
 
     Recovery never executes work: it only completes the missing view so a
-    later poll can dispatch the captured invocation exactly once.
+    later poll can dispatch the captured invocation exactly once. A run view
+    without an admission record is corrupt and fails closed.
     """
     try:
-        return store.get_run(run_id)
+        existing = store.get_run(run_id)
     except KeyError:
-        pass
+        existing = None
+    if existing is not None:
+        # Fail closed on a corrupt view-without-admission (F5): do not
+        # silently return it to clear overlap.
+        store.get_admission(run_id)
+        return existing
     admission = store.get_admission(run_id)
     return materialize_admitted_view(store=store, admission=admission)
 
@@ -143,6 +155,9 @@ def persist_stopped_run(
             f"cannot persist active workflow run with status {run.status!s}"
         )
 
+    # New admissions use store-backed run-###### ids (allocate_run_id); the
+    # run_ hex fallback only serves pre-admission stopped runs and legacy
+    # tests that persist without an admission record.
     key = run_id or f"run_{uuid4().hex}"
     now = datetime.now(UTC)
     sequence = 1

src/wf_api/runs.py

@@ -100,10 +100,13 @@ class WorkflowRunApi:
                 max_steps=limits.max_steps,
             )
 
-        # Durable admission ordering: recheck -> allocate/freeze -> persist
-        # admission -> materialize view -> dispatch captured -> persist
-        # stopped -> reconcile. A failed durable admission never dispatches,
-        # and dispatch never re-resolves the deployment.
+        # Durable admission ordering for manual runs: deployment recheck ->
+        # allocate/freeze -> persist admission -> materialize view -> dispatch
+        # captured -> persist stopped. A failed durable admission never
+        # dispatches, and dispatch never re-resolves the deployment.
+        # TODO(T11): hold the single-owner admission lock around this sequence
+        # once scheduler ownership lands; manual recheck here is only
+        # deployment validation (no schedule/capacity/overlap yet).
         store = self._run_store()
         run_id = store.allocate_run_id()
         environment = create_pinned_environment(
@@ -119,6 +122,9 @@ class WorkflowRunApi:
             max_steps=limits.max_steps,
         )
         materialize_admitted_view(store=store, admission=admission)
+        # TODO(T10): record a dispatch mark between materialize and execute so
+        # crash-after-dispatch (abandoned, failed without replay) is
+        # distinguishable from pending-dispatch (safe to dispatch later).
         plan = raw_plan_from_artifact(admission.environment.root_artifact)
         captured_tree = saved_subgraph_tree_from_snapshots(
             admission.environment.child_artifacts
@@ -282,7 +288,12 @@ class WorkflowRunApi:
         }
 
     async def inspect_run(self, *, run_id: str) -> RunResult:
-        """Return one durable stopped-run summary without debug trace entries."""
+        """Return one durable stopped-run summary without debug trace entries.
+
+        Admitted runs with no stopped checkpoint fail closed here (no
+        fabricated trace/output); checkpoint-free inspection arrives with
+        the scheduling administration surface (T13).
+        """
         record, run = load_stored_run(self._run_store(), run_id)
         environment = record.environment
         return _run_payload(

src/wf_artifacts/runs/store.py

@@ -42,7 +42,7 @@ class RunStore:
 
 
 class FileRunStore(RunStore):
-    """JSON file-backed stopped-run store for local development and tests.
+    """JSON file-backed admitted- and stopped-run store for local dev/tests.
 
     The internal lock protects individual writes inside one process only.
     `WorkflowRunApi.resume_run()` owns the same-process read/execute/write
@@ -125,11 +125,15 @@ class FileRunStore(RunStore):
             seq = 0
             if seq_path.exists():
                 try:
-                    seq = int(
-                        json.loads(seq_path.read_text(encoding="utf-8")).get("seq", 0)
-                    )
-                except ValueError, AttributeError:
-                    seq = 0
+                    raw = json.loads(seq_path.read_text(encoding="utf-8"))
+                    seq_value = raw.get("seq", 0) if isinstance(raw, dict) else None
+                    if not isinstance(seq_value, int) or seq_value < 0:
+                        raise ValueError(
+                            f"corrupt run-id sequence at {seq_path}: {raw!r}"
+                        )
+                    seq = seq_value
+                except (ValueError, AttributeError, TypeError) as exc:
+                    raise ValueError(f"corrupt run-id sequence at {seq_path}") from exc
             seq += 1
             self._write_json(seq_path, {"seq": seq})
             return f"run-{seq:06d}"

tests/artifacts/test_run_admission.py

@@ -11,6 +11,7 @@ from wf_api.run_lifecycle import (
     load_stored_run,
     materialize_admitted_view,
     persist_admission,
+    recover_admission_view,
 )
 from wf_artifacts import (
     PinnedRunEnvironment,
@@ -102,8 +103,6 @@ def test_run_ids_are_store_backed_across_restart(tmp_path: Path) -> None:
 
 
 def test_fault_before_admission_persists_nothing(tmp_path: Path, monkeypatch) -> None:
-    from wf_artifacts.runs import store as store_module
-
     store = FileRunStore(tmp_path)
     run_id = store.allocate_run_id()
 
@@ -122,13 +121,13 @@ def test_fault_before_admission_persists_nothing(tmp_path: Path, monkeypatch) ->
             schedule_id=None,
             schedule_revision=None,
         )
-    assert store_module.FileRunStore(tmp_path).list_admissions() == []
+    assert FileRunStore(tmp_path).list_admissions() == []
     with pytest.raises(KeyError):
         store.get_admission(run_id)
 
 
 def test_fault_between_admission_and_view_recovers_without_dispatch(
-    tmp_path: Path, monkeypatch
+    tmp_path: Path,
 ) -> None:
     store = FileRunStore(tmp_path)
     run_id = store.allocate_run_id()
@@ -147,8 +146,6 @@ def test_fault_between_admission_and_view_recovers_without_dispatch(
     with pytest.raises(KeyError):
         store.get_run(run_id)
     # Recovery materializes the missing view but never executes work.
-    from wf_api.run_lifecycle import recover_admission_view
-
     record = recover_admission_view(store=store, run_id=run_id)
     assert record.status is StoredRunStatus.ADMITTED
     assert store.get_run(run_id).id == run_id
@@ -179,3 +176,46 @@ def test_admitted_run_has_no_fabricated_checkpoint_or_output(
     # fabricate trace/output/step counts.
     with pytest.raises(ValueError, match="admitted"):
         load_stored_run(store, run_id)
+
+
+def test_corrupt_run_id_sequence_fails_closed(tmp_path: Path) -> None:
+    FileRunStore(tmp_path)
+    seq_path = tmp_path / "runs" / "_run_id_seq.json"
+    seq_path.write_text('{"seq": "not-an-int"}', encoding="utf-8")
+    with pytest.raises(ValueError, match="corrupt run-id sequence"):
+        FileRunStore(tmp_path).allocate_run_id()
+
+
+def test_view_without_admission_fails_closed(tmp_path: Path) -> None:
+    store = FileRunStore(tmp_path)
+    run_id = store.allocate_run_id()
+    admission = persist_admission(
+        store=store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={},
+        max_steps=None,
+        scheduled_at=None,
+        schedule_id=None,
+        schedule_revision=None,
+    )
+    materialize_admitted_view(store=store, admission=admission)
+    (tmp_path / "runs" / run_id / "admission.json").unlink()
+    with pytest.raises(KeyError, match="unknown run admission"):
+        recover_admission_view(store=store, run_id=run_id)
+
+
+def test_non_json_resolved_input_rejected_before_dispatch(tmp_path: Path) -> None:
+    store = FileRunStore(tmp_path)
+    run_id = store.allocate_run_id()
+    with pytest.raises(ValueError, match="must be finite"):
+        persist_admission(
+            store=store,
+            run_id=run_id,
+            environment=_env(),
+            resolved_input={"x": float("inf")},  # type: ignore[dict-item]
+            max_steps=None,
+            scheduled_at=None,
+            schedule_id=None,
+            schedule_revision=None,
+        )

tests/wf_api/test_run_admission_path.py

@@ -65,15 +65,26 @@ def test_manual_run_persists_admission_before_dispatch(tmp_path: Path) -> None:
 
 
 def test_fault_before_admission_never_dispatches(tmp_path: Path, monkeypatch) -> None:
+    import wf_api.runs as runs_module
+
     api, store = _api_with_echo(tmp_path / "fault-before")
+    real_write = store._write_json
+    dispatched: list[str] = []
+    real_plan = runs_module.raw_plan_from_artifact
 
-    def _fail_save(admission) -> None:  # type: ignore[no-untyped-def]
-        raise OSError("injected admission failure")
+    def _fail_on_admission(path: Path, payload: object) -> None:
+        if path.name == "admission.json":
+            raise OSError("injected admission failure")
+        real_write(path, payload)
 
-    # Admission persist precedes dispatch in run_deployment ordering: a failure
-    # here must propagate before any run view exists, so no dispatch could
-    # have produced a stopped checkpoint.
-    monkeypatch.setattr(store, "save_admission", _fail_save)
+    def _spy_plan(artifact):  # type: ignore[no-untyped-def]
+        dispatched.append("dispatch")
+        return real_plan(artifact)
+
+    # Real serialization fault at the admission file: nothing after it
+    # (plan building, dispatch, stopped persist) may run.
+    monkeypatch.setattr(store, "_write_json", _fail_on_admission)
+    monkeypatch.setattr(runs_module, "raw_plan_from_artifact", _spy_plan)
     try:
         asyncio.run(
             api.run_deployment(
@@ -84,8 +95,10 @@ def test_fault_before_admission_never_dispatches(tmp_path: Path, monkeypatch) ->
         raise AssertionError("fault must propagate")
     except OSError:
         pass
-    assert store.list_admissions() == []
-    assert store.list_runs() == []
+    assert dispatched == []
+    fresh = FileRunStore(store.root)
+    assert fresh.list_admissions() == []
+    assert fresh.list_runs() == []
 
 
 def test_captured_invocation_freezes_input(tmp_path: Path) -> None: