diff --git a/docs/thesis/generate.ps1 b/docs/thesis/generate.ps1 index e9158d6a..4ab9cb33 100644 --- a/docs/thesis/generate.ps1 +++ b/docs/thesis/generate.ps1 @@ -93,7 +93,7 @@ if (-not (Test-Path $include_markdown_filter)) { } $agent_results = Join-Path $PSScriptRoot "agent-challenge-results.md" -if ((Test-RenderNeedsAgentResults $RemainingArgs) -and -not (Test-Path $agent_results)) { +if ((Test-RenderNeedsAgentResults (@($InputFile) + $RemainingArgs)) -and -not (Test-Path $agent_results)) { Write-Error "agent-challenge-results.md is missing. Run generate_agent_challenge_evaluation.py first." exit 1 } diff --git a/docs/thesis/system-design-implementation.md b/docs/thesis/system-design-implementation.md index 6da82a87..1d1df990 100644 --- a/docs/thesis/system-design-implementation.md +++ b/docs/thesis/system-design-implementation.md @@ -1433,7 +1433,9 @@ would still need to preserve the lifecycle's transaction and ownership contracts; changing the storage engine alone would not prove those properties. -Scheduled deployment execution is not implemented. Nor does exposing a +Scheduled deployment execution is implemented within the documented first +slice. It remains bounded by the filesystem-backed, single-process store and +the explicitly enabled local/static server composition. Nor does exposing a provider's callable operations imply that its interactive widgets or entire user experience are reproduced through the workflow API. @@ -1501,11 +1503,11 @@ and data-binding model are understandable without implementation knowledge. ## Scheduling and the Surrounding Application -Scheduled deployment execution is a required product direction, not an -implemented capability. It introduces trigger identity, overlap policy, -and recovery decisions in addition to time-expression parsing. It should -build on the same run lifecycle rather than create a separate execution -model. Suspending an already-running workflow until a time or event is a +Scheduled deployment execution is implemented for the current slice. It +introduces trigger identity, overlap policy, and recovery decisions in +addition to time-expression parsing, and builds on the same run lifecycle +rather than creating a separate execution model. Extending it to distributed +workers or suspending an already-running workflow until a time or event is a related but distinct design question; a wait node is not specified here. The surrounding application is intended to combine assistant-backed chat diff --git a/docs/wf_cli.md b/docs/wf_cli.md index 98a95fa0..6f5cf3ed 100644 --- a/docs/wf_cli.md +++ b/docs/wf_cli.md @@ -117,7 +117,8 @@ wf-rpc-server --store-root .wf_store --enable-scheduler ``` or set `server.scheduler.enabled` (plus optional `poll_interval_s`, -`max_concurrent_runs`, `drain_grace_s`) in the neutral config. See +`max_concurrent_runs`, `drain_grace_s`) in a local/static neutral config. +MCP-backed servers reject this setting. See [`deployment scheduling operations`](deployment_scheduling.md). `admin registry` shows desired persisted source entries. It is separate from diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py index bfd64d5a..95d89c60 100644 --- a/src/wf_api/schedules.py +++ b/src/wf_api/schedules.py @@ -442,7 +442,6 @@ class WorkflowScheduleApi: schedule = store.get_schedule(schedule_id) schedule.deleted = True store.save_schedule(schedule) - store.save_candidate(None, schedule_id=schedule_id) return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) async def list_schedule_occurrences( diff --git a/src/wf_artifacts/runs/store.py b/src/wf_artifacts/runs/store.py index 7b3a37c7..ed1415ea 100644 --- a/src/wf_artifacts/runs/store.py +++ b/src/wf_artifacts/runs/store.py @@ -151,46 +151,32 @@ class FileRunStore(RunStore): 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.""" + def _next_sequence(self, filename: str, label: str) -> int: + """Atomically allocate the next value from one durable sequence file.""" with self._lock: - seq_path = self.runs_dir / "_run_id_seq.json" + seq_path = self.runs_dir / filename seq = 0 if seq_path.exists(): try: 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}" - ) + raise ValueError(f"corrupt {label} 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}" - - def allocate_resume_attempt_id(self) -> int: - """Allocate a store-backed resume-attempt identity (no reuse).""" - with self._lock: - seq_path = self.runs_dir / "_resume_attempt_seq.json" - seq = 0 - if seq_path.exists(): - try: - 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 attempt sequence at {seq_path}: {raw!r}" - ) - seq = seq_value - except (ValueError, AttributeError, TypeError) as exc: - raise ValueError(f"corrupt attempt sequence at {seq_path}") from exc + raise ValueError(f"corrupt {label} at {seq_path}") from exc seq += 1 self._write_json(seq_path, {"seq": seq}) return seq + def allocate_run_id(self) -> str: + """Allocate a store-backed run identity that survives restart.""" + return f"run-{self._next_sequence('_run_id_seq.json', 'run-id sequence'):06d}" + + def allocate_resume_attempt_id(self) -> int: + """Allocate a store-backed resume-attempt identity (no reuse).""" + return self._next_sequence("_resume_attempt_seq.json", "attempt sequence") + def save_resume_attempt(self, attempt: ResumeAttempt) -> None: with self._lock: self._write_json( diff --git a/tests/artifacts/test_store.py b/tests/artifacts/test_store.py index 566b9fa5..b9cb499a 100644 --- a/tests/artifacts/test_store.py +++ b/tests/artifacts/test_store.py @@ -94,6 +94,28 @@ def test_concurrent_deployment_saves_advance_revision_without_lost_updates( assert store.get_deployment("concurrent.personal").revision == 9 +def test_deployment_revision_increments_on_save(tmp_path) -> None: + artifacts = FileWorkflowArtifactStore(tmp_path) + artifacts.save_deployment( + WorkflowDeployment( + id="dep-1", + artifact_id="wf-1", + artifact_version=1, + bindings=[], + ) + ) + assert artifacts.get_deployment("dep-1").revision == 1 + artifacts.save_deployment( + WorkflowDeployment( + id="dep-1", + artifact_id="wf-1", + artifact_version=2, + bindings=[], + ) + ) + assert artifacts.get_deployment("dep-1").revision == 2 + + def test_file_store_loads_legacy_artifact_and_rewrites_canonical_shape( tmp_path, ) -> None: diff --git a/tests/docs/test_big_doc_links.py b/tests/docs/test_big_doc_links.py index 4d6e289e..87dcf5bb 100644 --- a/tests/docs/test_big_doc_links.py +++ b/tests/docs/test_big_doc_links.py @@ -70,7 +70,7 @@ def test_thesis_retires_campaign_without_removing_historical_evidence() -> None: archive = ROOT / "docs/historical/thesis/2026-09-07-retired-agent-evaluation.md" historical = archive.read_text(encoding="utf-8") assert "# Agent Challenge Harness" in historical - assert "36" in historical + assert "36 audited trials" in historical def test_thesis_bundle_has_reproducible_agent_evaluation_assets() -> None: diff --git a/tests/scheduling/test_lifecycle.py b/tests/scheduling/test_lifecycle.py index f1df75e6..47924fad 100644 --- a/tests/scheduling/test_lifecycle.py +++ b/tests/scheduling/test_lifecycle.py @@ -17,6 +17,7 @@ from pathlib import Path from typing import Any, cast import pytest +from pydantic import ValidationError from tests.artifacts.test_run_store import artifact as _artifact from tests.artifacts.test_run_store import deployment as _deployment @@ -588,7 +589,7 @@ async def test_tick_error_recorded_and_loop_continues(tmp_path: Path) -> None: broken_path = tmp_path / "schedules" / "broken" / "schedule.json" good_text = broken_path.read_text(encoding="utf-8") broken_path.write_text('{"id": "broken", "trigger": {"kind": "bogus"}}') - with pytest.raises(Exception): + with pytest.raises(ValidationError): await service.poll_once(intended + timedelta(seconds=1)) assert service.last_tick_error is not None assert service.run_store.list_runs() == [] @@ -689,7 +690,10 @@ async def test_live_runtime_failure_spares_healthy_sibling( hang_id, sib_id = await _poll_two(service, intended) await _wait_for(lambda: runtime.started.is_set()) await _wait_for( - lambda: service.run_store.get_run(sib_id).status.value == "failed" + lambda: ( + service.run_store.get_run(sib_id).status.value == "failed" + and not service.run_store.is_executing(sib_id) + ) ) failed = service.run_store.get_run(sib_id) assert failed.status.value == "failed" @@ -744,7 +748,10 @@ async def test_live_settlement_failure_spares_healthy_sibling( hang_id, sib_id = await _poll_two(service, intended) await _wait_for(lambda: runtime.started.is_set()) await _wait_for( - lambda: service.run_store.get_run(sib_id).status.value == "failed" + lambda: ( + service.run_store.get_run(sib_id).status.value == "failed" + and not service.run_store.is_executing(sib_id) + ) ) assert fired["done"] assert not service.run_store.is_executing(sib_id) diff --git a/tests/scheduling/test_ownership_binding.py b/tests/scheduling/test_ownership_binding.py index d80a6f8b..e1df3c9d 100644 --- a/tests/scheduling/test_ownership_binding.py +++ b/tests/scheduling/test_ownership_binding.py @@ -22,7 +22,6 @@ from tests.scheduling.controlled import ( ) from wf_artifacts.runs.store import FileRunStore from wf_scheduling.calendar import OneShotSource -from wf_scheduling.dispatch import StillRunning from wf_scheduling.models import Schedule from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError from wf_scheduling.poll import Scheduler @@ -80,13 +79,13 @@ def test_unrelated_held_lock_rejected_before_writes(tmp_path: Path) -> None: def spy(admission: Any, now: datetime) -> Any: calls.append(admission.id) - return StillRunning() + raise AssertionError("dispatcher must not run without covering ownership") sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") ownership = SchedulerOwnership(tmp_path / "other", owner="unrelated").acquire() try: - sched = _scheduler(sched_store, run_store, ownership) + sched = _scheduler(sched_store, run_store, ownership, script={"*": spy}) intended = ts(2026, 9, 8, 12, 0) _due(sched, sched_store, intended) with pytest.raises(SecondOwnerError): @@ -103,9 +102,12 @@ def test_two_lock_dirs_cannot_operate_same_stores(tmp_path: Path) -> None: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") intended = ts(2026, 9, 8, 12, 0) - covering = SchedulerOwnership(tmp_path, owner="covering").acquire() - other = SchedulerOwnership(tmp_path / "other", owner="other").acquire() + covering: SchedulerOwnership | None = None + other: SchedulerOwnership | None = None try: + covering = SchedulerOwnership(tmp_path, owner="covering").acquire() + other = SchedulerOwnership(tmp_path / "other", owner="other").acquire() + assert covering is not None and other is not None first = _scheduler(sched_store, run_store, covering, script={"*": "hang"}) _due(first, sched_store, intended) assert first.poll(intended)["a"].startswith("admit:run-") @@ -117,8 +119,10 @@ def test_two_lock_dirs_cannot_operate_same_stores(tmp_path: Path) -> None: second.poll(intended + timedelta(minutes=1)) assert len(run_store.list_runs()) == 1 finally: - covering.release() - other.release() + if covering is not None: + covering.release() + if other is not None: + other.release() def test_recover_with_unrelated_lock_rejected(tmp_path: Path) -> None: diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index ec3dc3dd..12d239b4 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -114,12 +114,9 @@ def _history(store: FileScheduleStore, sid: str) -> list[dict]: return page["occurrences"] # type: ignore[return-value] -def test_overlap_skip_blocks_and_late_drops() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched, store, runs, sources = _harness(root, capacity=4, script={"*": "hang"}) +def test_overlap_skip_blocks_and_late_drops(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, capacity=4, script={"*": "hang"}) + try: t0 = ts(2026, 9, 8, 12, 0) _add( sched, @@ -137,15 +134,13 @@ def test_overlap_skip_blocks_and_late_drops() -> None: sched.poll(t0 + timedelta(minutes=30)) kinds = [(r["kind"], r["resolved_at"]) for r in _history(store, "a")] assert any(k == "skipped-misfire" for k, _ in kinds) + finally: sched.ownership.release() -def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched, store, runs, sources = _harness(root, capacity=0) +def test_latest_coalesces_to_one_candidate_and_no_double_admit(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, capacity=0) + try: _add( sched, store, @@ -170,6 +165,7 @@ def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None: 2026, 9, 8, 13, 0 ) assert store.get_candidate("h") is None + finally: sched.ownership.release() @@ -217,7 +213,6 @@ def test_schedule_edit_between_poll_snapshot_and_admission_cannot_overwrite_term } ) super().update_schedule(edited, expected_revision=current.revision) - return current return current sched_store = EditOnPollRead(tmp_path / "sched") @@ -277,12 +272,9 @@ def test_trigger_edit_refreshes_managed_calendar_before_polling(tmp_path: Path) sched.ownership.release() -def test_parallel_limits_and_interrupted_slots() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched, store, runs, sources = _harness(root, capacity=4, script={"*": "hang"}) +def test_parallel_limits_and_interrupted_slots(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, capacity=4, script={"*": "hang"}) + try: t0 = ts(2026, 9, 8, 12, 0) _add( sched, @@ -319,17 +311,15 @@ def test_parallel_limits_and_interrupted_slots() -> None: r["kind"] == "skipped-overlap" and "12:15" in str(r["resolved_at"]) for r in _history(store, "p") ) + finally: sched.ownership.release() -def test_pause_is_not_downtime() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched, store, runs, sources = _harness( - root, capacity=4, script={"*": "complete"} - ) +def test_pause_is_not_downtime(tmp_path: Path) -> None: + sched, store, runs, sources = _harness( + tmp_path, capacity=4, script={"*": "complete"} + ) + try: _add( sched, store, @@ -352,6 +342,7 @@ def test_pause_is_not_downtime() -> None: ] assert "2026-09-08T10:00:00+00:00" not in admitted assert "2026-09-08T11:00:00+00:00" not in admitted + finally: sched.ownership.release() @@ -359,22 +350,35 @@ def test_long_downtime_is_bounded(tmp_path: Path) -> None: sched, store, runs, sources = _harness( tmp_path, capacity=4, script={"*": "complete"} ) - src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) - _add(sched, store, sources, "m", src, ts(2023, 9, 8, 12, 0), misfire="latest") - sched.poll(ts(2026, 9, 8, 12, 0, 0)) - assert src.next_calls + src.prev_calls <= SCAN_CAP + 2 - admitted = [r for r in _history(store, "m") if r["kind"] == "admitted"] - assert len(admitted) == 1 - # Latest-eligible <= now: a poll exactly at a due instant selects that - # instant (F3), not its exclusive predecessor. - assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts(2026, 9, 8, 12, 0) - assert ( - len([r for r in _history(store, "m") if r["kind"] == "interval-summary"]) == 1 - ) - # A second poll at the same instant admits nothing more. - sched.poll(ts(2026, 9, 8, 12, 0, 0)) - assert len([r for r in _history(store, "m") if r["kind"] == "admitted"]) == 1 - sched.ownership.release() + try: + src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) + _add( + sched, + store, + sources, + "m", + src, + ts(2023, 9, 8, 12, 0), + misfire="latest", + ) + sched.poll(ts(2026, 9, 8, 12, 0, 0)) + assert src.next_calls + src.prev_calls <= SCAN_CAP + 2 + admitted = [r for r in _history(store, "m") if r["kind"] == "admitted"] + assert len(admitted) == 1 + # Latest-eligible <= now: a poll exactly at a due instant selects that + # instant (F3), not its exclusive predecessor. + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts( + 2026, 9, 8, 12, 0 + ) + assert ( + len([r for r in _history(store, "m") if r["kind"] == "interval-summary"]) + == 1 + ) + # A second poll at the same instant admits nothing more. + sched.poll(ts(2026, 9, 8, 12, 0, 0)) + assert len([r for r in _history(store, "m") if r["kind"] == "admitted"]) == 1 + finally: + sched.ownership.release() def test_fairness_slow_schedule_not_starved(tmp_path: Path) -> None: diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py index 9bb80e2c..70bb7612 100644 --- a/tests/scheduling/test_preparation.py +++ b/tests/scheduling/test_preparation.py @@ -10,6 +10,7 @@ inventing a run. from __future__ import annotations import inspect +import typing from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, cast @@ -105,9 +106,9 @@ def test_production_scheduler_has_no_fixture_seams() -> None: # Collaborators are protocols consumed here, implemented elsewhere. assert "InvocationPreparer" in source assert "RunDispatcher" in source - assert isinstance( - poll_module.Scheduler.__init__.__annotations__["preparer"], object - ) + hints = typing.get_type_hints(poll_module.Scheduler.__init__) + assert hints["preparer"] is InvocationPreparer + assert hints["dispatcher"] is RunDispatcher def test_preparer_contract_is_a_protocol() -> None: @@ -117,52 +118,59 @@ def test_preparer_contract_is_a_protocol() -> None: def test_occurrence_bindings_resolve_into_admission(tmp_path: Path) -> None: sched, store, runs = _scheduler(tmp_path, script={"*": "hang"}) - intended = ts(2026, 9, 8, 12, 0) - model = _sched_model( - "b", - input_bindings=[ - { - "target": "team", - "expression": {"kind": "literal", "value": "engineering"}, - }, - { - "target": "report_time", - "expression": {"kind": "occurrence", "field": "scheduled_at"}, - }, - { - "target": "which", - "expression": {"kind": "occurrence", "field": "schedule_id"}, - }, - ], - ) - store.create_schedule(model) - store.save_consumed("b", intended - timedelta(hours=1)) - sched.sources["b"] = OneShotSource(intended) - result = sched.poll(intended) - assert result["b"].startswith("admit:run-") - run_id = result["b"].split(":", 1)[1] - admission = runs.get_admission(run_id) - assert admission.resolved_input["team"] == "engineering" - assert admission.resolved_input["report_time"] == intended.isoformat() - assert admission.resolved_input["which"] == "b" - assert admission.deployment_revision == 3 - assert admission.schedule_revision == 1 - sched.ownership.release() + try: + intended = ts(2026, 9, 8, 12, 0) + model = _sched_model( + "b", + input_bindings=[ + { + "target": "team", + "expression": {"kind": "literal", "value": "engineering"}, + }, + { + "target": "report_time", + "expression": { + "kind": "occurrence", + "field": "scheduled_at", + }, + }, + { + "target": "which", + "expression": {"kind": "occurrence", "field": "schedule_id"}, + }, + ], + ) + store.create_schedule(model) + store.save_consumed("b", intended - timedelta(hours=1)) + sched.sources["b"] = OneShotSource(intended) + result = sched.poll(intended) + assert result["b"].startswith("admit:run-") + run_id = result["b"].split(":", 1)[1] + admission = runs.get_admission(run_id) + assert admission.resolved_input["team"] == "engineering" + assert admission.resolved_input["report_time"] == intended.isoformat() + assert admission.resolved_input["which"] == "b" + assert admission.deployment_revision == 3 + assert admission.schedule_revision == 1 + finally: + sched.ownership.release() def test_unknown_deployment_rejects_without_a_run(tmp_path: Path) -> None: sched, store, runs = _scheduler(tmp_path) - intended = ts(2026, 9, 8, 12, 0) - store.create_schedule(_sched_model("gone", deployment_id="dep-missing")) - store.save_consumed("gone", intended - timedelta(hours=1)) - sched.sources["gone"] = OneShotSource(intended) - assert sched.poll(intended) == {"gone": "admit:None"} - assert runs.list_runs() == [] - assert runs.list_admissions() == [] - page = store.list_occurrences("gone", limit=100) - kinds = [r["kind"] for r in cast(list[dict[str, Any]], page["occurrences"])] - assert kinds == ["preflight-rejected"] - sched.ownership.release() + try: + intended = ts(2026, 9, 8, 12, 0) + store.create_schedule(_sched_model("gone", deployment_id="dep-missing")) + store.save_consumed("gone", intended - timedelta(hours=1)) + sched.sources["gone"] = OneShotSource(intended) + assert sched.poll(intended) == {"gone": "admit:None"} + assert runs.list_runs() == [] + assert runs.list_admissions() == [] + page = store.list_occurrences("gone", limit=100) + kinds = [r["kind"] for r in cast(list[dict[str, Any]], page["occurrences"])] + assert kinds == ["preflight-rejected"] + finally: + sched.ownership.release() def test_deployment_revision_mismatch_rejects_pinned_environment() -> None: @@ -198,46 +206,50 @@ def test_missing_required_input_rejects_without_a_run(tmp_path: Path) -> None: dispatcher=ScriptedDispatcher({"*": "hang"}), ownership=SchedulerOwnership(tmp_path, owner="test").acquire(), ) - intended = ts(2026, 9, 8, 12, 0) - sched_store.create_schedule(_sched_model("need")) - sched_store.save_consumed("need", intended - timedelta(hours=1)) - sched.sources["need"] = OneShotSource(intended) - assert sched.poll(intended) == {"need": "admit:None"} - assert run_store.list_runs() == [] - page = sched_store.list_occurrences("need", limit=100) - entry = cast(list[dict[str, Any]], page["occurrences"])[0] - assert entry["kind"] == "preflight-rejected" - assert "missing-input" in entry["reason"] - sched.ownership.release() + try: + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule(_sched_model("need")) + sched_store.save_consumed("need", intended - timedelta(hours=1)) + sched.sources["need"] = OneShotSource(intended) + assert sched.poll(intended) == {"need": "admit:None"} + assert run_store.list_runs() == [] + page = sched_store.list_occurrences("need", limit=100) + entry = cast(list[dict[str, Any]], page["occurrences"])[0] + assert entry["kind"] == "preflight-rejected" + assert "missing-input" in entry["reason"] + finally: + sched.ownership.release() def test_conflicting_schedule_targets_reject_without_a_run(tmp_path: Path) -> None: sched, store, runs = _scheduler(tmp_path) - intended = ts(2026, 9, 8, 12, 0) - store.create_schedule( - _sched_model( - "conflict", - input_bindings=[ - { - "target": "a", - "expression": {"kind": "literal", "value": 1}, - }, - { - "target": "a.b", - "expression": {"kind": "literal", "value": 2}, - }, - ], + try: + intended = ts(2026, 9, 8, 12, 0) + store.create_schedule( + _sched_model( + "conflict", + input_bindings=[ + { + "target": "a", + "expression": {"kind": "literal", "value": 1}, + }, + { + "target": "a.b", + "expression": {"kind": "literal", "value": 2}, + }, + ], + ) ) - ) - store.save_consumed("conflict", intended - timedelta(hours=1)) - sched.sources["conflict"] = OneShotSource(intended) - assert sched.poll(intended) == {"conflict": "admit:None"} - assert runs.list_runs() == [] - page = store.list_occurrences("conflict", limit=100) - entry = cast(list[dict[str, Any]], page["occurrences"])[0] - assert entry["kind"] == "preflight-rejected" - assert entry["reason"].startswith("invalid-input:") - sched.ownership.release() + store.save_consumed("conflict", intended - timedelta(hours=1)) + sched.sources["conflict"] = OneShotSource(intended) + assert sched.poll(intended) == {"conflict": "admit:None"} + assert runs.list_runs() == [] + page = store.list_occurrences("conflict", limit=100) + entry = cast(list[dict[str, Any]], page["occurrences"])[0] + assert entry["kind"] == "preflight-rejected" + assert entry["reason"].startswith("invalid-input:") + finally: + sched.ownership.release() def test_preparer_rejection_type_shape() -> None: diff --git a/tests/scheduling/test_recovery.py b/tests/scheduling/test_recovery.py index 8bddf37f..9938073f 100644 --- a/tests/scheduling/test_recovery.py +++ b/tests/scheduling/test_recovery.py @@ -70,7 +70,7 @@ def test_recovery_materializes_missing_view_as_pending(tmp_path: Path) -> None: ownership.release() assert any("pending-dispatch" in d for d in diags) assert run_store.get_run(admission.id).status.value == "admitted" - assert sched_recovery._is_pending(run_store, admission.id) + assert run_store.is_pending_dispatch(admission.id) def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> None: @@ -207,4 +207,4 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: finally: ownership.release() assert run_store.get_run(admission.id).status.value == "completed" - assert not sched_recovery._is_pending(run_store, admission.id) + assert not run_store.is_pending_dispatch(admission.id) diff --git a/tests/scheduling/test_recovery_stability.py b/tests/scheduling/test_recovery_stability.py index bde04fd9..81dac580 100644 --- a/tests/scheduling/test_recovery_stability.py +++ b/tests/scheduling/test_recovery_stability.py @@ -13,6 +13,8 @@ from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, cast +import pytest + from tests.artifacts.test_run_store import artifact as _artifact from tests.artifacts.test_run_store import deployment as _deployment from wf_api.run_lifecycle import ( @@ -23,7 +25,7 @@ from wf_api.run_lifecycle import ( restore_interrupted_run, ) from wf_artifacts import PinnedRunEnvironment -from wf_artifacts.runs.models import ResumeAttempt +from wf_artifacts.runs.models import ResumeAttempt, StoredRunStatus from wf_artifacts.runs.store import FileRunStore from wf_core import RunState, RunStatus from wf_scheduling import recovery as sched_recovery @@ -128,180 +130,147 @@ def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, An ] -def test_abandonment_decision_is_stable_across_recovery() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched_store = FileScheduleStore(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - run_id = _admit_interrupted(run_store, intended) - _mark_active(run_store, run_id, 9, intended) - first = _recover(sched_store, run_store, intended) - assert any("failed-closed" in d for d in first) - record = run_store.get_run(run_id) - assert record.status.value == "failed" - assert record.resume_readiness.value == "not_applicable" - assert len(record.diagnostics) == 1 - # Fresh store objects across the restart boundary: everything stable. - sched_store2 = FileScheduleStore(root / "sched") - run_store2 = FileRunStore(root / "runs") - second = _recover(sched_store2, run_store2, intended + timedelta(minutes=1)) - assert not any(run_id in d for d in second) - again = run_store2.get_run(run_id) - assert again.status.value == "failed" - assert again.resume_readiness.value == "not_applicable" - assert len(again.diagnostics) == 1 - assert _entries(sched_store2, "a", "failed") == _entries( - sched_store, "a", "failed" - ) - assert len(_entries(sched_store2, "a", "failed")) == 1 +def test_abandonment_decision_is_stable_across_recovery(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_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + first = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in first) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert len(record.diagnostics) == 1 + # Fresh store objects across the restart boundary: everything stable. + sched_store2 = FileScheduleStore(tmp_path / "sched") + run_store2 = FileRunStore(tmp_path / "runs") + second = _recover(sched_store2, run_store2, intended + timedelta(minutes=1)) + assert not any(run_id in d for d in second) + again = run_store2.get_run(run_id) + assert again.status.value == "failed" + assert again.resume_readiness.value == "not_applicable" + assert len(again.diagnostics) == 1 + assert _entries(sched_store2, "a", "failed") == _entries(sched_store, "a", "failed") + assert len(_entries(sched_store2, "a", "failed")) == 1 -def test_failed_readiness_and_inspection_agree() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched_store = FileScheduleStore(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - run_id = _admit_interrupted(run_store, intended) - _mark_active(run_store, run_id, 9, intended) - _recover(sched_store, run_store, intended) - record, _ = load_stored_run(run_store, run_id) - assert record.resume_readiness.value == "not_applicable" - try: - restore_interrupted_run(run_store, run_id) - raise AssertionError("failed run must not restore as interrupted") - except ValueError: - pass +def test_failed_readiness_and_inspection_agree(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_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + _recover(sched_store, run_store, intended) + record, _ = load_stored_run(run_store, run_id) + assert record.resume_readiness.value == "not_applicable" + with pytest.raises(ValueError): + restore_interrupted_run(run_store, run_id) -def test_genuinely_newer_result_repairs_after_decision() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched_store = FileScheduleStore(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - run_id = _admit_interrupted(run_store, intended) - _mark_active(run_store, run_id, 9, intended) - _recover(sched_store, run_store, intended) - assert run_store.get_run(run_id).status.value == "failed" - # A genuinely newer stopped result under a new matching attempt: the - # newer checkpoint repairs the summary and completes the attempt. - _mark_active(run_store, run_id, 10, intended) - _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=10) - diags = _recover(sched_store, run_store, intended + timedelta(minutes=1)) - assert any("fresh-result-resumable" in d for d in diags) - record = run_store.get_run(run_id) - assert record.status.value == "interrupted" - assert record.resume_readiness.value == "ready" - assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] - interrupted = _entries(sched_store, "a", "interrupted") - assert {e["checkpoint_id"] for e in interrupted} == {f"{run_id}.000002"} +def test_genuinely_newer_result_repairs_after_decision(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_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + _recover(sched_store, run_store, intended) + assert run_store.get_run(run_id).status.value == "failed" + # A genuinely newer stopped result under a new matching attempt: the + # newer checkpoint repairs the summary and completes the attempt. + _mark_active(run_store, run_id, 10, intended) + _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=10) + diags = _recover(sched_store, run_store, intended + timedelta(minutes=1)) + assert any("fresh-result-resumable" in d for d in diags) + record = run_store.get_run(run_id) + assert record.status.value == "interrupted" + assert record.resume_readiness.value == "ready" + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + interrupted = _entries(sched_store, "a", "interrupted") + assert {e["checkpoint_id"] for e in interrupted} == {f"{run_id}.000002"} -def test_completed_mismatch_decision_is_stable() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched_store = FileScheduleStore(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - 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="a", - schedule_revision=1, - ) - materialize_admitted_view(store=run_store, admission=admission) - _stopped(run_store, run_id, RunStatus.COMPLETED, attempt_id=3) - _mark_active(run_store, run_id, 5, intended) - _recover(sched_store, run_store, intended) - assert run_store.get_run(run_id).status.value == "failed" - second = _recover( - FileScheduleStore(root / "sched"), - FileRunStore(root / "runs"), - intended + timedelta(minutes=1), - ) - assert not any(run_id in d for d in second) - assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1 +def test_completed_mismatch_decision_is_stable(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 = 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="a", + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + _stopped(run_store, run_id, RunStatus.COMPLETED, attempt_id=3) + _mark_active(run_store, run_id, 5, intended) + _recover(sched_store, run_store, intended) + assert run_store.get_run(run_id).status.value == "failed" + second = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert not any(run_id in d for d in second) + assert len(FileRunStore(tmp_path / "runs").get_run(run_id).diagnostics) == 1 -def test_legacy_failed_run_gains_history_without_refail() -> None: - import tempfile - - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - sched_store = FileScheduleStore(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - run_id = _admit_interrupted(run_store, intended) - record = run_store.get_run(run_id) - from wf_artifacts.runs.models import StoredRunStatus - - run_store.save_run(record.model_copy(update={"status": StoredRunStatus.FAILED})) - assert _entries(sched_store, "a", "failed") == [] - diags = _recover(sched_store, run_store, intended) - assert any("terminal-reconciled" in d for d in diags) - assert not any("failed-closed" in d for d in diags) - assert run_store.get_run(run_id).diagnostics == [] - assert len(_entries(sched_store, "a", "failed")) == 1 +def test_legacy_failed_run_gains_history_without_refail(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_interrupted(run_store, intended) + record = run_store.get_run(run_id) + run_store.save_run(record.model_copy(update={"status": StoredRunStatus.FAILED})) + assert _entries(sched_store, "a", "failed") == [] + diags = _recover(sched_store, run_store, intended) + assert any("terminal-reconciled" in d for d in diags) + assert not any("failed-closed" in d for d in diags) + assert run_store.get_run(run_id).diagnostics == [] + assert len(_entries(sched_store, "a", "failed")) == 1 -def test_crash_between_decision_writes_recovers_history_once() -> None: - import tempfile +def test_crash_between_decision_writes_recovers_history_once(tmp_path: Path) -> None: + class FailFailedHistoryOnce(FileScheduleStore): + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False - with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - - class FailFailedHistoryOnce(FileScheduleStore): - def __init__(self, root: Path) -> None: - super().__init__(root) + def append_history(self, record: Any) -> None: + if self.armed and getattr(record, "kind", None) == "failed": self.armed = False + raise OSError("injected failed-history failure") + super().append_history(record) - def append_history(self, record: Any) -> None: - if self.armed and getattr(record, "kind", None) == "failed": - self.armed = False - raise OSError("injected failed-history failure") - super().append_history(record) + sched_store = FailFailedHistoryOnce(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_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + sched_store.armed = True - sched_store = FailFailedHistoryOnce(root / "sched") - run_store = FileRunStore(root / "runs") - sched_store.create_schedule(_sched_model("a")) - intended = ts(2026, 9, 8, 12, 0) - run_id = _admit_interrupted(run_store, intended) - _mark_active(run_store, run_id, 9, intended) - sched_store.armed = True - import pytest - - with pytest.raises(OSError, match="injected failed-history failure"): - _recover(sched_store, run_store, intended) - record = run_store.get_run(run_id) - assert record.status.value == "failed" - assert len(record.diagnostics) == 1 - assert _entries(sched_store, "a", "failed") == [] - # The decision (status + reason) survived; only history is missing. - second = _recover( - FileScheduleStore(root / "sched"), - FileRunStore(root / "runs"), - intended + timedelta(minutes=1), - ) - assert not any("failed-closed" in d for d in second) - assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1 - assert len(_entries(FileScheduleStore(root / "sched"), "a", "failed")) == 1 + with pytest.raises(OSError, match="injected failed-history failure"): + _recover(sched_store, run_store, intended) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert len(record.diagnostics) == 1 + assert _entries(sched_store, "a", "failed") == [] + # The decision (status + reason) survived; only history is missing. + second = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert not any("failed-closed" in d for d in second) + assert len(FileRunStore(tmp_path / "runs").get_run(run_id).diagnostics) == 1 + assert len(_entries(FileScheduleStore(tmp_path / "sched"), "a", "failed")) == 1 diff --git a/tests/scheduling/test_schedule_store.py b/tests/scheduling/test_schedule_store.py index eb5dabfc..04dabacd 100644 --- a/tests/scheduling/test_schedule_store.py +++ b/tests/scheduling/test_schedule_store.py @@ -141,7 +141,6 @@ def test_candidate_is_at_most_one(tmp_path: Path) -> None: store.save_candidate(first, schedule_id="a") store.save_candidate(second, schedule_id="a") assert store.get_candidate("a") is not None - assert store.get_candidate("a") is not None assert store.get_candidate("a").intended_at == datetime( # type: ignore[union-attr] 2026, 9, 8, 13, 0, tzinfo=UTC ) @@ -202,30 +201,6 @@ def test_inspection_payload_carries_contract_fields(tmp_path: Path) -> None: assert row["reason"] == "active=['run-1']" -def test_deployment_revision_increments_on_save(tmp_path: Path) -> None: - from wf_artifacts import FileWorkflowArtifactStore, WorkflowDeployment - - artifacts = FileWorkflowArtifactStore(tmp_path) - artifacts.save_deployment( - WorkflowDeployment( - id="dep-1", - artifact_id="wf-1", - artifact_version=1, - bindings=[], - ) - ) - assert artifacts.get_deployment("dep-1").revision == 1 - artifacts.save_deployment( - WorkflowDeployment( - id="dep-1", - artifact_id="wf-1", - artifact_version=2, - bindings=[], - ) - ) - assert artifacts.get_deployment("dep-1").revision == 2 - - def _tied_entry( kind: str, checkpoint_id: str | None, @@ -248,13 +223,14 @@ def _traverse(store: FileScheduleStore, limit: int) -> list[dict]: """Walk every page to the end, returning all rows in visit order.""" rows: list[dict] = [] cursor: str | None = None - while True: + for _ in range(1000): page = store.list_occurrences("a", cursor=cursor, limit=limit) rows.extend(page["occurrences"]) # type: ignore[arg-type] cursor = page["next_cursor"] # type: ignore[assignment] if cursor is None: assert page["total"] == len(rows) return rows + raise AssertionError(f"pagination did not terminate; visited {len(rows)} rows") def test_history_pagination_visits_every_tied_entry_once(tmp_path: Path) -> None: diff --git a/tests/wf_api/test_resume_attempt.py b/tests/wf_api/test_resume_attempt.py index 9cd43c5a..36007231 100644 --- a/tests/wf_api/test_resume_attempt.py +++ b/tests/wf_api/test_resume_attempt.py @@ -40,7 +40,9 @@ def _api(root: Path) -> tuple[WorkflowRunApi, FileRunStore]: return WorkflowRunApi(context), context.run_store -def test_resume_marks_active_attempt_with_store_backed_id(tmp_path: Path) -> None: +def test_resume_attempt_ids_are_monotonic_across_store_instances( + tmp_path: Path, +) -> None: api, store = _api(tmp_path / "resume-marker") started = asyncio.run( api.run_deployment(deployment_id="echo.personal", workflow_input={"text": "hi"}) @@ -55,7 +57,7 @@ def test_resume_marks_active_attempt_with_store_backed_id(tmp_path: Path) -> Non assert second == first + 1 -def test_active_attempt_blocks_second_resume(tmp_path: Path) -> None: +def test_active_attempt_marker_is_durable(tmp_path: Path) -> None: from datetime import UTC, datetime from wf_artifacts.runs.models import ResumeAttempt diff --git a/tests/wf_server/test_cli.py b/tests/wf_server/test_cli.py index 531bd2f4..34f47034 100644 --- a/tests/wf_server/test_cli.py +++ b/tests/wf_server/test_cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import pytest from typer.testing import CliRunner from wf_server.cli import app @@ -583,6 +584,10 @@ def test_rpc_server_cli_enable_scheduler_rejects_mcp_backed_server( "wf_server.cli.build_workflow_server_from_legacy_mcp_config", fake_build_mcp_server, ) + monkeypatch.setattr( + "wf_server.cli.uvicorn.run", + lambda *args, **kwargs: pytest.fail("server must not start"), + ) result = CliRunner().invoke( app, @@ -638,6 +643,10 @@ def test_rpc_server_cli_config_mcp_sources_reject_scheduler( fake_build_server, ) monkeypatch.setattr("wf_server.cli.create_rpc_app", fake_create_rpc_app) + monkeypatch.setattr( + "wf_server.cli.uvicorn.run", + lambda *args, **kwargs: pytest.fail("server must not start"), + ) result = CliRunner().invoke(app, ["--config", str(config_path)]) diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py index 86be7620..6a7f1ae5 100644 --- a/tests/wf_transport_rpc_http/test_schedules_rpc.py +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -90,9 +90,8 @@ def _constant_plan() -> RawWorkflowPlan: ) -async def _seed_server(tmp_path: Any, name: str = "sched-art") -> Any: +async def _seed_server(tmp_path: Any) -> Any: """Build a schedule-enabled server with one artifact + deployment.""" - _ = name server = build_local_static_workflow_server(tmp_path / "store", schedules=True) await server.api.create_artifact_from_plan( artifact_id="sched-art",