From d0d9fd581ec755beca3222afc2dc5ec563a088d9 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:19:28 +0700 Subject: [PATCH] sched: add durable admission record and admitted run view (T05) --- src/wf_api/run_lifecycle.py | 100 +++++++++++++- src/wf_artifacts/__init__.py | 2 + src/wf_artifacts/runs/__init__.py | 2 + src/wf_artifacts/runs/models.py | 32 ++++- src/wf_artifacts/runs/store.py | 54 +++++++- tests/artifacts/test_run_admission.py | 181 ++++++++++++++++++++++++++ 6 files changed, 364 insertions(+), 7 deletions(-) create mode 100644 tests/artifacts/test_run_admission.py diff --git a/src/wf_api/run_lifecycle.py b/src/wf_api/run_lifecycle.py index 5a362548..eb2a6852 100644 --- a/src/wf_api/run_lifecycle.py +++ b/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: - existing = store.get_run(run_id) - created_at = existing.created_at - sequence = store.get_latest_checkpoint(run_id).sequence + 1 + 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")) diff --git a/src/wf_artifacts/__init__.py b/src/wf_artifacts/__init__.py index 6773d9bd..1f02f4c0 100644 --- a/src/wf_artifacts/__init__.py +++ b/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", diff --git a/src/wf_artifacts/runs/__init__.py b/src/wf_artifacts/runs/__init__.py index 172efe7c..4e7c7c26 100644 --- a/src/wf_artifacts/runs/__init__.py +++ b/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", diff --git a/src/wf_artifacts/runs/models.py b/src/wf_artifacts/runs/models.py index 49eb5238..7e26ea7a 100644 --- a/src/wf_artifacts/runs/models.py +++ b/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. diff --git a/src/wf_artifacts/runs/store.py b/src/wf_artifacts/runs/store.py index 40cff285..d8ac5965 100644 --- a/src/wf_artifacts/runs/store.py +++ b/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" diff --git a/tests/artifacts/test_run_admission.py b/tests/artifacts/test_run_admission.py new file mode 100644 index 00000000..4be9a104 --- /dev/null +++ b/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)