sched: add durable admission record and admitted run view (T05)

src/wf_api/run_lifecycle.py

@@ -11,6 +11,7 @@ from wf_artifacts import (
     DiagnosticSeverity,
     PinnedRunEnvironment,
     ResumeReadiness,
+    RunAdmission,
     RunCheckpoint,
     RunStore,
     StoredRunStatus,
@@ -43,6 +44,88 @@ def create_pinned_environment(
     )
 
 
+def persist_admission(
+    *,
+    store: RunStore,
+    run_id: str,
+    environment: PinnedRunEnvironment,
+    resolved_input: dict[str, object],
+    max_steps: int | None,
+    scheduled_at: datetime | None = None,
+    schedule_id: str | None = None,
+    schedule_revision: int | None = None,
+    deployment_revision: int | None = None,
+) -> RunAdmission:
+    """Persist the authoritative admission record before dispatch.
+
+    The occurrence is decided here: the caller must treat this persist as the
+    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")
+    admission = RunAdmission(
+        id=run_id,
+        environment=environment,
+        resolved_input=frozen,
+        max_steps=max_steps,
+        scheduled_at=scheduled_at,
+        schedule_id=schedule_id,
+        schedule_revision=schedule_revision,
+        deployment_revision=deployment_revision,
+        created_at=datetime.now(UTC),
+    )
+    store.save_admission(admission)
+    return admission
+
+
+def materialize_admitted_view(
+    *,
+    store: RunStore,
+    admission: RunAdmission,
+) -> WorkflowRunRecord:
+    """Materialize the run admission view using the admission identity.
+
+    The view carries no checkpoint, trace, output, or step counts: the outcome
+    is unknown until dispatch completes and persists a stopped checkpoint.
+    """
+    now = datetime.now(UTC)
+    try:
+        existing = store.get_run(admission.id)
+    except KeyError:
+        existing = None
+    if existing is not None:
+        return existing
+    record = WorkflowRunRecord(
+        id=admission.id,
+        status=StoredRunStatus.ADMITTED,
+        resume_readiness=ResumeReadiness.NOT_APPLICABLE,
+        environment=admission.environment,
+        latest_checkpoint_id=None,
+        created_at=admission.created_at,
+        updated_at=now,
+    )
+    store.save_run(record)
+    return record
+
+
+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.
+    """
+    try:
+        return store.get_run(run_id)
+    except KeyError:
+        pass
+    admission = store.get_admission(run_id)
+    return materialize_admitted_view(store=store, admission=admission)
+
+
 def persist_stopped_run(
     *,
     store: RunStore,
@@ -65,9 +148,16 @@ def persist_stopped_run(
     sequence = 1
     created_at = now
     if run_id is not None:
+        try:
             existing = store.get_run(run_id)
+        except KeyError:
+            existing = None
+        if existing is not None:
             created_at = existing.created_at
+            try:
                 sequence = store.get_latest_checkpoint(run_id).sequence + 1
+            except KeyError:
+                sequence = 1
 
     status = StoredRunStatus(run.status.value)
     readiness = (
@@ -130,6 +220,10 @@ def restore_interrupted_run(
 def load_stored_run(store: RunStore, run_id: str) -> tuple[WorkflowRunRecord, RunState]:
     """Load any stopped run record together with its latest typed checkpoint."""
     record = store.get_run(run_id)
+    if record.status is StoredRunStatus.ADMITTED:
+        raise ValueError(
+            f"workflow run {run_id!r} is admitted but has no stopped checkpoint yet"
+        )
     checkpoint = store.get_latest_checkpoint(run_id)
     return record, load_run_state(checkpoint.state.model_dump(mode="json"))
 

src/wf_artifacts/__init__.py

@@ -46,6 +46,7 @@ from .runs import (
     FileRunStore,
     PinnedRunEnvironment,
     ResumeReadiness,
+    RunAdmission,
     RunCheckpoint,
     RunStore,
     StoredRunStatus,
@@ -72,6 +73,7 @@ __all__ = [
     "PinnedRunEnvironment",
     "RequiredCapability",
     "ResumeReadiness",
+    "RunAdmission",
     "RunCheckpoint",
     "RunStore",
     "SourceBinding",

src/wf_artifacts/runs/__init__.py

@@ -2,6 +2,7 @@ from .models import (
     CheckpointReason,
     PinnedRunEnvironment,
     ResumeReadiness,
+    RunAdmission,
     RunCheckpoint,
     StoredRunStatus,
     VersionedCheckpointState,
@@ -15,6 +16,7 @@ __all__ = [
     "FileRunStore",
     "PinnedRunEnvironment",
     "ResumeReadiness",
+    "RunAdmission",
     "RunCheckpoint",
     "RunStore",
     "StoredRunStatus",

src/wf_artifacts/runs/models.py

@@ -25,8 +25,13 @@ def ensure_run_id(run_id: str) -> str:
 
 
 class StoredRunStatus(StrEnum):
-    """Stopped runtime statuses supported by durable run persistence."""
+    """Durable run statuses including pre-dispatch admission.
 
+    ``ADMITTED`` marks a durably admitted run with no stopped checkpoint yet.
+    Never fabricate a checkpoint, trace, output, or step count for such runs.
+    """
+
+    ADMITTED = "admitted"
     INTERRUPTED = "interrupted"
     COMPLETED = "completed"
     FAILED = "failed"
@@ -67,12 +72,35 @@ class WorkflowRunRecord(BaseModel):
     status: StoredRunStatus
     resume_readiness: ResumeReadiness
     environment: PinnedRunEnvironment
-    latest_checkpoint_id: str = Field(pattern=RUN_ID_PATTERN)
+    latest_checkpoint_id: str | None = Field(default=None, pattern=RUN_ID_PATTERN)
     diagnostics: list[DependencyDiagnostic] = Field(default_factory=list)
     created_at: datetime
     updated_at: datetime
 
 
+class RunAdmission(BaseModel):
+    """Authoritative admission record persisted before dispatch.
+
+    The admission freezes the invocation (pinned environment, resolved input,
+    limits, deployment/schedule revisions, resolved UTC instant) under a
+    preassigned run identity. It is the recovery authority for partial
+    multi-file writes: the run view (``WorkflowRunRecord``) is materialized
+    from this same identity, and dispatch uses only the captured invocation.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str = Field(pattern=RUN_ID_PATTERN)
+    environment: PinnedRunEnvironment
+    resolved_input: dict[str, Any]
+    max_steps: int | None = None
+    scheduled_at: datetime | None = None
+    schedule_id: str | None = None
+    schedule_revision: int | None = None
+    deployment_revision: int | None = None
+    created_at: datetime
+
+
 class VersionedCheckpointState(BaseModel):
     """Lenient read envelope for stopped-run checkpoints.
 

src/wf_artifacts/runs/store.py

@@ -4,11 +4,11 @@ import json
 from pathlib import Path
 from threading import RLock
 
-from .models import RunCheckpoint, WorkflowRunRecord, ensure_run_id
+from .models import RunAdmission, RunCheckpoint, WorkflowRunRecord, ensure_run_id
 
 
 class RunStore:
-    """Persistence boundary for stopped run summaries and checkpoints."""
+    """Persistence boundary for admitted runs, summaries, and checkpoints."""
 
     def save_run(self, run: WorkflowRunRecord) -> None:
         raise NotImplementedError
@@ -28,6 +28,18 @@ class RunStore:
     def list_checkpoints(self, run_id: str) -> list[RunCheckpoint]:
         raise NotImplementedError
 
+    def save_admission(self, admission: RunAdmission) -> None:
+        raise NotImplementedError
+
+    def get_admission(self, run_id: str) -> RunAdmission:
+        raise NotImplementedError
+
+    def list_admissions(self) -> list[RunAdmission]:
+        raise NotImplementedError
+
+    def allocate_run_id(self) -> str:
+        raise NotImplementedError
+
 
 class FileRunStore(RunStore):
     """JSON file-backed stopped-run store for local development and tests.
@@ -87,6 +99,41 @@ class FileRunStore(RunStore):
             for path in sorted(directory.glob("*.json"))
         ]
 
+    def save_admission(self, admission: RunAdmission) -> None:
+        with self._lock:
+            self._write_json(
+                self._admission_path(admission.id),
+                admission.model_dump(mode="json"),
+            )
+
+    def get_admission(self, run_id: str) -> RunAdmission:
+        path = self._admission_path(run_id)
+        if not path.exists():
+            raise KeyError(f"unknown run admission {run_id!r}")
+        return RunAdmission.model_validate_json(path.read_text(encoding="utf-8"))
+
+    def list_admissions(self) -> list[RunAdmission]:
+        return [
+            RunAdmission.model_validate_json(path.read_text(encoding="utf-8"))
+            for path in sorted(self.runs_dir.glob("*/admission.json"))
+        ]
+
+    def allocate_run_id(self) -> str:
+        """Allocate a store-backed run identity that survives restart."""
+        with self._lock:
+            seq_path = self.runs_dir / "_run_id_seq.json"
+            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
+            seq += 1
+            self._write_json(seq_path, {"seq": seq})
+            return f"run-{seq:06d}"
+
     def _write_json(self, path: Path, payload: object) -> None:
         path.parent.mkdir(parents=True, exist_ok=True)
         temp_path = path.with_suffix(".json.tmp")
@@ -104,5 +151,8 @@ class FileRunStore(RunStore):
     def _run_path(self, run_id: str) -> Path:
         return self._run_directory(run_id) / "run.json"
 
+    def _admission_path(self, run_id: str) -> Path:
+        return self._run_directory(run_id) / "admission.json"
+
     def _checkpoint_path(self, run_id: str, sequence: int) -> Path:
         return self._run_directory(run_id) / "checkpoints" / f"{sequence:06d}.json"

tests/artifacts/test_run_admission.py

@@ -0,0 +1,181 @@
+"""Durable admission representation: persist before dispatch (T05)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+
+from wf_api.run_lifecycle import (
+    load_stored_run,
+    materialize_admitted_view,
+    persist_admission,
+)
+from wf_artifacts import (
+    PinnedRunEnvironment,
+    StoredRunStatus,
+    WorkflowArtifact,
+    WorkflowDeployment,
+)
+from wf_artifacts.runs.store import FileRunStore
+
+
+def _env() -> PinnedRunEnvironment:
+    deployment = WorkflowDeployment(
+        id="dep-1",
+        artifact_id="wf-1",
+        artifact_version=1,
+        bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+    )
+    artifact = WorkflowArtifact(
+        id="wf-1",
+        version=1,
+        title="Wf-1",
+        input_schema={"type": "object", "properties": {}},
+        output_schema={"type": "object", "properties": {}},
+        outcomes=("ok",),
+        plan={"name": "wf-1", "nodes": [], "edges": []},
+    )
+    return PinnedRunEnvironment(
+        deployment=deployment, root_artifact=artifact, child_artifacts=[]
+    )
+
+
+def test_admission_persists_authority_before_view(tmp_path: Path) -> None:
+    store = FileRunStore(tmp_path)
+    run_id = store.allocate_run_id()
+    at = datetime(2026, 9, 8, 13, 0, tzinfo=UTC)
+    admission = persist_admission(
+        store=store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={"team": "eng"},
+        max_steps=100,
+        scheduled_at=at,
+        schedule_id="sched-1",
+        schedule_revision=2,
+    )
+    assert admission.id == run_id
+    assert admission.scheduled_at == at
+    # Admission record exists before any run view or checkpoint.
+    assert store.get_admission(run_id).id == run_id
+    with pytest.raises(KeyError):
+        store.get_run(run_id)
+    record = materialize_admitted_view(store=store, admission=admission)
+    assert record.status is StoredRunStatus.ADMITTED
+    assert record.latest_checkpoint_id is None
+    # Inspection distinguishes admitted (no checkpoint) from stopped.
+    with pytest.raises(KeyError, match="no checkpoints"):
+        store.get_latest_checkpoint(run_id)
+
+
+def test_admission_view_round_trips_real_serialization(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={"a": [1, None, "x"]},
+        max_steps=None,
+        scheduled_at=None,
+        schedule_id=None,
+        schedule_revision=None,
+    )
+    record = materialize_admitted_view(store=store, admission=admission)
+    reloaded = store.get_run(run_id)
+    assert reloaded.id == record.id
+    assert reloaded.status is StoredRunStatus.ADMITTED
+    reloaded_admission = store.get_admission(run_id)
+    assert reloaded_admission.resolved_input == {"a": [1, None, "x"]}
+    assert reloaded_admission.environment.deployment.id == "dep-1"
+
+
+def test_run_ids_are_store_backed_across_restart(tmp_path: Path) -> None:
+    first = FileRunStore(tmp_path).allocate_run_id()
+    second = FileRunStore(tmp_path).allocate_run_id()
+    assert first != second
+    # Reopening the store never reuses identities.
+    third = FileRunStore(tmp_path).allocate_run_id()
+    assert third not in {first, second}
+
+
+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()
+
+    def _boom(path: Path, payload: object) -> None:
+        raise OSError("injected write failure")
+
+    monkeypatch.setattr(store, "_write_json", _boom)
+    with pytest.raises(OSError, match="injected"):
+        persist_admission(
+            store=store,
+            run_id=run_id,
+            environment=_env(),
+            resolved_input={},
+            max_steps=None,
+            scheduled_at=None,
+            schedule_id=None,
+            schedule_revision=None,
+        )
+    assert store_module.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
+) -> None:
+    store = FileRunStore(tmp_path)
+    run_id = store.allocate_run_id()
+    admission = persist_admission(
+        store=store,
+        run_id=run_id,
+        environment=_env(),
+        resolved_input={"team": "eng"},
+        max_steps=None,
+        scheduled_at=None,
+        schedule_id=None,
+        schedule_revision=None,
+    )
+    assert admission.id == run_id
+    # Crash before the view is materialized: admission exists, view missing.
+    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
+    # Idempotent: second recovery returns the same view.
+    again = recover_admission_view(store=store, run_id=run_id)
+    assert again.id == run_id
+
+
+def test_admitted_run_has_no_fabricated_checkpoint_or_output(
+    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,
+    )
+    record = materialize_admitted_view(store=store, admission=admission)
+    assert record.latest_checkpoint_id is None
+    assert store.list_checkpoints(run_id) == []
+    # Loading a stopped run from an admitted view must fail closed, not
+    # fabricate trace/output/step counts.
+    with pytest.raises(ValueError, match="admitted"):
+        load_stored_run(store, run_id)